feat(sync): add one-shot full sync (TRUNCATE + bulk INSERT)
- New `sync.fullsync` CLI: `python -m sync.fullsync config.yaml [--db F] [--table T] [--clear-change-log]` - AccessReader: list_user_tables / read_all_rows / read_all_log_ids - SqlWriter: table_exists / truncate_target (TRUNCATE w/ DELETE fallback) / bulk_insert (SET IDENTITY_INSERT + chunked fast_executemany) - Reuses FileMapping exclude/include rules (exclude beats include, same as capture) - Preserves Access IDs via IDENTITY_INSERT; target schemas have no FKs so TRUNCATE is safe - TableChangeLog NOT cleared by default (opt-in --clear-change-log) - tests/test_fullsync.py covers resolve_tables exclude/include precedence
This commit is contained in:
Misaka_Company
@@ -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()
|
||||
|
||||
159
src/sync/fullsync.py
Normal file
159
src/sync/fullsync.py
Normal file
@@ -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()
|
||||
@@ -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()
|
||||
|
||||
56
tests/test_fullsync.py
Normal file
56
tests/test_fullsync.py
Normal file
@@ -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) == []
|
||||
Reference in New Issue
Block a user