feat: config module with TOML loading
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
41
src/order_bak/config.py
Normal file
41
src/order_bak/config.py
Normal file
@@ -0,0 +1,41 @@
|
|||||||
|
import sys
|
||||||
|
import tomllib
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Config:
|
||||||
|
azure_connection_string: str
|
||||||
|
scan_page_size: int = 500
|
||||||
|
scan_delay_ms: int = 200
|
||||||
|
scan_dir: Path = Path("./scan")
|
||||||
|
delete_batch_size: int = 100
|
||||||
|
delete_delay_ms: int = 300
|
||||||
|
|
||||||
|
|
||||||
|
def load_config(config_path: Path) -> Config:
|
||||||
|
if not config_path.exists():
|
||||||
|
print(f"Config file not found: {config_path}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
with open(config_path, "rb") as f:
|
||||||
|
raw = tomllib.load(f)
|
||||||
|
|
||||||
|
azure = raw.get("azure", {})
|
||||||
|
connection_string = azure.get("connection_string")
|
||||||
|
if not connection_string:
|
||||||
|
print("Missing azure.connection_string in config")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
scan = raw.get("scan", {})
|
||||||
|
delete = raw.get("delete", {})
|
||||||
|
|
||||||
|
return Config(
|
||||||
|
azure_connection_string=connection_string,
|
||||||
|
scan_page_size=scan.get("page_size", 500),
|
||||||
|
scan_delay_ms=scan.get("delay_ms", 200),
|
||||||
|
scan_dir=Path(scan.get("scan_dir", "./scan")),
|
||||||
|
delete_batch_size=delete.get("batch_size", 100),
|
||||||
|
delete_delay_ms=delete.get("delay_ms", 300),
|
||||||
|
)
|
||||||
42
tests/test_config.py
Normal file
42
tests/test_config.py
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from order_bak.config import load_config
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_config_defaults(tmp_path):
|
||||||
|
config_path = tmp_path / "config.toml"
|
||||||
|
config_path.write_text(
|
||||||
|
'[azure]\nconnection_string = "conn_str_val"\n'
|
||||||
|
)
|
||||||
|
cfg = load_config(config_path)
|
||||||
|
assert cfg.azure_connection_string == "conn_str_val"
|
||||||
|
assert cfg.scan_page_size == 500
|
||||||
|
assert cfg.scan_delay_ms == 200
|
||||||
|
assert cfg.scan_dir == Path("./scan")
|
||||||
|
assert cfg.delete_batch_size == 100
|
||||||
|
assert cfg.delete_delay_ms == 300
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_config_custom_values(tmp_path):
|
||||||
|
config_path = tmp_path / "config.toml"
|
||||||
|
config_path.write_text(
|
||||||
|
'[azure]\nconnection_string = "conn_str_val"\n'
|
||||||
|
"[scan]\npage_size = 1000\ndelay_ms = 500\nscan_dir = \"/data/scan\"\n"
|
||||||
|
"[delete]\nbatch_size = 50\ndelay_ms = 100\n"
|
||||||
|
)
|
||||||
|
cfg = load_config(config_path)
|
||||||
|
assert cfg.scan_page_size == 1000
|
||||||
|
assert cfg.scan_delay_ms == 500
|
||||||
|
assert cfg.scan_dir == Path("/data/scan")
|
||||||
|
assert cfg.delete_batch_size == 50
|
||||||
|
assert cfg.delete_delay_ms == 100
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_config_missing_azure_raises(tmp_path):
|
||||||
|
config_path = tmp_path / "config.toml"
|
||||||
|
config_path.write_text("[scan]\npage_size = 100\n")
|
||||||
|
try:
|
||||||
|
load_config(config_path)
|
||||||
|
assert False, "should have raised"
|
||||||
|
except SystemExit:
|
||||||
|
pass
|
||||||
Reference in New Issue
Block a user