65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
"""Program schema — per-program DB table definitions + subprogram list."""
|
|
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
@dataclass
|
|
class ColumnDef:
|
|
name: str
|
|
type: str # SQL type: "CHAR(6)", "NUMERIC(4)", "VARCHAR(30)"
|
|
primary_key: bool = False
|
|
nullable: bool = False
|
|
default: Optional[str] = None
|
|
cobol_field: Optional[str] = None # COBOL field name if different
|
|
|
|
|
|
@dataclass
|
|
class TableDef:
|
|
name: str
|
|
columns: list[ColumnDef] = field(default_factory=list)
|
|
create_if_missing: bool = True
|
|
sql_name: Optional[str] = None # COBOL SQL table name if different from YAML name
|
|
|
|
|
|
@dataclass
|
|
class ProgramSchema:
|
|
program_id: str
|
|
db_tables: list[TableDef] = field(default_factory=list)
|
|
subprograms: list[str] = field(default_factory=list)
|
|
db_type: str = "SQLite"
|
|
db_name: str = "OVERTIME.DB"
|
|
|
|
@classmethod
|
|
def from_yaml(cls, path: str | Path) -> ProgramSchema:
|
|
import yaml
|
|
with open(path, encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
tables = []
|
|
for t in raw.get("db_tables", []):
|
|
cols = [ColumnDef(**c) for c in t.get("columns", [])]
|
|
tables.append(TableDef(
|
|
name=t["name"], columns=cols,
|
|
sql_name=t.get("sql_name"),
|
|
))
|
|
return cls(
|
|
program_id=raw["program_id"],
|
|
db_tables=tables,
|
|
subprograms=raw.get("subprograms", []),
|
|
db_type=raw.get("db_type", "SQLite"),
|
|
db_name=raw.get("db_name", "OVERTIME.DB"),
|
|
)
|
|
|
|
|
|
def load_schema(program_id: str, search_dirs: list[str | Path] | None = None) -> ProgramSchema:
|
|
"""Load per-program YAML schema by program ID."""
|
|
if search_dirs is None:
|
|
search_dirs = [Path(__file__).parent / "programs"]
|
|
for d in search_dirs:
|
|
p = Path(d) / f"{program_id}.yaml"
|
|
if p.exists():
|
|
return ProgramSchema.from_yaml(p)
|
|
raise FileNotFoundError(f"Schema not found for {program_id} in {search_dirs}")
|