Add legacy database structural reference for non-Windows machines
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.
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user