rmancinas
7e65fc81d5
docs/LEGACY_DATABASES.md documents all three source Access databases (every table, column, type, and known data-quality quirk) generated from a live read of the real files, so no Windows/Access driver is needed to understand their structure going forward. New migration/ tooling: catalog_schema.py connects to the real files and walks every table (including excluded scratch tables); render_catalog_md.py renders that into the doc's appendix. Raw output checked in at migration/catalog.json so the doc can be regenerated without touching Access again.
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""
|
|
Generates a complete structural catalog of every table in the legacy Access
|
|
source files (including excluded/scratch tables, for documentation
|
|
completeness) and writes it as JSON.
|
|
|
|
This is the source data for docs/LEGACY_DATABASES.md. Re-run this any time
|
|
the source .accdb/.mdb files change, then regenerate the doc from the JSON.
|
|
|
|
Must run on a Windows machine with the Microsoft Access Database Engine
|
|
ODBC driver installed (pyodbc can't reach Access files otherwise) — see
|
|
docs/LEGACY_DATABASES.md for why non-Windows machines can't do this step
|
|
themselves and have to work from this catalog / the staged Parquet output
|
|
instead.
|
|
|
|
Usage:
|
|
python catalog_schema.py --output catalog.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import extract
|
|
from config import SOURCES
|
|
|
|
|
|
def get_access_type_labels(cnxn, table_name: str) -> dict[str, str] | None:
|
|
"""Best-effort: cursor.columns() gives real Access type names/sizes
|
|
(VARCHAR(255), DOUBLE, LONGCHAR, LONGBINARY, COUNTER, BIT, CURRENCY...)
|
|
but decode-errors on a subset of tables (see extract.py). Returns None
|
|
on failure so the caller can fall back to pandas-inferred dtypes."""
|
|
try:
|
|
cursor = cnxn.cursor()
|
|
labels = {}
|
|
for col in cursor.columns(table=table_name):
|
|
size = col.column_size
|
|
labels[col.column_name] = f"{col.type_name}({size})" if size else col.type_name
|
|
return labels
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def catalog_source(source_name: str, source_cfg: dict) -> dict:
|
|
path = source_cfg["path"]
|
|
result = {"path": str(path), "tables": {}}
|
|
|
|
if not Path(path).exists():
|
|
result["error"] = "file not found"
|
|
return result
|
|
|
|
cnxn = extract.connect(path)
|
|
tables = extract.list_tables(cnxn)
|
|
|
|
for table_name in tables:
|
|
entry = {"excluded": table_name in source_cfg["exclude"]}
|
|
access_types = get_access_type_labels(cnxn, table_name)
|
|
|
|
try:
|
|
df = extract.read_table(cnxn, table_name)
|
|
except Exception as exc: # noqa: BLE001
|
|
entry["error"] = str(exc)
|
|
result["tables"][table_name] = entry
|
|
continue
|
|
|
|
original_columns = df.attrs.get("original_columns", list(df.columns))
|
|
columns = []
|
|
for orig_name, sanitized_name in zip(original_columns, df.columns):
|
|
dtype = str(df[sanitized_name].dtype)
|
|
has_nulls = bool(df[sanitized_name].isna().any()) if len(df) else None
|
|
type_label = None
|
|
if access_types is not None:
|
|
type_label = access_types.get(orig_name)
|
|
columns.append(
|
|
{
|
|
"name": orig_name,
|
|
"access_type": type_label,
|
|
"pandas_dtype": dtype,
|
|
"nullable": has_nulls,
|
|
}
|
|
)
|
|
|
|
entry["row_count"] = len(df)
|
|
entry["column_count"] = len(columns)
|
|
entry["columns"] = columns
|
|
entry["access_type_metadata_available"] = access_types is not None
|
|
result["tables"][table_name] = entry
|
|
|
|
return result
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, default=Path("catalog.json"))
|
|
args = parser.parse_args()
|
|
|
|
catalog = {}
|
|
for source_name, source_cfg in SOURCES.items():
|
|
print(f"Cataloging {source_name}...")
|
|
catalog[source_name] = catalog_source(source_name, source_cfg)
|
|
|
|
args.output.write_text(json.dumps(catalog, indent=2, default=str), encoding="utf-8")
|
|
print(f"Wrote {args.output}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|