diff --git a/src/sync/access_reader.py b/src/sync/access_reader.py index d880002..5e3b294 100644 --- a/src/sync/access_reader.py +++ b/src/sync/access_reader.py @@ -118,6 +118,43 @@ class AccessReader: raise return total + def list_user_tables(self) -> list[str]: + """Return user table names in this Access DB. + + Excludes system tables (``MSys*``) and temporary ``~*`` tables. The + change-log table ``TableChangeLog`` is NOT auto-excluded here — callers + apply their own ``FileMapping.exclude_tables`` / ``include_tables`` rules + on top of this list (see ``sync.fullsync.resolve_tables``). + """ + cur = self._connect().cursor() + names = [] + for r in cur.tables(tableType="TABLE"): + n = r.table_name + if n.startswith("MSys") or n.startswith("~"): + continue + names.append(n) + return names + + def read_all_rows(self, table: str): + """Return ``(columns, rows)`` for every row in ``table``. + + ``columns`` is the ordered list of column names; ``rows`` is a list of + tuples of **raw** pyodbc cell values (native Python types preserved) so + the downstream bulk insert keeps correct SQL Server types. An empty + table yields ``([...], [])`` with the real column list. + """ + cur = self._connect().cursor() + cur.execute(f'SELECT * FROM "{table}"') + cols = [c[0] for c in cur.description] + rows = [tuple(row) for row in cur.fetchall()] + return cols, rows + + def read_all_log_ids(self) -> list[int]: + """Return every ``ID`` from this file's ``TableChangeLog``, ascending.""" + cur = self._connect().cursor() + cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID") + return [r[0] for r in cur.fetchall()] + def close(self): if self._conn: self._conn.close() diff --git a/src/sync/fullsync.py b/src/sync/fullsync.py new file mode 100644 index 0000000..30571b9 --- /dev/null +++ b/src/sync/fullsync.py @@ -0,0 +1,159 @@ +"""One-shot full sync: TRUNCATE each target table, then bulk-insert every row +from the corresponding Access table. + +This is a manual, out-of-band operation — ``python -m sync.fullsync +config.yaml [--db FILE] [--table NAME] [--clear-change-log]``. It deliberately +bypasses the incremental ``SyncQueue`` pipeline and writes straight to the SQL +Server mirror tables. Use it to (re)seed a schema from scratch or to repair +drift between Access and SQL Server. + +Contract with the target tables (same as ``usp_SyncApply``): they already +exist and mirror the Access schema column-for-column, with ``ID`` as an +IDENTITY primary key. Because ``ID`` is an identity, ``SET IDENTITY_INSERT`` is +enabled during insert so the original Access primary keys survive — that is +what keeps the incremental sync's RecordID matching correct afterwards. +""" +from __future__ import annotations + +import argparse +import logging + +from .config import load_config, FileMapping, SyncConfig +from .access_reader import AccessReader +from .sql_writer import SqlWriter +from .logging_setup import setup_logging + +log = logging.getLogger("sync.fullsync") + + +def resolve_tables(reader: AccessReader, fm: FileMapping) -> list[str]: + """Tables to fully sync for one file, after exclude/include rules. + + Mirrors the precedence used by ``capture.capture_file``: a table in + ``exclude_tables`` is dropped even if it also appears in ``include_tables``. + System tables (``MSys*`` / ``~*``) are already filtered by + ``AccessReader.list_user_tables``. + """ + exclude = set(fm.exclude_tables or []) + include = set(fm.include_tables) if fm.include_tables else None + out = [] + for t in reader.list_user_tables(): + if t in exclude: + continue + if include is not None and t not in include: + continue + out.append(t) + return out + + +def full_sync_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter) -> dict: + """Full-sync every eligible table in one Access file. Returns a summary.""" + summary = {"file": fm.file, "tables": 0, "rows": 0, "skipped": []} + for access_table in resolve_tables(reader, fm): + target = fm.target_table(access_table) + if not writer.table_exists(fm.schema, target): + log.warning( + "skip %s -> %s.%s (target table missing)", + access_table, fm.schema, target, + ) + summary["skipped"].append(access_table) + continue + try: + cols, rows = reader.read_all_rows(access_table) + writer.truncate_target(fm.schema, target) + n = writer.bulk_insert(fm.schema, target, cols, rows) + summary["tables"] += 1 + summary["rows"] += n + log.info( + "full-synced %s -> %s.%s : %d rows", + access_table, fm.schema, target, n, + ) + except Exception: + log.exception( + "full sync failed for %s -> %s.%s", + access_table, fm.schema, target, + ) + summary["skipped"].append(access_table) + return summary + + +def full_sync( + cfg: SyncConfig, + db_filter: str | None = None, + table_filter: str | None = None, + clear_change_log: bool = False, +) -> list[dict]: + """Run a full sync across all (optionally filtered) files. + + ``db_filter`` limits to a single Access file (matched on ``FileMapping.file``). + ``table_filter`` restricts every considered file to that one table by + overriding ``include_tables``. ``clear_change_log`` additionally empties the + ``TableChangeLog`` on each synced file after loading (use with care — it + mutates the Access side so the incremental service won't replay old deltas). + """ + files = cfg.files + if db_filter: + files = [f for f in files if f.file == db_filter] + if not files: + log.warning("no file matches --db %r", db_filter) + return [] + if table_filter: + files = [f.model_copy(update={"include_tables": [table_filter]}) for f in files] + + writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table) + summaries = [] + try: + for fm in files: + reader = AccessReader(fm.source_path(cfg), cfg.access.driver) + try: + summary = full_sync_file(fm, reader, writer) + summaries.append(summary) + if clear_change_log: + try: + ids = reader.read_all_log_ids() + deleted = reader.delete_log_ids( + ids, 500, 3 + ) if ids else 0 + log.info("cleared %d TableChangeLog rows from %s", deleted, fm.file) + except Exception: + log.exception("clear-change-log failed for %s", fm.file) + finally: + reader.close() + finally: + writer.close() + return summaries + + +def main(): + """CLI entry point: ``python -m sync.fullsync config.yaml [options]``.""" + ap = argparse.ArgumentParser( + description="One-shot full sync: Access -> SQL Server (TRUNCATE + bulk INSERT)." + ) + ap.add_argument("config", nargs="?", default="config.yaml", + help="path to config.yaml (default: config.yaml)") + ap.add_argument("--db", help="limit to one Access file (by FileMapping.file)") + ap.add_argument("--table", help="limit to one table (applies to all matched files)") + ap.add_argument("--clear-change-log", action="store_true", + help="after loading, clear TableChangeLog on the synced files") + args = ap.parse_args() + + cfg = load_config(args.config) + setup_logging(cfg.logging) + + summaries = full_sync( + cfg, + db_filter=args.db, + table_filter=args.table, + clear_change_log=args.clear_change_log, + ) + + total_tables = sum(s["tables"] for s in summaries) + total_rows = sum(s["rows"] for s in summaries) + log.info("FULL SYNC COMPLETE: %d tables, %d rows", total_tables, total_rows) + for s in summaries: + if s["skipped"]: + log.warning(" %s: skipped %s", s["file"], s["skipped"]) + + +if __name__ == "__main__": + main() diff --git a/src/sync/sql_writer.py b/src/sync/sql_writer.py index 755864a..332612c 100644 --- a/src/sync/sql_writer.py +++ b/src/sync/sql_writer.py @@ -132,6 +132,93 @@ class SqlWriter: ) return cur.rowcount + def table_exists(self, schema: str, table: str) -> bool: + """True if ``[schema].[table]`` exists in the target database.""" + cur = self._conn.cursor() + cur.execute( + "SELECT 1 FROM sys.tables t JOIN sys.schemas s " + "ON s.schema_id = t.schema_id " + "WHERE s.name = ? AND t.name = ?", + schema, table, + ) + return cur.fetchone() is not None + + def _has_identity(self, schema: str, table: str) -> bool: + """True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK).""" + cur = self._conn.cursor() + cur.execute( + "SELECT 1 FROM sys.tables t JOIN sys.schemas s " + "ON s.schema_id = t.schema_id JOIN sys.columns c " + "ON c.object_id = t.object_id " + "WHERE s.name = ? AND t.name = ? AND c.is_identity = 1", + schema, table, + ) + return cur.fetchone() is not None + + def truncate_target(self, schema: str, table: str) -> None: + """Empty ``[schema].[table]`` prior to a full reload. + + Prefers ``TRUNCATE TABLE`` (fast, minimal logging). This project's + target schemas have no foreign keys, so TRUNCATE always succeeds; the + ``DELETE FROM`` fallback only matters if a future FK is added or the + table sits under snapshot isolation. + """ + full = f"[{schema}].[{table}]" + cur = self._conn.cursor() + try: + cur.execute(f"TRUNCATE TABLE {full}") + except pyodbc.Error: + cur.execute(f"DELETE FROM {full}") + + def bulk_insert( + self, schema: str, table: str, columns: list[str], rows: list[tuple] + ) -> int: + """Insert every ``row`` into ``[schema].[table]``. + + ``columns`` and the tuples in ``rows`` must be positionally aligned. + When the target has an IDENTITY column (the ``ID`` PK in every target + here), ``SET IDENTITY_INSERT`` is enabled so the original Access primary + keys are preserved — required for the incremental sync's RecordID + matching to keep working afterwards. + + Rows are inserted in 1000-row chunks. ``fast_executemany`` is tried + first for speed; if a row's types can't be inferred (e.g. a leading + NULL), it transparently retries the chunk without it. Returns the total + number of rows inserted. + """ + if not columns or not rows: + return 0 + full = f"[{schema}].[{table}]" + col_list = ", ".join(f"[{c}]" for c in columns) + placeholders = ", ".join("?" for _ in columns) + has_identity = self._has_identity(schema, table) + cur = self._conn.cursor() + if has_identity: + cur.execute(f"SET IDENTITY_INSERT {full} ON") + try: + inserted = 0 + fast = True + for i in range(0, len(rows), 1000): + chunk = rows[i:i + 1000] + for attempt in range(2): + try: + cur.fast_executemany = fast + cur.executemany( + f"INSERT INTO {full} ({col_list}) VALUES ({placeholders})", + chunk, + ) + break + except pyodbc.Error: + if attempt == 0 and fast: + fast = False # retry this chunk without fast_executemany + continue + raise + inserted += len(chunk) + return inserted + finally: + if has_identity: + cur.execute(f"SET IDENTITY_INSERT {full} OFF") + def close(self) -> None: """Close the underlying pyodbc connection.""" self._conn.close() diff --git a/tests/test_fullsync.py b/tests/test_fullsync.py new file mode 100644 index 0000000..e4de2b4 --- /dev/null +++ b/tests/test_fullsync.py @@ -0,0 +1,56 @@ +"""Unit tests for the full-sync table-resolution rules. + +``resolve_tables`` is pure (no DB) once given a reader stub, so it is safe to +exercise without Access/SQL connections. Run on a host with the project venv +(e.g. 114), where ``pyodbc`` imports cleanly: + + .venv\\Scripts\\python.exe -m pytest tests\\test_fullsync.py -q +""" +from sync.config import FileMapping +from sync.fullsync import resolve_tables + + +class FakeReader: + """Mirrors AccessReader.list_user_tables: system tables are already + filtered here, so resolve_tables only needs to apply the exclude/include + rules from the FileMapping.""" + + def __init__(self, tables): + self._tables = [ + t for t in tables + if not (t.startswith("MSys") or t.startswith("~")) + ] + + def list_user_tables(self): + return list(self._tables) + + +def make_fm(**kw): + base = dict(file="x.accdb", root="2026", schema="s", year_suffix="") + base.update(kw) + return FileMapping(**base) + + +def test_excludes_system_and_config_tables(): + fm = make_fm(exclude_tables=["TableChangeLog", "Foo"]) + r = FakeReader(["T1", "TableChangeLog", "MSysSysObjects", "~tmp", "Foo", "Bar"]) + assert resolve_tables(r, fm) == ["T1", "Bar"] + + +def test_include_tables_restricts_scope(): + fm = make_fm(include_tables=["T1", "T2"]) + r = FakeReader(["T1", "T2", "T3", "TableChangeLog"]) + assert resolve_tables(r, fm) == ["T1", "T2"] + + +def test_exclude_beats_include(): + # Precedence matches capture.capture_file: exclude wins over include. + fm = make_fm(include_tables=["T1"], exclude_tables=["T1"]) + r = FakeReader(["T1", "T2"]) + assert resolve_tables(r, fm) == [] + + +def test_empty_when_all_excluded(): + fm = make_fm(exclude_tables=["T1", "T2"]) + r = FakeReader(["T1", "T2"]) + assert resolve_tables(r, fm) == []