9c3cc38a44
ACE ODBC reports lock contention ("无法更新；当前被锁定。(-1102)")
as a generic pyodbc.Error (HY000), NOT pyodbc.OperationalError, so the
old except clause never caught it and gave up after the first collision
— making cleanup_lock_retries:3 a no-op for the 氩弧焊/高写入库 -1102 case.
Now catch pyodbc.Error and retry only on lock messages ("被锁定"/"-1102"),
re-raising other errors immediately instead of retrying them 3×.
170 lines
6.6 KiB
Python
170 lines
6.6 KiB
Python
"""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
|
||
) -> int:
|
||
"""Delete the given log-row IDs in chunks, retrying on lock contention.
|
||
|
||
Returns the total number of rows actually deleted (sum of per-chunk
|
||
``cursor.rowcount``), so the caller can report honest counts. A no-op
|
||
(returns 0) when ``ids`` is empty. Retries with linear backoff because
|
||
the live client may briefly hold a page lock on ``TableChangeLog``.
|
||
Raises on final lock failure.
|
||
"""
|
||
if not ids:
|
||
return 0
|
||
total = 0
|
||
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,
|
||
)
|
||
total += cur.rowcount
|
||
break
|
||
except pyodbc.Error as e:
|
||
# ACE ODBC reports lock contention ("无法更新；当前被锁定。
|
||
# (-1102)") as a generic ``pyodbc.Error`` (HY000), NOT as
|
||
# ``pyodbc.OperationalError`` — so the old handler silently
|
||
# missed it and gave up after the first collision. Retry only
|
||
# on lock messages; re-raise anything else so genuine errors
|
||
# surface immediately instead of being retried 3×.
|
||
msg = str(e)
|
||
is_lock = "被锁定" in msg or "-1102" in msg
|
||
if is_lock and attempt < retries - 1:
|
||
time.sleep(0.2 * (attempt + 1))
|
||
else:
|
||
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()
|
||
self._conn = None
|