Commit 1c8d0bdd0a
Verified · cmc
Layout: unified · split
src/srht_contrib/config.py +1
| @@ -37,6 +37,7 @@ class Settings(BaseSettings): | |||
| 37 | poll_interval_seconds: int = Field(default=900, alias="POLL_INTERVAL_SECONDS") | 37 | poll_interval_seconds: int = Field(default=900, alias="POLL_INTERVAL_SECONDS") |
| 38 | sync_overlap_hours: int = Field(default=1, alias="SYNC_OVERLAP_HOURS") | 38 | sync_overlap_hours: int = Field(default=1, alias="SYNC_OVERLAP_HOURS") |
| 39 | srht_request_delay_seconds: float = Field(default=0.5, alias="SRHT_REQUEST_DELAY_SECONDS") | 39 | srht_request_delay_seconds: float = Field(default=0.5, alias="SRHT_REQUEST_DELAY_SECONDS") |
| 40 | sqlite_busy_timeout_seconds: float = Field(default=30.0, alias="SQLITE_BUSY_TIMEOUT_SECONDS") | ||
| 40 | discovery_batch_size: int = Field(default=5, alias="DISCOVERY_BATCH_SIZE") | 41 | discovery_batch_size: int = Field(default=5, alias="DISCOVERY_BATCH_SIZE") |
| 41 | indexed_actor_repoll_seconds: int = Field(default=21600, alias="INDEXED_ACTOR_REPOLL_SECONDS") | 42 | indexed_actor_repoll_seconds: int = Field(default=21600, alias="INDEXED_ACTOR_REPOLL_SECONDS") |
| 42 | discovery_error_backoff_seconds: int = Field(default=3600, alias="DISCOVERY_ERROR_BACKOFF_SECONDS") | 43 | discovery_error_backoff_seconds: int = Field(default=3600, alias="DISCOVERY_ERROR_BACKOFF_SECONDS") |
src/srht_contrib/db.py +20 −3
| @@ -3,7 +3,7 @@ from __future__ import annotations | |||
| 3 | from collections.abc import Generator | 3 | from collections.abc import Generator |
| 4 | 4 | ||
| 5 | from fastapi import HTTPException, Request, status | 5 | from fastapi import HTTPException, Request, status |
| 6 | from sqlalchemy import Engine, create_engine, text | 6 | from sqlalchemy import Engine, create_engine, event, text |
| 7 | from sqlalchemy.pool import StaticPool | 7 | from sqlalchemy.pool import StaticPool |
| 8 | from sqlalchemy.orm import Session, declarative_base, sessionmaker | 8 | from sqlalchemy.orm import Session, declarative_base, sessionmaker |
| 9 | 9 | ||
| @@ -13,11 +13,28 @@ Base = declarative_base() | |||
| 13 | 13 | ||
| 14 | 14 | ||
| 15 | def make_engine(settings: Settings) -> Engine: | 15 | def make_engine(settings: Settings) -> Engine: |
| 16 | connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {} | 16 | connect_args = {} |
| 17 | if settings.database_url.startswith("sqlite"): | ||
| 18 | connect_args = { | ||
| 19 | "check_same_thread": False, | ||
| 20 | "timeout": settings.sqlite_busy_timeout_seconds, | ||
| 21 | } | ||
| 17 | engine_kwargs = {"future": True, "connect_args": connect_args} | 22 | engine_kwargs = {"future": True, "connect_args": connect_args} |
| 18 | if settings.database_url in {"sqlite://", "sqlite:///:memory:"}: | 23 | if settings.database_url in {"sqlite://", "sqlite:///:memory:"}: |
| 19 | engine_kwargs["poolclass"] = StaticPool | 24 | engine_kwargs["poolclass"] = StaticPool |
| 20 | return create_engine(settings.database_url, **engine_kwargs) | 25 | engine = create_engine(settings.database_url, **engine_kwargs) |
| 26 | |||
| 27 | if settings.database_url.startswith("sqlite"): | ||
| 28 | @event.listens_for(engine, "connect") | ||
| 29 | def _configure_sqlite(dbapi_connection, connection_record) -> None: # type: ignore[unused-ignore] | ||
| 30 | cursor = dbapi_connection.cursor() | ||
| 31 | cursor.execute(f"PRAGMA busy_timeout = {int(settings.sqlite_busy_timeout_seconds * 1000)}") | ||
| 32 | if settings.database_url not in {"sqlite://", "sqlite:///:memory:"}: | ||
| 33 | cursor.execute("PRAGMA journal_mode = WAL") | ||
| 34 | cursor.execute("PRAGMA synchronous = NORMAL") | ||
| 35 | cursor.close() | ||
| 36 | |||
| 37 | return engine | ||
| 21 | 38 | ||
| 22 | 39 | ||
| 23 | def make_session_factory(settings: Settings) -> sessionmaker[Session]: | 40 | def make_session_factory(settings: Settings) -> sessionmaker[Session]: |
src/srht_contrib/scripts/enqueue_actors.py +18 −9
| @@ -34,6 +34,9 @@ def enqueue_actors(username_file: Path, *, stagger_seconds: int = 300, start_at: | |||
| 34 | queued_at = start_at or datetime.now(tz=UTC) | 34 | queued_at = start_at or datetime.now(tz=UTC) |
| 35 | inserted = 0 | 35 | inserted = 0 |
| 36 | 36 | ||
| 37 | batch_size = 250 | ||
| 38 | queued_in_batch = 0 | ||
| 39 | |||
| 37 | with session_factory() as db: | 40 | with session_factory() as db: |
| 38 | for index, actor in enumerate(usernames): | 41 | for index, actor in enumerate(usernames): |
| 39 | next_poll_after = queued_at + timedelta(seconds=index * stagger_seconds) | 42 | next_poll_after = queued_at + timedelta(seconds=index * stagger_seconds) |
| @@ -49,15 +52,21 @@ def enqueue_actors(username_file: Path, *, stagger_seconds: int = 300, start_at: | |||
| 49 | ) | 52 | ) |
| 50 | db.add(tracked_actor) | 53 | db.add(tracked_actor) |
| 51 | inserted += 1 | 54 | inserted += 1 |
| 52 | continue | 55 | else: |
| 53 | 56 | tracked_actor.is_active = True | |
| 54 | tracked_actor.is_active = True | 57 | if tracked_actor.queued_for_discovery_at is None: |
| 55 | if tracked_actor.queued_for_discovery_at is None: | 58 | tracked_actor.queued_for_discovery_at = queued_at |
| 56 | tracked_actor.queued_for_discovery_at = queued_at | 59 | if tracked_actor.last_polled_at is None and tracked_actor.discovery_state != "indexed": |
| 57 | if tracked_actor.last_polled_at is None and tracked_actor.discovery_state != "indexed": | 60 | tracked_actor.discovery_state = "queued" |
| 58 | tracked_actor.discovery_state = "queued" | 61 | tracked_actor.next_poll_after = next_poll_after |
| 59 | tracked_actor.next_poll_after = next_poll_after | 62 | |
| 60 | db.commit() | 63 | queued_in_batch += 1 |
| 64 | if queued_in_batch >= batch_size: | ||
| 65 | db.commit() | ||
| 66 | queued_in_batch = 0 | ||
| 67 | |||
| 68 | if queued_in_batch: | ||
| 69 | db.commit() | ||
| 61 | 70 | ||
| 62 | return inserted | 71 | return inserted |
| 63 | 72 | ||