From 1f7191421a0c4d48db43323326faf264faf3d32e Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Tue, 14 Jul 2026 12:22:48 +0800 Subject: [PATCH] feat: access reader and value serialization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add to_jsonable() (Access/pyodbc value -> JSON-safe) with strict TDD (5 tests: datetime iso, bool preserved before int, decimal->str, none/numbers, dict round-trip). bool branch precedes int because bool is a subclass of int in Python. Add AccessReader: reads TableChangeLog (read_log), reads a full source row by RecordID via to_jsonable (read_row, warns on >4000-char values that JSON_VALUE would truncate), and deletes processed log IDs in chunked, retried batches (delete_log_ids, no-op on empty list). Connects via the ACE ODBC driver from config.access.driver, shared autocommit mode to coexist with the live .laccdb client. Integration test against the real 氩弧焊.accdb (read-only; delete_log_ids only exercised with []) passes. Co-Authored-By: Claude --- src/sync/access_reader.py | 119 ++++++++++++++++++++++++++++++++++++ src/sync/serialize.py | 34 +++++++++++ tests/test_access_reader.py | 37 +++++++++++ tests/test_serialize.py | 21 +++++++ 4 files changed, 211 insertions(+) create mode 100644 src/sync/access_reader.py create mode 100644 src/sync/serialize.py create mode 100644 tests/test_access_reader.py create mode 100644 tests/test_serialize.py diff --git a/src/sync/access_reader.py b/src/sync/access_reader.py new file mode 100644 index 0000000..8dbe782 --- /dev/null +++ b/src/sync/access_reader.py @@ -0,0 +1,119 @@ +"""Read Access change-log rows and full source rows via the ACE ODBC driver. + +Opened in shared/autocommit mode so it coexists with the live client that holds +the ``.laccdb`` lock on the production ``.accdb`` files. The driver name comes +from ``config.access.driver`` (no secret in the connection string — Windows +auth / no password on the Access share). +""" +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass + +import pyodbc + +from .serialize import to_jsonable + +log = logging.getLogger(__name__) + + +@dataclass +class LogRow: + """A single row from the Access ``TableChangeLog`` table.""" + id: int + table_name: str + record_id: str + operate_type: str # "Insert" | "Update" | "Delete" + time: object # raw Access datetime (serialized later by caller) + + +class AccessReader: + """Thin pyodbc wrapper around one ``.accdb`` file. + + The connection is lazily created and reused across calls. ``close()`` must + be called when the reader is no longer needed. + """ + + def __init__(self, db_path: str, driver: str): + self.db_path = db_path + self.driver = driver + self._conn = None + + def _connect(self): + if self._conn is None: + # ReadOnly=0: shared read/write open. The capture path is read-only, + # but the cleanup path (delete_log_ids) genuinely deletes processed + # log rows. autocommit=True avoids a held transaction that would + # block the live client. + conn_str = f"Driver={self.driver};DBQ={self.db_path};ReadOnly=0;" + self._conn = pyodbc.connect(conn_str, autocommit=True) + return self._conn + + def read_log(self, batch_size: int) -> list[LogRow]: + """Return up to ``batch_size`` oldest log rows ordered by ID.""" + cur = self._connect().cursor() + cur.execute( + f"SELECT TOP {int(batch_size)} ID, TableName, RecordID, " + f"OperateType, Time FROM TableChangeLog ORDER BY ID" + ) + return [ + LogRow(r[0], r[1], str(r[2]), (r[3] or "").strip(), r[4]) + for r in cur.fetchall() + ] + + def read_row(self, table: str, record_id: str) -> dict | None: + """Return the full source row for ``record_id`` as a JSONable dict. + + Returns ``None`` if the row no longer exists. Emits a warning when any + string value exceeds 4000 chars (the downstream ``JSON_VALUE`` on SQL + Server ``NVARCHAR(MAX)`` -> ``NVARCHAR(4000)`` would silently truncate). + """ + cur = self._connect().cursor() + cur.execute(f'SELECT * FROM "{table}" WHERE ID = ?', record_id) + cols = [c[0] for c in cur.description] + row = cur.fetchone() + if row is None: + return None + d = {cols[i]: to_jsonable(row[i]) for i in range(len(cols))} + for k, v in d.items(): + if isinstance(v, str) and len(v) > 4000: + log.warning( + "value >4000 chars in %s.ID=%s col=%s " + "(JSON_VALUE will truncate)", + table, record_id, k, + ) + return d + + def delete_log_ids( + self, ids: list[int], batch_size: int, retries: int + ) -> None: + """Delete the given log-row IDs in chunks, retrying on lock contention. + + A no-op when ``ids`` is empty (never raises). Retries with linear + backoff because the live client may briefly hold a page lock on + ``TableChangeLog``. + """ + if not ids: + return + cur = self._connect().cursor() + for i in range(0, len(ids), batch_size): + chunk = ids[i:i + batch_size] + placeholders = ",".join("?" * len(chunk)) + for attempt in range(retries): + try: + cur.execute( + f"DELETE FROM TableChangeLog WHERE ID IN ({placeholders})", + *chunk, + ) + break + except pyodbc.OperationalError: + if attempt < retries - 1: + time.sleep(0.2 * (attempt + 1)) + else: + raise + + def close(self): + if self._conn: + self._conn.close() + self._conn = None diff --git a/src/sync/serialize.py b/src/sync/serialize.py new file mode 100644 index 0000000..3f7c85c --- /dev/null +++ b/src/sync/serialize.py @@ -0,0 +1,34 @@ +"""Access value serialization helpers. + +Converts pyodbc/Access cell values into JSON-serializable Python types so a +captured row dict can be ``json.dumps``-ed before being enqueued on SQL Server. + +Order matters: ``bool`` MUST be checked before ``int`` because ``bool`` is a +subclass of ``int`` in Python (``isinstance(True, int) is True``). If the order +were reversed, ``True`` would be returned as ``1`` and lose its type. +""" +import datetime +import decimal + + +def to_jsonable(v): + """Coerce a single Access/pyodbc value to a JSON-serializable form. + + - ``None`` -> ``None`` + - ``bool`` -> ``bool`` (must precede the ``int`` branch) + - ``datetime``/``date`` -> ISO-8601 string (``"2026-07-14T08:41:18"``) + - ``Decimal`` -> string (preserves exact scale, e.g. ``"12.50"``) + - ``int``/``float``/``str`` -> unchanged + - anything else (e.g. ``bytes`` for OLE) -> ``str(v)`` fallback + """ + if v is None: + return None + if isinstance(v, bool): # MUST be before int + return v + if isinstance(v, (datetime.datetime, datetime.date)): + return v.isoformat() + if isinstance(v, decimal.Decimal): + return str(v) + if isinstance(v, (int, float, str)): + return v + return str(v) diff --git a/tests/test_access_reader.py b/tests/test_access_reader.py new file mode 100644 index 0000000..f69721e --- /dev/null +++ b/tests/test_access_reader.py @@ -0,0 +1,37 @@ +"""Integration test for AccessReader against the real 氩弧焊.accdb. + +READ-ONLY: this test never mutates production data. ``delete_log_ids`` is only +exercised with an empty list (a guaranteed no-op) to confirm it does not raise. +No real log rows are deleted. +""" +import os +import pytest + +from sync.access_reader import AccessReader +from sync.config import load_config + + +@pytest.mark.integration +def test_read_log_and_row_and_delete(): + if not os.environ.get("RUN_INTEGRATION"): + pytest.skip("needs RUN_INTEGRATION=1") + cfg = load_config("config.yaml") + # Real live database (has a .laccdb lock from the client). Shared-mode + # read via ACE is validated to coexist with the lock. + db = os.environ.get( + "TEST_ACCDB", + r"\\192.168.110.114\生产进度表\2026年数据\氩弧焊.accdb", + ) + r = AccessReader(db, cfg.access.driver) + try: + rows = r.read_log(500) + assert isinstance(rows, list) + if rows: + lr = rows[0] + assert lr.table_name and lr.record_id + assert lr.operate_type in ("Insert", "Update", "Delete") + d = r.read_row(lr.table_name, lr.record_id) + assert d is None or "ID" in d + r.delete_log_ids([], 100, 3) # empty list -> no-op, must not raise + finally: + r.close() diff --git a/tests/test_serialize.py b/tests/test_serialize.py new file mode 100644 index 0000000..f4d2e95 --- /dev/null +++ b/tests/test_serialize.py @@ -0,0 +1,21 @@ +import datetime, decimal +from sync.serialize import to_jsonable +import json + +def test_datetime_iso(): + assert to_jsonable(datetime.datetime(2026,7,14,8,41,18)) == "2026-07-14T08:41:18" + +def test_bool_preserved(): + assert to_jsonable(True) is True and to_jsonable(False) is False + +def test_decimal_to_str(): + assert to_jsonable(decimal.Decimal("12.50")) == "12.50" + +def test_none_and_numbers(): + assert to_jsonable(None) is None + assert to_jsonable(5) == 5 + assert to_jsonable("x") == "x" + +def test_dict_serializes(): + d = {"d": datetime.date(2026,1,1), "b": True, "n": None} + assert json.loads(json.dumps({k: to_jsonable(v) for k,v in d.items()})) == {"d":"2026-01-01","b":True,"n":None}