CLI

pyproject.toml

@@ -17,11 +17,15 @@ include = [
  { path = "tests", format = "sdist" }
 ]
 
+[tool.poetry.scripts]
+aaq-sync = "aaq_sync.cli:aaq_sync"
+
 [tool.poetry.dependencies]
 python = "^3.11"
 attrs = "^23.1.0"
-sqlalchemy = "^2.0.16"
+click = "^8.1.4"
 httpx = "^0.24.1"
+sqlalchemy = "^2.0.16"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.3.0"
@@ -31,22 +35,23 @@ pytest-cov = "^4.0.0"
 pytest-httpx = "^0.22.0"
 pytest-postgresql = "^5.0.0"
 ruff = "^0.0.261"
+types-click = "^7.1.8"
 
 [tool.poetry.group.lsp]
 optional = true
 
 [tool.poetry.group.lsp.dependencies]
-python-lsp-server = "^1.7.2"
-python-lsp-ruff = "^1.4.0"
-python-lsp-black = "^1.2.1"
 pylsp-mypy = "^0.6.6"
+python-lsp-black = "^1.2.1"
+python-lsp-ruff = "^1.4.0"
+python-lsp-server = "^1.7.2"
 
 [tool.black]
 preview = true
 
 [tool.mypy]
-ignore_missing_imports = false
 check_untyped_defs = true
+ignore_missing_imports = false
 
 [tool.pytest.ini_options]
 addopts = "--cov=src --cov-report=term-missing"

src/aaq_sync/cli.py

@@ -0,0 +1,76 @@
+import click
+from httpx import URL as HttpURL
+from sqlalchemy import URL as DbURL
+from sqlalchemy import create_engine
+from sqlalchemy.engine import make_url as make_db_url
+from sqlalchemy.orm import Session
+
+from .data_export_client import ExportClient
+from .data_models import Base, get_models
+from .sync import sync_model_items
+
+MODEL_MAPPING = {m.__tablename__: m for m in get_models()}
+
+
+class OptMixin:
+ @classmethod
+ def option(cls, *param_decls, **kw):
+ kw.setdefault("required", True)
+ kw.setdefault("type", cls())
+ return click.option(*param_decls, **kw)
+
+
+class DbURLParam(click.ParamType, OptMixin):
+ name = "db_url"
+
+ def convert(self, value, param, ctx):
+ url = make_db_url(value)
+ if url.drivername == "postgresql":
+ url = url.set(drivername="postgresql+psycopg")
+ return url
+
+
+class HttpURLParam(click.ParamType, OptMixin):
+ name = "http_url"
+
+ def convert(self, value, param, ctx):
+ return HttpURL(value)
+
+
+class TableChoiceParam(click.Choice, OptMixin):
+ name = "table"
+ envvar_list_splitter = ","
+
+ def __init__(self):
+ super().__init__(choices=sorted(MODEL_MAPPING.keys()))
+
+ def convert(self, value, param, ctx):
+ return MODEL_MAPPING[super().convert(value, param, ctx)]
+
+
+@click.command(context_settings={"auto_envvar_prefix": "AAQ_SYNC"})
+@DbURLParam.option("--db-url", help="Database URL.")
+@HttpURLParam.option("--export-url", help="Data export API URL.")
+@click.option("--export-token", type=str, required=True, help="Export API auth token.")
+@TableChoiceParam.option(
+ "tables", "--table", multiple=True, help="Table to sync. (Multiple allowed.)"
+)
+def aaq_sync(
+ db_url: DbURL,
+ export_url: HttpURL,
+ export_token: str,
+ tables: list[type[Base]],
+):
+ """
+ Sync one or more AAQ tables from the given data export API endpoint to the
+ given database.
+ """
+ dbengine = create_engine(db_url, echo=False)
+ with (
+ Session(dbengine) as session,
+ ExportClient(export_url, export_token) as exporter,
+ ):
+ for table in tables:
+ click.echo(f"Syncing {table.__tablename__} ...")
+ synced = sync_model_items(table, exporter, session)
+ click.echo(f"Synced {len(synced)} {table.__tablename__} items.")

