updates 33.1 8

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,

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

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(

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",
 ]

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

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",
+]

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)

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))

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

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"