feat: config model and yaml loader

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>
This commit is contained in:
Misaka_Company
2026-07-14 11:53:20 +08:00
parent fb025bb061
commit 7c6cb10b91
4 changed files with 142 additions and 0 deletions

44
tests/test_config.py Normal file
View File

@@ -0,0 +1,44 @@
import pathlib, textwrap
from sync.config import load_config, FileMapping
def test_load_config_parses_fields(tmp_path):
cfg_text = textwrap.dedent("""
sql_server:
conn_str: "Driver={ODBC Driver 17 for SQL Server};Server=s;Database=d;UID=u;PWD=p;"
sync_queue_table: "dbo.SyncQueue"
access:
driver: "{Microsoft Access Driver (*.accdb, *.mdb)}"
roots:
2026: "\\\\\\\\srv\\\\2026"
2025: "\\\\\\\\srv\\\\2025"
runtime:
poll_interval_seconds: 10
capture_batch_size: 500
apply_batch_size: 200
max_retries: 5
retry_backoff_seconds: 30
cleanup_batch_size: 200
cleanup_lock_retries: 3
files:
- file: "氩弧焊.accdb"
root: 2026
schema: "TIGWelding"
year_suffix: "_YEAR2026"
exclude_tables: ["TableChangeLog", "氩弧焊每日催货落实记录_停"]
""")
p = tmp_path / "config.yaml"
p.write_text(cfg_text, encoding="utf-8")
cfg = load_config(str(p))
assert cfg.sql_server.sync_queue_table == "dbo.SyncQueue"
assert cfg.runtime.poll_interval_seconds == 10
assert cfg.files[0].schema == "TIGWelding"
assert cfg.files[0].year_suffix == "_YEAR2026"
assert "2026" in cfg.files[0].source_path(cfg)
def test_file_mapping_target_table_applies_year_suffix():
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026",
exclude_tables=[], include_tables=None)
assert fm.target_table("一车间记录") == "一车间记录_YEAR2026"
fm2 = FileMapping(file="y.accdb", root="2025", schema="s", year_suffix="",
exclude_tables=[], include_tables=None)
assert fm2.target_table("26年压力表合同数据") == "26年压力表合同数据"