src/aaq_sync/data_models.py

@@ -1,3 +1,4 @@
+from collections.abc import Collection
 from datetime import datetime
 from typing import Any, Self, TypeVar
 
@@ -64,6 +65,10 @@ def pkey_value(self) -> tuple:
  return self.__mapper__.primary_key_from_instance(self)
 
 
+def get_models() -> Collection[type[Base]]:
+ return {m.class_ for m in Base.registry.mappers}
+
+
 # dataclass options (such as kw_only) aren't inherited from parent classes.
 class FAQModel(Base, kw_only=True):
  """

src/aaq_sync/sync.py

@@ -1,4 +1,4 @@
-from collections.abc import Generator, Iterable
+from collections.abc import Generator, Iterable, Sequence
 from dataclasses import asdict
 from typing import TypeVar
 
@@ -44,27 +44,33 @@ def fetch_existing(model: type[TBase], session: Session) -> Iterable[TBase]:
  return session.scalars(select(model).order_by(*model.__table__.primary_key))
 
 
-def store_new(news: Iterable[TBase], session: Session) -> Iterable[TBase]:
+def store_new(news: Iterable[TBase], session: Session) -> Sequence[TBase]:
  """
  Store new items in the database.
+
+ NOTE: This is intended to be a low-level operation and thus doesn't commit
+ the transaction.
  """
  news = IteratorWithFinishedCheck(news)
  if news.finished:
  return []
  olds = fetch_existing(type(news.peek_next()), session)
  stored: list[TBase] = []
- with session.begin(nested=True):
- for new in filter_existing(olds, news):
- session.add(new)
- stored.append(new)
+ for new in filter_existing(olds, news):
+ session.add(new)
+ stored.append(new)
  return stored
 
 
 def sync_model_items(
  model: type[TBase], exporter: ExportClient, session: Session
-) -> Iterable[TBase]:
+) -> Sequence[TBase]:
  """
  Fetch model items from the data export API and store the new ones in the database.
