# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Reflect database schemas into Python model modules.\n\n:author: Bulgheroni Antonio\n"""
from pathlib import Path
from typing import Any
from pwiz import make_introspector # type: ignore[import-untyped]
from mafw.db.db_wizard import dump_models
[docs]
def _reflect_database(
*,
output_file: Path,
database: str,
engine: str | None,
schema: str | None,
host: str | None,
port: int | None,
user: str | None,
password: str | None,
tables: tuple[str, ...] | None,
preserve_order: bool,
include_views: bool,
ignore_unknown: bool,
snake_case: bool,
) -> None:
"""Inspect a database and write its model definitions to ``output_file``."""
selected_engine = engine or ('sqlite' if Path(database).exists() else 'postgresql')
if selected_engine in ('sqlite', 'sqlite3'):
keys = ['schema']
values: list[str | int | None] = [schema]
else:
keys = ['host', 'port', 'user', 'schema', 'password']
values = [host, port, user, schema, password]
connection_options: dict[str, Any] = {key: value for key, value in zip(keys, values, strict=True) if value}
introspector = make_introspector(selected_engine, database, **connection_options)
with output_file.open('w', encoding='utf-8') as out_file:
dump_models(
out_file,
introspector,
tables,
preserve_order=preserve_order,
include_views=include_views,
ignore_unknown=ignore_unknown,
snake_case=snake_case,
)