From dfb1991ae315c2373d4e2bfbad9cbceb3ff76e7c Mon Sep 17 00:00:00 2001 From: buenosairesam Date: Mon, 10 Aug 2026 00:32:47 -0300 Subject: [PATCH] updates 33.1 8 --- soleprint/station/tools/graphgen/schema.py | 17 +- soleprint/station/tools/modelgen/__init__.py | 7 +- soleprint/station/tools/modelgen/__main__.py | 92 ++++++++ .../tools/modelgen/generator/__init__.py | 4 + .../tools/modelgen/generator/jsonschema.py | 116 ++++++++++ .../tools/modelgen/loader/extract/__init__.py | 13 +- .../tools/modelgen/loader/extract/database.py | 192 ++++++++++++++++ .../loader/extract/sqlalchemy_models.py | 217 ++++++++++++++++++ .../station/tools/modelgen/loader/schema.py | 5 + .../station/tools/modelgen/pyproject.toml | 8 +- 10 files changed, 663 insertions(+), 8 deletions(-) create mode 100644 soleprint/station/tools/modelgen/generator/jsonschema.py create mode 100644 soleprint/station/tools/modelgen/loader/extract/database.py create mode 100644 soleprint/station/tools/modelgen/loader/extract/sqlalchemy_models.py diff --git a/soleprint/station/tools/graphgen/schema.py b/soleprint/station/tools/graphgen/schema.py index f1cd0bb..d1fc945 100644 --- a/soleprint/station/tools/graphgen/schema.py +++ b/soleprint/station/tools/graphgen/schema.py @@ -130,10 +130,19 @@ def _convert_modelgen(loader: Any, source: str) -> dict: fields = [] for field in model_def.fields: type_str = _type_str(field.type_hint) - fk_target = None + # Prefer explicit metadata set by introspection extractors + # (DatabaseExtractor / SqlAlchemyExtractor); fall back to inference. + fk_target = getattr(field, "foreign_key", None) + if fk_target: + relationships.append({ + "from_model": model_def.name, + "from_field": field.name, + "to_model": fk_target, + "type": "FK", + }) # FK: type name that matches another model - if type_str in all_names: + elif type_str in all_names: fk_target = type_str relationships.append({ "from_model": model_def.name, @@ -145,10 +154,12 @@ def _convert_modelgen(loader: Any, source: str) -> dict: elif type_str == "FK": fk_target = None # target unknown from extractor + is_pk = getattr(field, "primary_key", False) or field.name == "id" + fields.append({ "name": field.name, "type": type_str, - "pk": field.name == "id", + "pk": is_pk, "fk": fk_target, "m2m": type_str == "M2M", "nullable": field.optional, diff --git a/soleprint/station/tools/modelgen/__init__.py b/soleprint/station/tools/modelgen/__init__.py index e9d09fa..ce5334f 100644 --- a/soleprint/station/tools/modelgen/__init__.py +++ b/soleprint/station/tools/modelgen/__init__.py @@ -6,7 +6,8 @@ Generates typed models from various sources to various output formats. Input sources: - Configuration files (soleprint config.json style) - Python dataclasses in schema/ folder -- Existing codebases: Django, SQLAlchemy, Prisma (for extraction) +- Existing codebases: Django, SQLAlchemy (for extraction) +- Live databases: any SQLAlchemy dialect (PostgreSQL, MySQL, SQLite, ...) Output formats: - pydantic: Pydantic BaseModel classes @@ -14,15 +15,17 @@ Output formats: - typescript: TypeScript interfaces - protobuf: Protocol Buffer definitions - prisma: Prisma schema +- schema: graphgen-compatible schema.json (portable schema source) Usage: python -m soleprint.station.tools.modelgen from-config -c config.json -o models.py python -m soleprint.station.tools.modelgen from-schema -o models/ --targets pydantic,typescript python -m soleprint.station.tools.modelgen extract --source /path/to/django --targets pydantic + python -m soleprint.station.tools.modelgen from-db --url sqlite:///app.db --targets typescript,schema -o out/ python -m soleprint.station.tools.modelgen list-formats """ -__version__ = "0.2.0" +__version__ = "0.3.0" from .generator import GENERATORS, BaseGenerator from .loader import ConfigLoader, load_config diff --git a/soleprint/station/tools/modelgen/__main__.py b/soleprint/station/tools/modelgen/__main__.py index aa3331d..7eff8bf 100644 --- a/soleprint/station/tools/modelgen/__main__.py +++ b/soleprint/station/tools/modelgen/__main__.py @@ -178,6 +178,53 @@ def cmd_extract(args): print("Done!") +def cmd_from_db(args): + """Extract models from a live database (any SQLAlchemy dialect).""" + from .loader.extract.database import DatabaseExtractor + + include = {t.strip() for t in args.include.split(",")} if args.include else None + exclude = {t.strip() for t in args.exclude.split(",")} if args.exclude else None + + extractor = DatabaseExtractor( + url=args.url, + schema=args.schema, + include=include, + exclude=exclude, + ) + + print(f"Reflecting database: {args.url}") + try: + models, enums = extractor.extract() + except RuntimeError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + + print(f"Extracted {len(models)} models, {len(enums)} enums") + + # Parse targets + targets = [t.strip() for t in args.targets.split(",")] + output_dir = Path(args.output) + + for target in targets: + if target not in GENERATORS: + print(f"Warning: Unknown target '{target}', skipping", file=sys.stderr) + continue + + generator = GENERATORS[target]() + ext = generator.file_extension() + + # Determine output filename (use target name to avoid overwrites) + if len(targets) == 1 and args.output.endswith(ext): + output_file = output_dir + else: + output_file = output_dir / f"models_{target}{ext}" + + print(f"Generating {target} to: {output_file}") + generator.generate((models, enums), output_file) + + print("Done!") + + def cmd_generate(args): """Generate all targets from a JSON config file.""" import json @@ -337,6 +384,51 @@ def main(): ) extract_parser.set_defaults(func=cmd_extract) + # from-db command (live database introspection, any dialect) + db_parser = subparsers.add_parser( + "from-db", + help="Extract models from a live database (any SQLAlchemy dialect)", + ) + db_parser.add_argument( + "--url", + "-u", + type=str, + required=True, + help="SQLAlchemy connection URL (e.g. postgresql://…, mysql://…, sqlite:///path.db)", + ) + db_parser.add_argument( + "--schema", + type=str, + default=None, + help="Database schema to reflect (dialect-dependent; default: connection default)", + ) + db_parser.add_argument( + "--include", + type=str, + default=None, + help="Comma-separated table names to include (default: all)", + ) + db_parser.add_argument( + "--exclude", + type=str, + default=None, + help="Comma-separated table names to exclude", + ) + db_parser.add_argument( + "--output", + "-o", + type=str, + required=True, + help="Output path (file or directory)", + ) + db_parser.add_argument( + "--targets", + "-t", + type=str, + default="typescript", + help=f"Comma-separated output targets ({formats_str})", + ) + db_parser.set_defaults(func=cmd_from_db) # generate command (config-driven multi-target) gen_parser = subparsers.add_parser( diff --git a/soleprint/station/tools/modelgen/generator/__init__.py b/soleprint/station/tools/modelgen/generator/__init__.py index 7e1b55c..9a83ac5 100644 --- a/soleprint/station/tools/modelgen/generator/__init__.py +++ b/soleprint/station/tools/modelgen/generator/__init__.py @@ -14,6 +14,7 @@ from typing import Dict, Type from .base import BaseGenerator from .django import DjangoGenerator +from .jsonschema import JsonSchemaGenerator from .prisma import PrismaGenerator from .protobuf import ProtobufGenerator from .pydantic import PydanticGenerator @@ -32,6 +33,8 @@ GENERATORS: Dict[str, Type[BaseGenerator]] = { "proto": ProtobufGenerator, # Alias "prisma": PrismaGenerator, "strawberry": StrawberryGenerator, + "schema": JsonSchemaGenerator, + "jsonschema": JsonSchemaGenerator, # Alias } __all__ = [ @@ -42,5 +45,6 @@ __all__ = [ "TypeScriptGenerator", "ProtobufGenerator", "PrismaGenerator", + "JsonSchemaGenerator", "GENERATORS", ] diff --git a/soleprint/station/tools/modelgen/generator/jsonschema.py b/soleprint/station/tools/modelgen/generator/jsonschema.py new file mode 100644 index 0000000..6069dc2 --- /dev/null +++ b/soleprint/station/tools/modelgen/generator/jsonschema.py @@ -0,0 +1,116 @@ +""" +JSON Schema Generator + +Emits a graphgen-compatible ``schema.json`` — the canonical, portable schema +"source" artifact that downstream tools (graphgen, databrowse) read directly. + +Format (consumed by graphgen/schema.py::_load_json_schema): + + { + "models": { + "Users": { + "doc": "...", + "fields": { + "id": {"type": "int", "pk": true, "nullable": false}, + "name": {"type": "str", "nullable": false} + } + }, + "Posts": { + "fields": { + "user_id": {"type": "FK:Users", "nullable": false} + } + } + } + } +""" + +import json +from enum import Enum +from pathlib import Path +from typing import Any, List + +from ..helpers import unwrap_optional +from ..loader.schema import EnumDefinition, ModelDefinition +from .base import BaseGenerator + + +class JsonSchemaGenerator(BaseGenerator): + """Generates a graphgen-compatible schema.json from model definitions.""" + + def file_extension(self) -> str: + return ".json" + + def generate(self, models, output_path: Path) -> None: + output_path.parent.mkdir(parents=True, exist_ok=True) + + if hasattr(models, "models"): + # SchemaLoader + model_defs = list(models.models) + list(getattr(models, "api_models", [])) + elif isinstance(models, tuple): + # (models, enums) tuple + model_defs = list(models[0]) + elif isinstance(models, list): + model_defs = list(models) + else: + raise ValueError(f"Unsupported input type: {type(models)}") + + model_names = {self.map_name(m.name) for m in model_defs} + + out = {"models": {}} + for model_def in model_defs: + out["models"][self.map_name(model_def.name)] = self._model( + model_def, model_names + ) + + output_path.write_text(json.dumps(out, indent=2) + "\n") + + def _model(self, model_def: ModelDefinition, model_names: set) -> dict: + entry: dict = {} + if getattr(model_def, "docstring", None): + entry["doc"] = model_def.docstring.strip().splitlines()[0] + + fields: dict = {} + for field in model_def.fields: + fields[field.name] = self._field(field, model_names) + entry["fields"] = fields + return entry + + def _field(self, field: Any, model_names: set) -> dict: + base, is_opt = unwrap_optional(field.type_hint) + nullable = bool(getattr(field, "optional", False) or is_opt) + + fk_target = getattr(field, "foreign_key", None) + type_str = self._type_str(base) + + # Resolve the relationship-aware type string graphgen expects. + if fk_target: + type_value = f"FK:{self.map_name(fk_target)}" + elif type_str in model_names: + type_value = f"FK:{type_str}" + elif type_str == "M2M": + type_value = "M2M" + else: + type_value = type_str + + out: dict = {"type": type_value, "nullable": nullable} + if getattr(field, "primary_key", False): + out["pk"] = True + if getattr(field, "unique", False): + out["unique"] = True + return out + + @staticmethod + def _type_str(t: Any) -> str: + if t is None: + return "Any" + if isinstance(t, str): + return t + if isinstance(t, type) and issubclass(t, Enum): + return t.__name__ + if hasattr(t, "__name__"): + return t.__name__ + return str(t) + + +# Backwards/alternate name used by the registry alias. +SchemaGenerator = JsonSchemaGenerator diff --git a/soleprint/station/tools/modelgen/loader/extract/__init__.py b/soleprint/station/tools/modelgen/loader/extract/__init__.py index 82b1831..08a7bde 100644 --- a/soleprint/station/tools/modelgen/loader/extract/__init__.py +++ b/soleprint/station/tools/modelgen/loader/extract/__init__.py @@ -11,10 +11,19 @@ from typing import Dict, Type from .base import BaseExtractor from .django import DjangoExtractor +from .sqlalchemy_models import SqlAlchemyExtractor -# Registry of available extractors +# Registry of code-source extractors (auto-detectable via detect()). +# Note: live-database introspection lives in database.py (DatabaseExtractor), +# invoked explicitly via the `from-db` command since it takes a URL, not a path. EXTRACTORS: Dict[str, Type[BaseExtractor]] = { "django": DjangoExtractor, + "sqlalchemy": SqlAlchemyExtractor, } -__all__ = ["BaseExtractor", "DjangoExtractor", "EXTRACTORS"] +__all__ = [ + "BaseExtractor", + "DjangoExtractor", + "SqlAlchemyExtractor", + "EXTRACTORS", +] diff --git a/soleprint/station/tools/modelgen/loader/extract/database.py b/soleprint/station/tools/modelgen/loader/extract/database.py new file mode 100644 index 0000000..3de0285 --- /dev/null +++ b/soleprint/station/tools/modelgen/loader/extract/database.py @@ -0,0 +1,192 @@ +""" +Database Extractor + +All-terrain DDL extractor: reflects a live database via SQLAlchemy's Inspector +and produces modelgen's intermediate representation (ModelDefinition / EnumDefinition). + +Works across any dialect SQLAlchemy supports (PostgreSQL, MySQL, SQLite, ...) — +the dialect is abstracted by the connection URL. + +SQLAlchemy is an optional dependency (it is imported lazily) so that core modelgen +stays pure-stdlib and standalone. Install with: pip install "sqlalchemy>=2.0" +(plus a driver for non-sqlite dialects, e.g. psycopg2 / pymysql). + +Example: + extractor = DatabaseExtractor("sqlite:////tmp/test.db") + models, enums = extractor.extract() +""" + +from typing import Any, List, Optional + +from ..schema import EnumDefinition, FieldDefinition, ModelDefinition + +_INSTALL_HINT = ( + "DatabaseExtractor requires SQLAlchemy. Install it with:\n" + ' pip install "sqlalchemy>=2.0"\n' + "(plus a driver for your dialect, e.g. psycopg2 for PostgreSQL, pymysql for MySQL; " + "sqlite needs none)." +) + + +def _to_model_name(table_name: str) -> str: + """Convert a table name to a PascalCase model name (users -> Users).""" + parts = [p for p in table_name.replace("-", "_").split("_") if p] + return "".join(p[:1].upper() + p[1:] for p in parts) or table_name + + +def _map_column_type(col_type: Any) -> tuple[Any, Optional[str]]: + """Map a SQLAlchemy column type to an IR type hint. + + Returns (type_hint, enum_name). type_hint is either a Python type or one of + modelgen's special string names (see types.py). enum_name is set only for + enum columns, so the caller can register/reference the enum. + """ + import sqlalchemy as sa + + # Enum (named DB enum, e.g. Postgres ENUM, or SQLAlchemy Enum) + if isinstance(col_type, sa.Enum): + name = col_type.name or "Enum" + return _to_model_name(name), name + + # Dialect-specific types are matched by class name (UUID, JSONB, ARRAY, ...) + tname = type(col_type).__name__.upper() + if "UUID" in tname: + return "UUID", None + if "JSON" in tname: # JSON, JSONB + return "dict", None + if "ARRAY" in tname: + return "list", None + + # Generic types — most specific first (subclass relationships matter). + if isinstance(col_type, sa.Boolean): + return bool, None + if isinstance(col_type, sa.BigInteger): + return "bigint", None + if isinstance(col_type, (sa.SmallInteger, sa.Integer)): + return int, None + if isinstance(col_type, (sa.Numeric, sa.Float)): + return float, None + if isinstance(col_type, sa.Text): + return "text", None + if isinstance(col_type, sa.String): + return str, None + if isinstance(col_type, (sa.DateTime, sa.Date, sa.Time)): + return "datetime", None + if isinstance(col_type, sa.LargeBinary): + return "bytes", None + + # Fallback: try the type's declared python_type. + try: + py = col_type.python_type + return {str: str, int: int, float: float, bool: bool}.get(py, str), None + except Exception: + return str, None + + +class DatabaseExtractor: + """Reflects a live database into modelgen's IR via SQLAlchemy.""" + + def __init__( + self, + url: str, + schema: Optional[str] = None, + include: Optional[set] = None, + exclude: Optional[set] = None, + ): + self.url = url + self.schema = schema + self.include = include + self.exclude = exclude or set() + + def extract(self) -> tuple[List[ModelDefinition], List[EnumDefinition]]: + try: + import sqlalchemy as sa + except ImportError as e: # pragma: no cover - exercised only without the extra + raise RuntimeError(_INSTALL_HINT) from e + + engine = sa.create_engine(self.url) + inspector = sa.inspect(engine) + + table_names = inspector.get_table_names(schema=self.schema) + if self.include: + table_names = [t for t in table_names if t in self.include] + table_names = [t for t in table_names if t not in self.exclude] + + models: List[ModelDefinition] = [] + enums: dict[str, EnumDefinition] = {} + + for table in table_names: + models.append(self._extract_table(inspector, table, enums)) + + engine.dispose() + return models, list(enums.values()) + + def _extract_table( + self, inspector: Any, table: str, enums: dict + ) -> ModelDefinition: + columns = inspector.get_columns(table, schema=self.schema) + + # Primary key columns + try: + pk_cols = set( + inspector.get_pk_constraint(table, schema=self.schema).get( + "constrained_columns", [] + ) + or [] + ) + except Exception: + pk_cols = set() + + # Single-column unique constraints + unique_cols: set = set() + try: + for uc in inspector.get_unique_constraints(table, schema=self.schema): + cols = uc.get("column_names", []) or [] + if len(cols) == 1: + unique_cols.add(cols[0]) + except Exception: + pass + + # Foreign keys: constrained column -> referred model name + fk_targets: dict = {} + try: + for fk in inspector.get_foreign_keys(table, schema=self.schema): + referred = fk.get("referred_table") + for col in fk.get("constrained_columns", []) or []: + if referred: + fk_targets[col] = _to_model_name(referred) + except Exception: + pass + + fields: List[FieldDefinition] = [] + for col in columns: + name = col["name"] + type_hint, enum_name = _map_column_type(col["type"]) + + if enum_name and enum_name not in enums: + values = list(getattr(col["type"], "enums", []) or []) + enums[enum_name] = EnumDefinition( + name=_to_model_name(enum_name), + values=[(v, v) for v in values], + ) + + fk_target = fk_targets.get(name) + is_pk = name in pk_cols + # Keep the scalar column type as the type hint; the relationship is + # carried by foreign_key metadata (downstream consumers like graphgen + # read that, so non-graph targets keep the correct scalar type). + optional = bool(col.get("nullable", True)) and not is_pk + + fields.append( + FieldDefinition( + name=name, + type_hint=type_hint, + default=col.get("default"), + optional=optional, + primary_key=is_pk, + foreign_key=fk_target, + unique=name in unique_cols, + ) + ) + + return ModelDefinition(name=_to_model_name(table), fields=fields) diff --git a/soleprint/station/tools/modelgen/loader/extract/sqlalchemy_models.py b/soleprint/station/tools/modelgen/loader/extract/sqlalchemy_models.py new file mode 100644 index 0000000..8401c0b --- /dev/null +++ b/soleprint/station/tools/modelgen/loader/extract/sqlalchemy_models.py @@ -0,0 +1,217 @@ +""" +SQLAlchemy Extractor + +Extracts model definitions from SQLAlchemy declarative model *code* (not a live +database — see database.py for live introspection). + +Pure AST parsing (no SQLAlchemy import needed), mirroring django.py. Detects +classes that declare ``__tablename__`` or inherit from a declarative ``Base`` / +``DeclarativeBase`` and parses their ``Column(...)`` assignments, including +``ForeignKey(...)`` relationships. +""" + +import ast +from pathlib import Path +from typing import Dict, List, Optional + +from ..schema import EnumDefinition, FieldDefinition, ModelDefinition +from .base import BaseExtractor + +# SQLAlchemy column type names -> modelgen IR type hints. +SQLALCHEMY_TYPES = { + "Integer": int, + "SmallInteger": int, + "BigInteger": "bigint", + "String": str, + "Unicode": str, + "VARCHAR": str, + "Text": "text", + "UnicodeText": "text", + "Boolean": bool, + "Float": float, + "Numeric": float, + "DECIMAL": float, + "Date": "datetime", + "DateTime": "datetime", + "Time": "datetime", + "JSON": "dict", + "JSONB": "dict", + "UUID": "UUID", + "Uuid": "UUID", + "LargeBinary": "bytes", + "ARRAY": "list", +} + + +def _to_model_name(table_name: str) -> str: + parts = [p for p in table_name.replace("-", "_").split("_") if p] + return "".join(p[:1].upper() + p[1:] for p in parts) or table_name + + +class SqlAlchemyExtractor(BaseExtractor): + """Extracts models from SQLAlchemy declarative model code.""" + + def detect(self) -> bool: + for py in self.source_path.rglob("*.py"): + try: + content = py.read_text() + except Exception: + continue + if "sqlalchemy" in content and ( + "__tablename__" in content or "declarative_base" in content + or "DeclarativeBase" in content + ): + return True + return False + + def extract(self) -> tuple[List[ModelDefinition], List[EnumDefinition]]: + # Pass 1: collect class -> __tablename__ so FK 'table.col' refs resolve + # back to the owning model class name. + class_nodes: List[ast.ClassDef] = [] + for py in self.source_path.rglob("*.py"): + try: + tree = ast.parse(py.read_text()) + except Exception: + continue + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef) and self._is_model(node): + class_nodes.append(node) + + table_to_class: Dict[str, str] = {} + for node in class_nodes: + tablename = self._tablename(node) + if tablename: + table_to_class[tablename] = node.name + + # Pass 2: build models. + models = [self._parse_model(node, table_to_class) for node in class_nodes] + return models, [] + + def _is_model(self, node: ast.ClassDef) -> bool: + if self._tablename(node): + return True + for base in node.bases: + if isinstance(base, ast.Name) and base.id in ("Base", "DeclarativeBase"): + return True + if isinstance(base, ast.Attribute) and base.attr in ( + "Base", + "DeclarativeBase", + ): + return True + return False + + def _tablename(self, node: ast.ClassDef) -> Optional[str]: + for item in node.body: + if isinstance(item, ast.Assign): + for target in item.targets: + if ( + isinstance(target, ast.Name) + and target.id == "__tablename__" + and isinstance(item.value, ast.Constant) + ): + return str(item.value.value) + return None + + def _parse_model( + self, node: ast.ClassDef, table_to_class: Dict[str, str] + ) -> ModelDefinition: + fields: List[FieldDefinition] = [] + for item in node.body: + field = None + if isinstance(item, ast.Assign): + if item.targets and isinstance(item.targets[0], ast.Name): + field = self._parse_column( + item.targets[0].id, item.value, table_to_class + ) + elif isinstance(item, ast.AnnAssign) and isinstance( + item.target, ast.Name + ): + field = self._parse_column( + item.target.id, item.value, table_to_class + ) + if field: + fields.append(field) + + return ModelDefinition( + name=node.name, fields=fields, docstring=ast.get_docstring(node) + ) + + def _parse_column( + self, name: str, value: ast.expr, table_to_class: Dict[str, str] + ) -> Optional[FieldDefinition]: + if name.startswith("_"): + return None + if not isinstance(value, ast.Call): + return None + + func_name = self._call_name(value) + # Support both `Column(...)` and 2.0-style `mapped_column(...)`. + if func_name not in ("Column", "mapped_column"): + return None + + type_hint = str + fk_target: Optional[str] = None + + # Positional args: a type (Name or Call) and/or a ForeignKey(...) call. + for arg in value.args: + if isinstance(arg, ast.Call) and self._call_name(arg) == "ForeignKey": + fk_target = self._foreign_key_target(arg, table_to_class) + elif isinstance(arg, ast.Name): + type_hint = SQLALCHEMY_TYPES.get(arg.id, str) + elif isinstance(arg, ast.Call): + inner = self._call_name(arg) + if inner == "ForeignKey": + fk_target = self._foreign_key_target(arg, table_to_class) + elif inner: + type_hint = SQLALCHEMY_TYPES.get(inner, str) + + primary_key = False + nullable = True + unique = False + for kw in value.keywords: + if kw.arg == "primary_key" and isinstance(kw.value, ast.Constant): + primary_key = kw.value.value is True + elif kw.arg == "nullable" and isinstance(kw.value, ast.Constant): + nullable = kw.value.value is not False + elif kw.arg == "unique" and isinstance(kw.value, ast.Constant): + unique = kw.value.value is True + elif kw.arg == "ForeignKey" and isinstance(kw.value, ast.Call): + fk_target = self._foreign_key_target(kw.value, table_to_class) + + # Primary keys are implicitly NOT NULL. + if primary_key: + nullable = False + + # Keep the scalar column type; the relationship is carried by the + # foreign_key metadata (graphgen reads it; scalar targets stay correct). + + return FieldDefinition( + name=name, + type_hint=type_hint, + default=None, + optional=nullable, + primary_key=primary_key, + foreign_key=fk_target, + unique=unique, + ) + + @staticmethod + def _call_name(call: ast.Call) -> Optional[str]: + if isinstance(call.func, ast.Name): + return call.func.id + if isinstance(call.func, ast.Attribute): + return call.func.attr + return None + + @staticmethod + def _foreign_key_target( + call: ast.Call, table_to_class: Dict[str, str] + ) -> Optional[str]: + if not call.args: + return None + arg = call.args[0] + if not isinstance(arg, ast.Constant) or not isinstance(arg.value, str): + return None + # "table.column" -> table -> owning model class name (or PascalCase table) + table = arg.value.split(".")[0] + return table_to_class.get(table, _to_model_name(table)) diff --git a/soleprint/station/tools/modelgen/loader/schema.py b/soleprint/station/tools/modelgen/loader/schema.py index 10f461b..ac8705e 100644 --- a/soleprint/station/tools/modelgen/loader/schema.py +++ b/soleprint/station/tools/modelgen/loader/schema.py @@ -27,6 +27,11 @@ class FieldDefinition: type_hint: Any default: Any = dc.MISSING optional: bool = False + # Optional DB/schema metadata (set by introspection extractors; ignored by + # loaders/generators that don't need it). + primary_key: bool = False + foreign_key: Optional[str] = None # target model name + unique: bool = False @dataclass diff --git a/soleprint/station/tools/modelgen/pyproject.toml b/soleprint/station/tools/modelgen/pyproject.toml index 4f30d69..63b5c1a 100644 --- a/soleprint/station/tools/modelgen/pyproject.toml +++ b/soleprint/station/tools/modelgen/pyproject.toml @@ -4,11 +4,17 @@ build-backend = "setuptools.build_meta" [project] name = "soleprint-modelgen" -version = "0.2.0" +version = "0.3.0" description = "Multi-source, multi-target model code generator" requires-python = ">=3.10" dependencies = [] +# Optional extras. Core modelgen is pure-stdlib and standalone; live-database +# extraction (`from-db`) needs SQLAlchemy plus a driver for non-sqlite dialects +# (e.g. psycopg2 for PostgreSQL, pymysql for MySQL). +[project.optional-dependencies] +db = ["sqlalchemy>=2.0"] + [project.scripts] modelgen = "modelgen.__main__:main"