+
+ NOTE: This is intended to be a high-level operation and thus commits the
+ transaction.
  """
  model_items = exporter.get_model_items(model)
- return store_new(model_items, session)
+ with session.begin():
+ return store_new(model_items, session)

tests/helpers.py

@@ -0,0 +1,43 @@
+import json
+from collections.abc import Generator
+from contextlib import contextmanager
+from dataclasses import asdict
+from importlib import resources
+
+from attrs import define
+from sqlalchemy import Engine, select
+from sqlalchemy.orm import Session
+
+from aaq_sync.data_models import FAQModel
+
+
+def read_test_data(path: str) -> str:
+ return (resources.files(__package__) / "test_data" / path).read_text()
+
+
+@define
+class Database:
+ engine: Engine
+
+ @property
+ def url_with_password(self):
+ return self.engine.url.render_as_string(hide_password=False)
+
+ @contextmanager
+ def session(self, **kw) -> Generator[Session, None, None]:
+ kw.setdefault("expire_on_commit", False)
+ with Session(self.engine, **kw) as session:
+ yield session
+
+ def fetch_faqs(self) -> list[FAQModel]:
+ query = select(FAQModel).order_by(FAQModel.faq_id)
+ with self.session() as session:
+ return list(session.scalars(query))
+
+ def faq_json_to_db(self, path: str) -> list[dict]:
+ faq_dicts = json.loads(read_test_data(path))["result"]
+ faqs = [FAQModel.from_json(faqd) for faqd in faq_dicts]
+ with self.session() as session:
+ session.add_all(faqs)
+ session.commit()
+ return [asdict(m) for m in sorted(faqs, key=lambda m: m.pkey_value())]

tests/test_cli.py

@@ -0,0 +1,103 @@
+import json
+
+import pytest
+from click.testing import CliRunner
+from httpx import URL
+
+from aaq_sync.cli import aaq_sync
+from aaq_sync.data_models import Base, FAQModel
+
+from .fake_data_export import FakeDataExport
+from .helpers import Database, read_test_data
+
+OPTS_DB = ("--db-url", "postgresql://db.example.com:5432/testdb")
+OPTS_EXPORT_URL = ("--export-url", "http://export.example.com")
+OPTS_EXPORT_TOKEN = ("--export-token", "faketoken")
+OPTS_EXPORT = (*OPTS_EXPORT_URL, *OPTS_EXPORT_TOKEN)
+OPTS_TABLE = ("--table", "faqmatches")
+
+
+@pytest.fixture()
+def runner():
+ """
+ Fixture providing a CliRunner with an isolated filesystem.
+ """
+ runner = CliRunner()
+ with runner.isolated_filesystem():
+ yield runner
+
+
+@pytest.fixture()
+def db(dbengine):
+ Base.metadata.create_all(dbengine)
+ return Database(dbengine)
+
+
+@pytest.fixture()
+def fake_data_export(httpx_mock):
+ return FakeDataExport(URL("https://127.0.0.100:1234/"), httpx_mock)
+
+
+def test_missing_opts(runner):
+ """
+ All required options must be provided.
+ """
+ result = runner.invoke(aaq_sync, [*OPTS_EXPORT, *OPTS_TABLE])
+ assert "Missing option '--db-url'" in result.output
+ assert result.exit_code != 0
+
+ result = runner.invoke(aaq_sync, [*OPTS_DB, *OPTS_EXPORT_TOKEN, *OPTS_TABLE])
+ assert "Missing option '--export-url'" in result.output
+ assert result.exit_code != 0
+
+ result = runner.invoke(aaq_sync, [*OPTS_DB, *OPTS_EXPORT_URL, *OPTS_TABLE])
+ assert "Missing option '--export-token'" in result.output
+ assert result.exit_code != 0
+
+ result = runner.invoke(aaq_sync, [*OPTS_DB, *OPTS_EXPORT])
+ assert "Missing option '--table'" in result.output
+ assert result.exit_code != 0
+
+
+def test_sync_faqmatches(runner, fake_data_export, db):
+ """
+ All FAQ items are synced from the data export API to the db.
+ """
+ faqds = json.loads(read_test_data("two_faqs.json"))["result"]
+ [faq1, faq2] = [FAQModel.from_json(faqd) for faqd in faqds]
+ fake_data_export.faqmatches.extend(faqds)
+
+ assert db.fetch_faqs() == []
+
+ opts = [
+ *("--db-url", db.engine.url),
+ *("--export-url", fake_data_export.base_url),
+ *("--export-token", "faketoken"),
+ *("--table", "faqmatches"),
+ ]
+
+ result = runner.invoke(aaq_sync, opts)
+ print(result.output)
+ assert result.exit_code == 0
+ assert db.fetch_faqs() == [faq1, faq2]
+
+
+def test_sync_faqmatches_envvars(runner, fake_data_export, db, monkeypatch):
+ """
+ All config options can be provided through envvars.
+ """
+ faqds = json.loads(read_test_data("two_faqs.json"))["result"]
+ [faq1, faq2] = [FAQModel.from_json(faqd) for faqd in faqds]
+ fake_data_export.faqmatches.extend(faqds)
+
+ assert db.fetch_faqs() == []
+
+ monkeypatch.setenv("AAQ_SYNC_DB_URL", db.url_with_password)
+ monkeypatch.setenv("AAQ_SYNC_EXPORT_URL", str(fake_data_export.base_url))
+ monkeypatch.setenv("AAQ_SYNC_EXPORT_TOKEN", "faketoken")
+ monkeypatch.setenv("AAQ_SYNC_TABLES", "faqmatches")
+
+ result = runner.invoke(aaq_sync, [])
+ print(result.output)
+ assert result.exit_code == 0
+ assert db.fetch_faqs() == [faq1, faq2]