Route both upsert and delete branches off a single ranked CTE (rn=1 per RecordID over all pending ops ordered by SourceLogID DESC). The previous design used two independent ranked CTEs, which let a stale Delete outrank a newer Insert for the same RecordID and silently drop the row. Also gitignore .claude/ and .workbuddy/ runtime dirs.
Wires capture -> apply -> cleanup into cycle(cfg): per-file capture and
cleanup each wrapped in try/except + log.exception so one file's failure
does not abort the cycle; apply failure does not block cleanup; writer is
always closed in finally. run(cfg) loops cycle with sleep; main() loads
config from argv. logging_setup uses RotatingFileHandler 10MBx5 + console.
Unit tests cover all three error-isolation branches via mocks (no real
end-to-end smoke; integration deferred to Task 9 pilot).
Co-Authored-By: Claude <noreply@anthropic.com>
Adds SqlWriter: a pyodbc-backed writer that dedup-inserts into
dbo.SyncQueue (IF NOT EXISTS guarded by UX_SyncQueue_Dedup), invokes
dbo.usp_SyncApply, and reports applied SourceLogIDs.
Connection is opened with autocommit=True per the controller revision:
usp_SyncApply manages its own transaction internally (BEGIN/ROLLBACK),
and an outer pyodbc transaction would conflict on ROLLBACK (SQL error
266). The dedup IF NOT EXISTS...INSERT is a single atomic statement.
Integration test self-cleans via SourceFile='sqlw_test.accdb' marker;
conn_str comes from the gitignored config.yaml (no hardcoded creds).
Co-Authored-By: Claude <noreply@anthropic.com>
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 <noreply@anthropic.com>
- sql/01_sync_queue.sql: idempotent DDL for dbo.SyncQueue (PK + unique
dedup index + pending lookup index), safe to re-run.
- sql/02_sync_apply.sql: dbo.usp_SyncApply (@MaxRetries INT=5). Per
distinct (TargetSchema,TargetTable) it builds column projections from
sys.columns (excludes ID key/computed/identity/rowversion) and runs a
dynamic-SQL MERGE (last-write-wins via ROW_NUMBER over SourceLogID DESC)
for Insert/Update plus a DELETE for the last op = Delete.
SET IDENTITY_INSERT ON preserves Access PKs.
- tests/conftest.py: sql_conn fixture reads conn_str from gitignored
config.yaml via load_config; skipped without RUN_INTEGRATION=1.
- tests/test_apply_proc.py: integration test covering IDENTITY-preserving
INSERT, last-write-wins UPDATE, BIT conversion, and DELETE; cleans up.
Deviation from the brief's procedure (root-cause fix, design preserved):
every JSON path key is quoted ('$."col"') so non-ASCII column names
(e.g. Chinese 名字/数量) parse correctly. Without quoting, JSON_VALUE
raises "JSON path format is not correct" on Chinese columns, which is the
real target schema for this Access->SQL Server sync.
Co-Authored-By: Claude <noreply@anthropic.com>
Add Pydantic config models (SqlServerConfig, AccessConfig, RuntimeConfig,
FileMapping, SyncConfig) and a YAML loader (load_config). FileMapping provides
source_path() and target_table() helpers; number-typed YAML keys/values (e.g.
root: 2026) are coerced to str via coerce_numbers_to_str. Includes
config.example.yaml template (config.yaml with real credentials stays
gitignored) and pyproject.toml pytest config (pythonpath=src).
Co-Authored-By: Claude <noreply@anthropic.com>
