3a3d3d0292
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>
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""Main service loop: capture -> apply -> cleanup, with per-file error isolation.
|
|
|
|
``cycle(cfg)`` runs one full pass over every configured file:
|
|
1. capture — read each file's change log and stage rows into SyncQueue;
|
|
2. apply — drain the queue via ``dbo.usp_SyncApply``;
|
|
3. cleanup — delete applied log rows from each file's Access log.
|
|
|
|
Each file's capture and cleanup is wrapped in its own try/except so one
|
|
file's failure is logged and the cycle continues; the writer is always closed
|
|
in a ``finally``. ``run(cfg)`` loops ``cycle`` with a sleep; ``main()``
|
|
loads the config from ``argv[1]`` (default ``config.yaml``).
|
|
"""
|
|
from __future__ import annotations
|
|
import sys
|
|
import time
|
|
import logging
|
|
|
|
from .config import load_config
|
|
from .access_reader import AccessReader
|
|
from .sql_writer import SqlWriter
|
|
from .capture import capture_file
|
|
from .cleanup import cleanup_file
|
|
from .logging_setup import setup_logging
|
|
|
|
log = logging.getLogger("sync.service")
|
|
|
|
|
|
def run(cfg):
|
|
"""Run ``cycle`` forever, sleeping ``poll_interval_seconds`` between passes.
|
|
|
|
Configures logging once on entry. Intended to be started by ``main()``
|
|
under the service host (e.g. NSSM). Not unit-tested (infinite loop);
|
|
``cycle()`` is the testable unit.
|
|
"""
|
|
setup_logging(cfg.logging)
|
|
while True:
|
|
cycle(cfg)
|
|
time.sleep(cfg.runtime.poll_interval_seconds)
|
|
|
|
|
|
def cycle(cfg):
|
|
"""One capture -> apply -> cleanup pass over all files.
|
|
|
|
Per-file capture/cleanup failures are logged and do not abort the cycle.
|
|
Apply failure does not block cleanup. The writer is always closed in a
|
|
``finally``. Safe to call directly from tests (does not sleep or loop).
|
|
"""
|
|
writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table)
|
|
try:
|
|
total_captured = 0
|
|
for fm in cfg.files:
|
|
reader = AccessReader(fm.source_path(cfg), cfg.access.driver)
|
|
try:
|
|
n = capture_file(fm, reader, writer, cfg)
|
|
total_captured += n
|
|
except Exception:
|
|
log.exception("capture failed for %s", fm.file)
|
|
finally:
|
|
reader.close()
|
|
log.info("captured %d rows", total_captured)
|
|
|
|
try:
|
|
writer.call_apply(cfg.runtime.max_retries)
|
|
log.info("apply done")
|
|
except Exception:
|
|
log.exception("apply failed")
|
|
|
|
for fm in cfg.files:
|
|
reader = AccessReader(fm.source_path(cfg), cfg.access.driver)
|
|
try:
|
|
c = cleanup_file(fm, reader, writer, cfg)
|
|
if c:
|
|
log.info("cleaned %d log rows from %s", c, fm.file)
|
|
except Exception:
|
|
log.exception("cleanup failed for %s", fm.file)
|
|
finally:
|
|
reader.close()
|
|
finally:
|
|
writer.close()
|
|
|
|
|
|
def main():
|
|
"""Entry point: load config from argv[1] (default config.yaml) and run."""
|
|
cfg_path = sys.argv[1] if len(sys.argv) > 1 else "config.yaml"
|
|
cfg = load_config(cfg_path)
|
|
try:
|
|
run(cfg)
|
|
except KeyboardInterrupt:
|
|
log.info("stopped")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|