update_reference_types.py

#!/usr/bin/env python3
"""Regenerate the logicalType -> native type tables on the Data Source Reference pages.

For each page under docs/docs/reference/ the table is derived directly from the
type converters in datacontract.export.sql_type_converter (and, for file-based
sources, datacontract.export.duckdb_type_converter), so the docs always match
the code. The table is injected between AUTOGENERATED TYPE MAPPING markers;
on the first run the block is appended to the end of the page.

    python update_reference_types.py
"""

from open_data_contract_standard.model import SchemaProperty

from datacontract.export.duckdb_type_converter import convert_to_duckdb_csv_type
from datacontract.export.sql_type_converter import convert_to_sql_type
from datacontract.model.exceptions import DataContractException
from docs_examples import DOCS

# JSX comments (not HTML comments) — MDX 3 rejects `<!-- -->`.
START = "{/* AUTOGENERATED TYPE MAPPING: do not edit by hand; regenerate with update_reference_types.py */}"
END = "{/* END AUTOGENERATED TYPE MAPPING */}"

LOGICAL_TYPES = ["string", "integer", "number", "boolean", "date", "timestamp", "time", "object", "array"]

# page -> (server_type for convert_to_sql_type, display name of the type column)
SQL_PAGES = {
    "snowflake": ("snowflake", "Snowflake type"),
    "bigquery": ("bigquery", "BigQuery type"),
    "databricks": ("databricks", "Databricks type"),
    "postgres": ("postgres", "Postgres type"),
    "redshift": ("redshift", "Redshift type"),
    "mysql": ("mysql", "MySQL type"),
    "sqlserver": ("sqlserver", "SQL Server type"),
    "oracle": ("oracle", "Oracle type"),
    "trino": ("trino", "Trino type"),
    "dataframe": ("dataframe", "Spark type"),
}

# file-based pages all read through DuckDB
FILE_PAGES = ["local", "s3", "gcs", "azure"]

SQL_INTRO = (
    "When no `physicalType` is declared, the CLI derives the native type from the "
    "`logicalType` — for example in `datacontract export sql` and the dbt exports. "
    "This table is generated from the converter in the CLI's code:"
)

FILE_INTRO = (
    "When no `physicalType` is declared, the CLI derives the DuckDB column type from the "
    "`logicalType` when reading files. This table is generated from the converters in the CLI's code:"
)


def _convert(prop: SchemaProperty, fn, *args) -> str:
    try:
        result = fn(prop, *args)
    except (NotImplementedError, DataContractException):
        return "*(not supported)*"
    return f"`{result}`" if result else "—"


def sql_table(server_type: str, column: str) -> str:
    rows = [
        f"| `{lt}` | {_convert(SchemaProperty(name='example', logicalType=lt), convert_to_sql_type, server_type)} |"
        for lt in LOGICAL_TYPES
    ]
    return f"| `logicalType` | {column} |\n|---|---|\n" + "\n".join(rows)


def file_table() -> str:
    rows = []
    for lt in LOGICAL_TYPES:
        prop = SchemaProperty(name="example", logicalType=lt)
        csv = _convert(prop, convert_to_duckdb_csv_type)
        parquet = _convert(prop, convert_to_sql_type, "local")
        rows.append(f"| `{lt}` | {csv} | {parquet} |")
    return "| `logicalType` | CSV read type | Parquet read type |\n|---|---|---|\n" + "\n".join(rows)


def upsert(page_name: str, block: str) -> None:
    page = DOCS / "reference" / f"{page_name}.md"
    text = page.read_text()
    marked = f"{START}\n\n{block}\n\n{END}"
    if START in text and END in text:
        start_idx = text.index(START)
        end_idx = text.index(END) + len(END)
        new = text[:start_idx] + marked + text[end_idx:]
    else:
        new = text.rstrip("\n") + "\n\n" + marked + "\n"
    page.write_text(new)
    print(f"ok reference/{page_name}.md")


def main() -> None:
    for page_name, (server_type, column) in SQL_PAGES.items():
        block = f"### Logical type mapping\n\n{SQL_INTRO}\n\n{sql_table(server_type, column)}"
        upsert(page_name, block)
    for page_name in FILE_PAGES:
        block = f"### Logical type mapping\n\n{FILE_INTRO}\n\n{file_table()}"
        upsert(page_name, block)


if __name__ == "__main__":
    main()