diff --git a/cli/__init__.py b/cli/__init__.py index e69de29..92c90cb 100644 --- a/cli/__init__.py +++ b/cli/__init__.py @@ -0,0 +1,10 @@ +"""FastStack CLI — Click-based entry point for the `faststack` command.""" + +import click + + +@click.group() +@click.version_option(version="0.1.0", prog_name="faststack") +def cli() -> None: + """FastStack — Hybrid FastAPI framework + CLI generator.""" + pass diff --git a/cli/field_mappings.py b/cli/field_mappings.py new file mode 100644 index 0000000..6278f2d --- /dev/null +++ b/cli/field_mappings.py @@ -0,0 +1,171 @@ +"""Type mappings from YAML field types to SQLAlchemy, Pydantic, and Python types. + +Supports all 13 field types defined in the FastStack design plan: +string, text, integer, float, boolean, datetime, date, uuid, decimal, json, enum, array, jsonb. +""" + +from __future__ import annotations + +# --------------------------------------------------------------------------- +# YAML type -> SQLAlchemy column type string (used in template rendering) +# --------------------------------------------------------------------------- +SQLALCHEMY_TYPE_MAP: dict[str, str] = { + "string": "String(255)", + "text": "Text", + "integer": "Integer", + "float": "Float", + "boolean": "Boolean", + "datetime": "DateTime", + "date": "Date", + "uuid": "UUID(as_uuid=True)", + "decimal": "Numeric(10, 2)", + "json": "JSON", + "enum": "Enum", # special handling -- needs the enum class name + "array": "ARRAY", # special handling -- needs inner type, PostgreSQL only + "jsonb": "JSONB", +} + +# --------------------------------------------------------------------------- +# YAML type -> Pydantic field type string +# --------------------------------------------------------------------------- +PYDANTIC_TYPE_MAP: dict[str, str] = { + "string": "str", + "text": "str", + "integer": "int", + "float": "float", + "boolean": "bool", + "datetime": "datetime", + "date": "date", + "uuid": "UUID", + "decimal": "Decimal", + "json": "dict", + "enum": "str", # will be Literal[...] in schema, str in general + "array": "list", # will be list[inner] in schema + "jsonb": "dict", +} + +# --------------------------------------------------------------------------- +# YAML type -> Python native type string +# --------------------------------------------------------------------------- +PYTHON_TYPE_MAP: dict[str, str] = { + "string": "str", + "text": "str", + "integer": "int", + "float": "float", + "boolean": "bool", + "datetime": "datetime", + "date": "date", + "uuid": "uuid.UUID", + "decimal": "Decimal", + "json": "dict", + "enum": "str", + "array": "list", + "jsonb": "dict", +} + +# --------------------------------------------------------------------------- +# SQLAlchemy import map -- which imports are needed for each YAML type +# --------------------------------------------------------------------------- +_SQLALCHEMY_IMPORT_MAP: dict[str, list[str]] = { + "string": ["String"], + "text": ["Text"], + "integer": ["Integer"], + "float": ["Float"], + "boolean": ["Boolean"], + "datetime": ["DateTime"], + "date": ["Date"], + "uuid": ["UUID"], + "decimal": ["Numeric"], + "json": ["JSON"], + "enum": ["Enum"], + "array": ["ARRAY"], + "jsonb": ["JSONB"], +} + +ALL_YAML_TYPES = frozenset(SQLALCHEMY_TYPE_MAP.keys()) + + +def _validate_type(yaml_type: str) -> None: + """Raise ValueError if *yaml_type* is not a recognised YAML field type.""" + if yaml_type not in ALL_YAML_TYPES: + raise ValueError( + f"Unknown YAML field type: {yaml_type!r}. " + f"Supported types: {', '.join(sorted(ALL_YAML_TYPES))}" + ) + + +# --------------------------------------------------------------------------- +# Public helpers +# --------------------------------------------------------------------------- + + +def get_sqlalchemy_type(yaml_type: str, **kwargs: str) -> str: + """Return the full SQLAlchemy column-type expression for *yaml_type*. + + Special keyword arguments: + - ``enum_class`` (str): Required when *yaml_type* is ``"enum"``. + Produces e.g. ``Enum(StatusEnum)``. + - ``items`` (str): Required when *yaml_type* is ``"array"``. + Produces e.g. ``ARRAY(String)``. + """ + _validate_type(yaml_type) + + if yaml_type == "enum": + enum_class = kwargs.get("enum_class") + if not enum_class: + raise ValueError("enum type requires 'enum_class' kwarg for SQLAlchemy mapping") + return f"Enum({enum_class})" + + if yaml_type == "array": + items = kwargs.get("items") + if not items: + raise ValueError("array type requires 'items' kwarg for SQLAlchemy mapping") + # Map the inner YAML type to its SQLAlchemy type name (without params) + inner_sa = SQLALCHEMY_TYPE_MAP.get(items) + if inner_sa is None: + raise ValueError(f"Unknown inner type for array: {items!r}") + # Strip any parenthesised params for the inner type -- ARRAY(String) not ARRAY(String(255)) + inner_name = inner_sa.split("(")[0] + return f"ARRAY({inner_name})" + + return SQLALCHEMY_TYPE_MAP[yaml_type] + + +def get_pydantic_type(yaml_type: str, **kwargs: str | list[str]) -> str: + """Return the Pydantic type annotation string for *yaml_type*. + + Special keyword arguments: + - ``values`` (list[str]): For ``"enum"`` type, returns ``Literal["a", "b", "c"]``. + - ``items`` (str): For ``"array"`` type, returns e.g. ``list[str]``. + """ + _validate_type(yaml_type) + + if yaml_type == "enum": + values = kwargs.get("values") + if values and isinstance(values, list): + quoted = ", ".join(f'"{v}"' for v in values) + return f"Literal[{quoted}]" + return PYDANTIC_TYPE_MAP[yaml_type] + + if yaml_type == "array": + items = kwargs.get("items") + if items and isinstance(items, str): + inner_py = PYDANTIC_TYPE_MAP.get(items) + if inner_py is None: + raise ValueError(f"Unknown inner type for array: {items!r}") + return f"list[{inner_py}]" + return PYDANTIC_TYPE_MAP[yaml_type] + + return PYDANTIC_TYPE_MAP[yaml_type] + + +def get_python_type(yaml_type: str, **kwargs: str | list[str]) -> str: + """Return the Python native type string for *yaml_type*.""" + _validate_type(yaml_type) + return PYTHON_TYPE_MAP[yaml_type] + + +def get_sqlalchemy_imports(yaml_type: str) -> list[str]: + """Return the list of SQLAlchemy imports needed for *yaml_type*.""" + _validate_type(yaml_type) + return list(_SQLALCHEMY_IMPORT_MAP[yaml_type]) diff --git a/cli/model_introspector.py b/cli/model_introspector.py new file mode 100644 index 0000000..eb76d83 --- /dev/null +++ b/cli/model_introspector.py @@ -0,0 +1,464 @@ +"""AST-based model introspector for SQLAlchemy 2.0 models. + +Reads Python model files and extracts field definitions, relationships, +and metadata into EntityDefinition objects — the same format as the +YAML parser produces, so the rest of the pipeline treats both sources +identically. + +Only supports patterns that FastStack generates. Does NOT attempt to +handle arbitrary SQLAlchemy code. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +from cli.yaml_parser import EntityDefinition, FieldDefinition, RelationshipDefinition + +# ── Reverse type mapping: Python annotation → YAML type ───────────── + +ANNOTATION_TO_YAML_TYPE: dict[str, str] = { + "str": "string", + "int": "integer", + "float": "float", + "bool": "boolean", + "datetime": "datetime", + "date": "date", + "UUID": "uuid", + "uuid.UUID": "uuid", + "Decimal": "decimal", + "dict": "json", +} + + +# ── Public API ─────────────────────────────────────────────────────── + + +def introspect_model(path: Path) -> EntityDefinition: + """Parse a SQLAlchemy model file and return an EntityDefinition. + + Extracts: + - Class name and base class from class definition + - ``__tablename__`` from class body + - ``Mapped[type]`` fields from annotated assignments + - ``mapped_column()`` arguments (unique, nullable, default, ForeignKey) + - ``relationship()`` calls (back_populates, target entity) + - Enum classes defined above the model (``str, enum.Enum`` subclasses) + """ + source = path.read_text() + tree = ast.parse(source) + + # First pass: collect enum classes defined in the file. + enum_classes = _collect_enum_classes(tree) + + # Second pass: find the model class (first non-enum class). + model_node = _find_model_class(tree, enum_classes) + if model_node is None: + raise ValueError(f"No model class found in {path}") + + name = model_node.name + base = _extract_base_name(model_node) + table_name = _extract_tablename(model_node) + + fields: list[FieldDefinition] = [] + explicit_relationships: list[RelationshipDefinition] = [] + # FK-derived relationships are added only when no explicit relationship() + # targets the same entity, so we collect them separately. + fk_relationships: list[RelationshipDefinition] = [] + + for node in model_node.body: + # Skip non-annotated assignments and plain assignments (like __tablename__). + if not isinstance(node, ast.AnnAssign): + continue + if node.target is None or not isinstance(node.target, ast.Name): + continue + + field_name = node.target.id + annotation = node.annotation + + # ── relationship() ──────────────────────────────────── + if _is_relationship_call(node.value): + assert isinstance(node.value, ast.Call) # narrowing for mypy + rel = _parse_relationship(field_name, annotation, node.value, name) + if rel is not None: + explicit_relationships.append(rel) + continue + + # ── mapped_column() or bare annotation ──────────────── + mapped_type = _resolve_annotation_type(annotation, enum_classes) + nullable = _annotation_is_nullable(annotation) + + # Defaults from mapped_column() args + unique = False + default = None + foreign_key: str | None = None + enum_values: list[str] | None = None + + if mapped_type == "enum": + enum_name = _extract_enum_class_name(annotation) + if enum_name and enum_name in enum_classes: + enum_values = enum_classes[enum_name] + + if node.value is not None and _is_mapped_column_call(node.value): + assert isinstance(node.value, ast.Call) # narrowing for mypy + col_info = _parse_mapped_column(node.value) + unique = col_info.get("unique", False) + if col_info.get("default") is not None: + default = col_info["default"] + if col_info.get("foreign_key") is not None: + foreign_key = col_info["foreign_key"] + + # Record FK-derived relationship (may be superseded by an explicit one). + if foreign_key is not None: + ref_entity = _fk_to_entity_name(foreign_key, name, table_name) + rel_type = "self_referential" if ref_entity == name else "many_to_one" + fk_relationships.append( + RelationshipDefinition( + field_name=field_name, + type=rel_type, + target_entity=ref_entity, + back_populates=None, + ) + ) + + field = FieldDefinition( + name=field_name, + type=mapped_type, + required=not nullable, + unique=unique, + default=default, + references=_fk_to_entity_name(foreign_key, name, table_name) if foreign_key else None, + enum_values=enum_values if enum_values else [], + ) + fields.append(field) + + # Merge relationships: explicit relationship() calls take precedence over + # FK-derived ones. Only add a FK-derived relationship when no explicit + # relationship() targets the same entity with a compatible type. + relationships = list(explicit_relationships) + for fk_rel in fk_relationships: + has_explicit = any( + r.target_entity == fk_rel.target_entity + and r.type in ("many_to_one", "self_referential") + for r in explicit_relationships + ) + if not has_explicit: + relationships.append(fk_rel) + + return EntityDefinition( + name=name, + base=base, + table_name=table_name, + fields=fields, + relationships=relationships, + ) + + +# ── Enum collection ────────────────────────────────────────────────── + + +def _collect_enum_classes(tree: ast.Module) -> dict[str, list[str]]: + """Return ``{ClassName: [value1, value2, ...]}`` for ``str, enum.Enum`` classes.""" + enums: dict[str, list[str]] = {} + for node in ast.iter_child_nodes(tree): + if not isinstance(node, ast.ClassDef): + continue + if not _is_enum_class(node): + continue + values: list[str] = [] + for item in node.body: + if isinstance(item, ast.Assign): + for target in item.targets: + if isinstance(target, ast.Name) and target.id.isupper(): + val = _extract_constant(item.value) + if val is not None: + values.append(val) + enums[node.name] = values + return enums + + +def _is_enum_class(node: ast.ClassDef) -> bool: + """Return True if the class inherits from ``(str, enum.Enum)`` or ``(str, Enum)``.""" + base_names = [_base_name_str(b) for b in node.bases] + return "str" in base_names and any(n in ("enum.Enum", "Enum") for n in base_names) + + +# ── Model-class detection ──────────────────────────────────────────── + + +def _find_model_class(tree: ast.Module, enum_classes: dict[str, list[str]]) -> ast.ClassDef | None: + """Return the first non-enum ClassDef in the module (the model).""" + for node in ast.iter_child_nodes(tree): + if isinstance(node, ast.ClassDef) and node.name not in enum_classes: + return node + return None + + +def _extract_base_name(node: ast.ClassDef) -> str: + """Return the first base class name as a string.""" + if node.bases: + return _base_name_str(node.bases[0]) + return "" + + +def _base_name_str(node: ast.expr) -> str: + """Convert a base-class AST node to a dotted string.""" + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + return f"{_base_name_str(node.value)}.{node.attr}" + return "" + + +def _extract_tablename(node: ast.ClassDef) -> str: + """Extract the ``__tablename__`` string from the class body.""" + for item in node.body: + if isinstance(item, ast.Assign): + for target in item.targets: + if isinstance(target, ast.Name) and target.id == "__tablename__": + val = _extract_constant(item.value) + if val is not None: + return val + return "" + + +# ── Annotation helpers ─────────────────────────────────────────────── + + +def _resolve_annotation_type(annotation: ast.expr, enum_classes: dict[str, list[str]]) -> str: + """Map a ``Mapped[X]`` annotation to a YAML type string.""" + inner = _unwrap_mapped(annotation) + if inner is None: + return "string" # fallback + + # Strip Optional / nullable union (X | None). + inner = _strip_none_union(inner) + + # list[...] → "array" + if isinstance(inner, ast.Subscript) and _name_of(inner.value) == "list": + return "array" + + type_str = _name_of(inner) + + # Check if the annotation refers to a known enum class. + if type_str in enum_classes: + return "enum" + + return ANNOTATION_TO_YAML_TYPE.get(type_str, "string") + + +def _annotation_is_nullable(annotation: ast.expr) -> bool: + """Return True if the annotation contains ``| None`` or ``Optional[...]``.""" + inner = _unwrap_mapped(annotation) + if inner is None: + return False + return _is_none_union(inner) + + +def _unwrap_mapped(annotation: ast.expr) -> ast.expr | None: + """If ``Mapped[X]``, return ``X``. Otherwise return ``None``.""" + if isinstance(annotation, ast.Subscript) and _name_of(annotation.value) == "Mapped": + return annotation.slice + return None + + +def _strip_none_union(node: ast.expr) -> ast.expr: + """Given ``X | None``, return ``X``. Otherwise return node unchanged.""" + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): + if _is_none(node.right): + return node.left + if _is_none(node.left): + return node.right + return node + + +def _is_none_union(node: ast.expr) -> bool: + """Return True if the node is a union containing None (``X | None``).""" + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): + return _is_none(node.right) or _is_none(node.left) + return False + + +def _is_none(node: ast.expr) -> bool: + """Return True if the node represents ``None``.""" + return isinstance(node, ast.Constant) and node.value is None + + +def _extract_enum_class_name(annotation: ast.expr) -> str | None: + """Extract the enum class name from ``Mapped[EnumName]``.""" + inner = _unwrap_mapped(annotation) + if inner is None: + return None + inner = _strip_none_union(inner) + name = _name_of(inner) + return name if name else None + + +def _name_of(node: ast.expr) -> str: + """Return a simple or dotted name from a Name or Attribute node.""" + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + return f"{_name_of(node.value)}.{node.attr}" + return "" + + +# ── mapped_column() parsing ────────────────────────────────────────── + + +def _is_mapped_column_call(node: ast.expr | None) -> bool: + """Return True if the node is a ``mapped_column(...)`` call.""" + return isinstance(node, ast.Call) and _name_of(node.func) == "mapped_column" + + +def _parse_mapped_column(call: ast.Call) -> dict: + """Extract structured info from a ``mapped_column(...)`` call. + + Returns a dict with optional keys: unique, default, foreign_key. + """ + info: dict = {} + + # Scan positional args for ForeignKey("table.col"). + for arg in call.args: + fk = _extract_foreign_key(arg) + if fk is not None: + info["foreign_key"] = fk + + # Scan keyword args. + for kw in call.keywords: + if kw.arg == "unique" and isinstance(kw.value, ast.Constant): + info["unique"] = kw.value.value + elif kw.arg == "default": + info["default"] = _extract_default(kw.value) + elif kw.arg == "nullable" and isinstance(kw.value, ast.Constant): + # We already infer nullable from annotation; this is a backup. + pass + + return info + + +def _extract_foreign_key(node: ast.expr) -> str | None: + """If the node is ``ForeignKey("table.col")``, return the string arg.""" + if isinstance(node, ast.Call) and _name_of(node.func) == "ForeignKey": + if node.args and isinstance(node.args[0], ast.Constant): + return str(node.args[0].value) + return None + + +def _extract_default(node: ast.expr) -> str | None: + """Best-effort extraction of a default value as a string.""" + if isinstance(node, ast.Constant): + return str(node.value) + # Handle Enum member access like PostStatus.DRAFT. + if isinstance(node, ast.Attribute): + return f"{_name_of(node.value)}.{node.attr}" + # Handle simple names like True, False. + if isinstance(node, ast.Name): + return node.id + return None + + +# ── relationship() parsing ─────────────────────────────────────────── + + +def _is_relationship_call(node: ast.expr | None) -> bool: + """Return True if the node is a ``relationship(...)`` call.""" + return isinstance(node, ast.Call) and _name_of(node.func) == "relationship" + + +def _parse_relationship( + field_name: str, + annotation: ast.expr, + call: ast.Call, + model_name: str, +) -> RelationshipDefinition | None: + """Parse a ``relationship()`` call into a RelationshipDefinition.""" + target = _resolve_relationship_target(annotation) + if target is None: + return None + + back_populates: str | None = None + for kw in call.keywords: + if kw.arg == "back_populates" and isinstance(kw.value, ast.Constant): + back_populates = str(kw.value.value) + + # Determine relationship kind from annotation shape. + rel_type = _infer_relationship_type(annotation, target, model_name) + + return RelationshipDefinition( + field_name=field_name, + type=rel_type, + target_entity=target, + back_populates=back_populates, + ) + + +def _resolve_relationship_target(annotation: ast.expr) -> str | None: + """Extract the target entity name from a relationship annotation. + + Handles: + - ``Mapped["User"]`` → ``"User"`` + - ``Mapped[list["Post"]]`` → ``"Post"`` + """ + inner = _unwrap_mapped(annotation) + if inner is None: + return None + + # Mapped["User"] — string constant (forward ref). + if isinstance(inner, ast.Constant) and isinstance(inner.value, str): + return inner.value + + # Mapped[list["Post"]] — subscript with list. + if isinstance(inner, ast.Subscript) and _name_of(inner.value) == "list": + item = inner.slice + if isinstance(item, ast.Constant) and isinstance(item.value, str): + return item.value + + # Mapped[User] — bare name (non-forward-ref). + name = _name_of(inner) + if name: + return name + + return None + + +def _infer_relationship_type(annotation: ast.expr, target: str, model_name: str) -> str: + """Infer the relationship type from the annotation shape. + + - ``Mapped["X"]`` → ``"many_to_one"`` + - ``Mapped[list["X"]]`` → ``"one_to_many"`` + - self-referential → ``"self_referential"`` + """ + inner = _unwrap_mapped(annotation) + if inner is not None and isinstance(inner, ast.Subscript) and _name_of(inner.value) == "list": + return "one_to_many" + if target == model_name: + return "self_referential" + return "many_to_one" + + +# ── FK → entity name resolution ───────────────────────────────────── + + +def _fk_to_entity_name(fk_string: str | None, model_name: str, table_name: str) -> str: + """Convert ``"users.id"`` → ``"User"`` (singular, capitalised). + + For self-referential FKs (table matches own table), returns the model name. + """ + if fk_string is None: + return "" + table, _, _col = fk_string.partition(".") + if table == table_name: + return model_name + # Naive singularisation: strip trailing "s" and title-case. + # This matches FastStack's generated table names (lower-case plural). + singular = table.rstrip("s") if table.endswith("s") else table + return singular.title() + + +def _extract_constant(node: ast.expr) -> str | None: + """Return the string value of a Constant node, or None.""" + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + return None diff --git a/cli/yaml_parser.py b/cli/yaml_parser.py new file mode 100644 index 0000000..16908c5 --- /dev/null +++ b/cli/yaml_parser.py @@ -0,0 +1,214 @@ +"""Parse ``entities.yaml`` files into structured :class:`EntityDefinition` objects. + +Handles relationship resolution, table-name pluralization, and validation +of cross-entity references. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + +import inflect +import yaml + +# Shared inflect engine +_inflect_engine = inflect.engine() + + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +@dataclass +class FieldDefinition: + """A single field within an entity.""" + + name: str + type: str # YAML type (string, uuid, enum, etc.) + required: bool = False + unique: bool = False + default: str | None = None + references: str | None = None # FK target entity name, or "self" + on_delete: str = "SET NULL" # CASCADE, SET NULL, RESTRICT + enum_values: list[str] = field(default_factory=list) # for enum type + items: str | None = None # for array type -- inner type name + + +@dataclass +class RelationshipDefinition: + """A resolved relationship between entities.""" + + field_name: str # e.g. "user_id" or "tags" + type: str # "many_to_one", "many_to_many", "self_referential" + target_entity: str # e.g. "User" or "self" + back_populates: str | None = None # e.g. "posts" + + +@dataclass +class EntityDefinition: + """Fully parsed entity definition from YAML.""" + + name: str + base: str = "FullAuditedEntity" + fields: list[FieldDefinition] = field(default_factory=list) + relationships: list[RelationshipDefinition] = field(default_factory=list) + searchable: list[str] = field(default_factory=list) + table_name: str = "" # auto-generated pluralized name + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _pluralize(name: str) -> str: + """Return a lowercase, pluralized table name for *name*. + + Uses *inflect* for proper English pluralization. + """ + # Convert CamelCase to snake_case first + snake = _camel_to_snake(name) + plural = _inflect_engine.plural_noun(snake) + # inflect returns False if the word is already plural + return plural if plural else snake + + +def _camel_to_snake(name: str) -> str: + """Convert ``CamelCase`` to ``snake_case``.""" + import re + + s1 = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", name) + return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", s1).lower() + + +def _resolve_back_populates( + source_entity: str, + rel_type: str, +) -> str: + """Generate the ``back_populates`` name for the *other* side of a relationship. + + For a many_to_one from Post -> User, the User side gets ``back_populates="posts"``. + For self-referential, returns ``"children"``. + """ + if rel_type == "self_referential": + return "children" + # Pluralize the source entity name for the reverse side + return _pluralize(source_entity) + + +# --------------------------------------------------------------------------- +# Main parser +# --------------------------------------------------------------------------- + + +def parse_entities_yaml(path: Path) -> list[EntityDefinition]: + """Parse an ``entities.yaml`` file into :class:`EntityDefinition` objects. + + Resolves relationships: + - uuid field with ``references: EntityName`` -> many_to_one relationship + - many_to_many field with ``references: EntityName`` -> many_to_many relationship + - uuid field with ``references: self`` -> self_referential relationship + + Generates ``back_populates`` names using *inflect* for pluralization. + Validates that referenced entities exist in the YAML. + + Parameters + ---------- + path: + Path to the YAML file. + + Returns + ------- + list[EntityDefinition] + Parsed entity definitions with resolved relationships. + + Raises + ------ + FileNotFoundError + If *path* does not exist. + ValueError + If the YAML is structurally invalid or references unknown entities. + """ + if not path.exists(): + raise FileNotFoundError(f"entities.yaml not found: {path}") + + raw = yaml.safe_load(path.read_text(encoding="utf-8")) + + if not isinstance(raw, dict) or "entities" not in raw: + raise ValueError("YAML must contain a top-level 'entities' key") + + raw_entities: dict[str, dict] = raw["entities"] + entity_names = set(raw_entities.keys()) + + # First pass: build EntityDefinition objects with fields + entities: list[EntityDefinition] = [] + for entity_name, entity_data in raw_entities.items(): + entity_data = entity_data or {} + entity = EntityDefinition( + name=entity_name, + base=entity_data.get("base", "FullAuditedEntity"), + table_name=_pluralize(entity_name), + searchable=list(entity_data.get("searchable", [])), + ) + + raw_fields: dict[str, dict] = entity_data.get("fields", {}) + for field_name, field_data in raw_fields.items(): + field_data = field_data or {} + fd = FieldDefinition( + name=field_name, + type=field_data.get("type", "string"), + required=bool(field_data.get("required", False)), + unique=bool(field_data.get("unique", False)), + default=field_data.get("default"), + references=field_data.get("references"), + on_delete=field_data.get("on_delete", "SET NULL"), + enum_values=list(field_data.get("values", [])), + items=field_data.get("items"), + ) + entity.fields.append(fd) + + entities.append(entity) + + # Second pass: resolve relationships and validate references + for entity in entities: + for fd in entity.fields: + if fd.references is None: + continue + + # Validate reference target + if fd.references != "self" and fd.references not in entity_names: + raise ValueError( + f"Entity '{entity.name}' field '{fd.name}' references " + f"unknown entity '{fd.references}'. " + f"Known entities: {', '.join(sorted(entity_names))}" + ) + + # Determine relationship type + if fd.references == "self": + rel = RelationshipDefinition( + field_name=fd.name, + type="self_referential", + target_entity=entity.name, + back_populates=_resolve_back_populates(entity.name, "self_referential"), + ) + elif fd.type == "uuid": + rel = RelationshipDefinition( + field_name=fd.name, + type="many_to_one", + target_entity=fd.references, + back_populates=_resolve_back_populates(entity.name, "many_to_one"), + ) + else: + # Treat other typed references as many_to_many + rel = RelationshipDefinition( + field_name=fd.name, + type="many_to_many", + target_entity=fd.references, + back_populates=_resolve_back_populates(entity.name, "many_to_many"), + ) + + entity.relationships.append(rel) + + return entities diff --git a/pyproject.toml b/pyproject.toml index 251c210..de42511 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ jinja2 = ">=3.1" inflect = ">=7.0" uvicorn = ">=0.34.0" python-json-logger = ">=2.0" +pyyaml = "^6.0.3" [tool.poetry.group.dev.dependencies] pytest = ">=8.0" @@ -35,6 +36,7 @@ black = ">=24.0" aiosqlite = ">=0.20.0" httpx = ">=0.28.0" mypy = ">=1.13.0" +types-pyyaml = "^6.0.12.20250915" [build-system] requires = ["poetry-core"] @@ -67,6 +69,8 @@ ignore = [ [tool.ruff.lint.per-file-ignores] "tests/*" = ["C901"] # allow complex test helper functions +"cli/model_introspector.py" = ["C901"] # AST parsing is inherently complex +"cli/yaml_parser.py" = ["C901"] # YAML parsing with relationship resolution [tool.black] line-length = 100 diff --git a/tests/test_cli/test_field_mappings.py b/tests/test_cli/test_field_mappings.py new file mode 100644 index 0000000..9787e60 --- /dev/null +++ b/tests/test_cli/test_field_mappings.py @@ -0,0 +1,273 @@ +"""Tests for cli.field_mappings — type mapping from YAML to SQLAlchemy / Pydantic / Python.""" + +from __future__ import annotations + +import pytest + +from cli.field_mappings import ( + ALL_YAML_TYPES, + PYDANTIC_TYPE_MAP, + PYTHON_TYPE_MAP, + SQLALCHEMY_TYPE_MAP, + get_pydantic_type, + get_python_type, + get_sqlalchemy_imports, + get_sqlalchemy_type, +) + +# --------------------------------------------------------------------------- +# All 13 supported YAML types +# --------------------------------------------------------------------------- +ALL_13_TYPES = [ + "string", + "text", + "integer", + "float", + "boolean", + "datetime", + "date", + "uuid", + "decimal", + "json", + "enum", + "array", + "jsonb", +] + + +class TestTypeMapsCompleteness: + """Verify every type map covers all 13 types.""" + + def test_all_yaml_types_constant(self): + assert len(ALL_YAML_TYPES) == 13 + assert set(ALL_13_TYPES) == ALL_YAML_TYPES + + def test_sqlalchemy_map_covers_all_types(self): + for t in ALL_13_TYPES: + assert t in SQLALCHEMY_TYPE_MAP, f"Missing SQLAlchemy mapping for {t}" + + def test_pydantic_map_covers_all_types(self): + for t in ALL_13_TYPES: + assert t in PYDANTIC_TYPE_MAP, f"Missing Pydantic mapping for {t}" + + def test_python_map_covers_all_types(self): + for t in ALL_13_TYPES: + assert t in PYTHON_TYPE_MAP, f"Missing Python mapping for {t}" + + +# --------------------------------------------------------------------------- +# SQLAlchemy type map values +# --------------------------------------------------------------------------- + + +class TestSQLAlchemyTypeMap: + @pytest.mark.parametrize( + "yaml_type, expected", + [ + ("string", "String(255)"), + ("text", "Text"), + ("integer", "Integer"), + ("float", "Float"), + ("boolean", "Boolean"), + ("datetime", "DateTime"), + ("date", "Date"), + ("uuid", "UUID(as_uuid=True)"), + ("decimal", "Numeric(10, 2)"), + ("json", "JSON"), + ("enum", "Enum"), + ("array", "ARRAY"), + ("jsonb", "JSONB"), + ], + ) + def test_sqlalchemy_map_values(self, yaml_type, expected): + assert SQLALCHEMY_TYPE_MAP[yaml_type] == expected + + +# --------------------------------------------------------------------------- +# Pydantic type map values +# --------------------------------------------------------------------------- + + +class TestPydanticTypeMap: + @pytest.mark.parametrize( + "yaml_type, expected", + [ + ("string", "str"), + ("text", "str"), + ("integer", "int"), + ("float", "float"), + ("boolean", "bool"), + ("datetime", "datetime"), + ("date", "date"), + ("uuid", "UUID"), + ("decimal", "Decimal"), + ("json", "dict"), + ("enum", "str"), + ("array", "list"), + ("jsonb", "dict"), + ], + ) + def test_pydantic_map_values(self, yaml_type, expected): + assert PYDANTIC_TYPE_MAP[yaml_type] == expected + + +# --------------------------------------------------------------------------- +# Python type map values +# --------------------------------------------------------------------------- + + +class TestPythonTypeMap: + @pytest.mark.parametrize( + "yaml_type, expected", + [ + ("string", "str"), + ("text", "str"), + ("integer", "int"), + ("float", "float"), + ("boolean", "bool"), + ("datetime", "datetime"), + ("date", "date"), + ("uuid", "uuid.UUID"), + ("decimal", "Decimal"), + ("json", "dict"), + ("enum", "str"), + ("array", "list"), + ("jsonb", "dict"), + ], + ) + def test_python_map_values(self, yaml_type, expected): + assert PYTHON_TYPE_MAP[yaml_type] == expected + + +# --------------------------------------------------------------------------- +# get_sqlalchemy_type helper +# --------------------------------------------------------------------------- + + +class TestGetSQLAlchemyType: + def test_simple_types(self): + assert get_sqlalchemy_type("string") == "String(255)" + assert get_sqlalchemy_type("integer") == "Integer" + assert get_sqlalchemy_type("uuid") == "UUID(as_uuid=True)" + assert get_sqlalchemy_type("jsonb") == "JSONB" + + def test_enum_with_class(self): + result = get_sqlalchemy_type("enum", enum_class="StatusEnum") + assert result == "Enum(StatusEnum)" + + def test_enum_without_class_raises(self): + with pytest.raises(ValueError, match="enum_class"): + get_sqlalchemy_type("enum") + + def test_array_with_items(self): + result = get_sqlalchemy_type("array", items="string") + assert result == "ARRAY(String)" + + def test_array_with_integer_items(self): + result = get_sqlalchemy_type("array", items="integer") + assert result == "ARRAY(Integer)" + + def test_array_without_items_raises(self): + with pytest.raises(ValueError, match="items"): + get_sqlalchemy_type("array") + + def test_array_with_unknown_inner_type_raises(self): + with pytest.raises(ValueError, match="Unknown inner type"): + get_sqlalchemy_type("array", items="foobar") + + def test_unknown_type_raises(self): + with pytest.raises(ValueError, match="Unknown YAML field type"): + get_sqlalchemy_type("nonexistent") + + +# --------------------------------------------------------------------------- +# get_pydantic_type helper +# --------------------------------------------------------------------------- + + +class TestGetPydanticType: + def test_simple_types(self): + assert get_pydantic_type("string") == "str" + assert get_pydantic_type("integer") == "int" + assert get_pydantic_type("uuid") == "UUID" + assert get_pydantic_type("decimal") == "Decimal" + + def test_enum_with_values(self): + result = get_pydantic_type("enum", values=["admin", "editor", "viewer"]) + assert result == 'Literal["admin", "editor", "viewer"]' + + def test_enum_without_values_returns_str(self): + result = get_pydantic_type("enum") + assert result == "str" + + def test_array_with_items(self): + result = get_pydantic_type("array", items="string") + assert result == "list[str]" + + def test_array_with_integer_items(self): + result = get_pydantic_type("array", items="integer") + assert result == "list[int]" + + def test_array_without_items_returns_list(self): + result = get_pydantic_type("array") + assert result == "list" + + def test_unknown_type_raises(self): + with pytest.raises(ValueError, match="Unknown YAML field type"): + get_pydantic_type("nonexistent") + + +# --------------------------------------------------------------------------- +# get_python_type helper +# --------------------------------------------------------------------------- + + +class TestGetPythonType: + def test_simple_types(self): + assert get_python_type("string") == "str" + assert get_python_type("uuid") == "uuid.UUID" + assert get_python_type("decimal") == "Decimal" + + def test_unknown_type_raises(self): + with pytest.raises(ValueError, match="Unknown YAML field type"): + get_python_type("nonexistent") + + +# --------------------------------------------------------------------------- +# get_sqlalchemy_imports helper +# --------------------------------------------------------------------------- + + +class TestGetSQLAlchemyImports: + @pytest.mark.parametrize( + "yaml_type, expected_imports", + [ + ("string", ["String"]), + ("text", ["Text"]), + ("integer", ["Integer"]), + ("float", ["Float"]), + ("boolean", ["Boolean"]), + ("datetime", ["DateTime"]), + ("date", ["Date"]), + ("uuid", ["UUID"]), + ("decimal", ["Numeric"]), + ("json", ["JSON"]), + ("enum", ["Enum"]), + ("array", ["ARRAY"]), + ("jsonb", ["JSONB"]), + ], + ) + def test_imports_for_each_type(self, yaml_type, expected_imports): + result = get_sqlalchemy_imports(yaml_type) + assert result == expected_imports + + def test_unknown_type_raises(self): + with pytest.raises(ValueError, match="Unknown YAML field type"): + get_sqlalchemy_imports("nonexistent") + + def test_returns_new_list_each_call(self): + """Ensure we get a copy, not the internal list.""" + a = get_sqlalchemy_imports("string") + b = get_sqlalchemy_imports("string") + assert a == b + assert a is not b diff --git a/tests/test_cli/test_model_introspector.py b/tests/test_cli/test_model_introspector.py new file mode 100644 index 0000000..2ac3dfb --- /dev/null +++ b/tests/test_cli/test_model_introspector.py @@ -0,0 +1,355 @@ +"""Tests for cli.model_introspector — AST-based SQLAlchemy model reader. + +Each test writes a model string to a temp file, introspects it, and verifies +the resulting EntityDefinition matches expectations. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from cli.model_introspector import introspect_model + +# ── Fixtures: model source strings ─────────────────────────────────── + +MODEL_SIMPLE = """\ +import uuid +from datetime import datetime +from sqlalchemy import String, Text, Boolean +from sqlalchemy.orm import Mapped, mapped_column +from faststack_core.base.entity import FullAuditedEntity + +class User(FullAuditedEntity): + __tablename__ = "users" + email: Mapped[str] = mapped_column(String(255), unique=True) + name: Mapped[str] = mapped_column(String(255)) + bio: Mapped[str | None] = mapped_column(Text) + is_active: Mapped[bool] = mapped_column(Boolean, default=True) +""" + +MODEL_WITH_FK = """\ +import uuid +from sqlalchemy import String, ForeignKey +from sqlalchemy.orm import Mapped, mapped_column, relationship +from faststack_core.base.entity import AuditedEntity + +class Post(AuditedEntity): + __tablename__ = "posts" + title: Mapped[str] = mapped_column(String(255)) + user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id")) + user: Mapped["User"] = relationship(back_populates="posts") +""" + +MODEL_WITH_ENUM = """\ +import enum +from sqlalchemy import String +from sqlalchemy.orm import Mapped, mapped_column +from faststack_core.base.entity import Entity + +class PostStatus(str, enum.Enum): + DRAFT = "draft" + PUBLISHED = "published" + ARCHIVED = "archived" + +class Post(Entity): + __tablename__ = "posts" + title: Mapped[str] = mapped_column(String(255)) + status: Mapped[PostStatus] = mapped_column(default=PostStatus.DRAFT) +""" + +MODEL_SELF_REF = """\ +import uuid +from sqlalchemy import String, ForeignKey +from sqlalchemy.orm import Mapped, mapped_column, relationship +from faststack_core.base.entity import Entity + +class Category(Entity): + __tablename__ = "categories" + name: Mapped[str] = mapped_column(String(255)) + parent_id: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("categories.id")) + parent: Mapped["Category"] = relationship(back_populates="children") +""" + +MODEL_ONE_TO_MANY = """\ +import uuid +from sqlalchemy import String +from sqlalchemy.orm import Mapped, mapped_column, relationship +from faststack_core.base.entity import FullAuditedEntity + +class User(FullAuditedEntity): + __tablename__ = "users" + name: Mapped[str] = mapped_column(String(255)) + posts: Mapped[list["Post"]] = relationship(back_populates="user") +""" + +MODEL_MULTIPLE_TYPES = """\ +import uuid +from datetime import datetime, date +from decimal import Decimal +from sqlalchemy import String, Integer, Float, DateTime, Date, Numeric, JSON +from sqlalchemy.orm import Mapped, mapped_column +from faststack_core.base.entity import Entity + +class Product(Entity): + __tablename__ = "products" + name: Mapped[str] = mapped_column(String(255)) + quantity: Mapped[int] = mapped_column(Integer) + price: Mapped[float] = mapped_column(Float) + cost: Mapped[Decimal] = mapped_column(Numeric(10, 2)) + created_at: Mapped[datetime] = mapped_column(DateTime) + launch_date: Mapped[date] = mapped_column(Date) + metadata_: Mapped[dict] = mapped_column(JSON) +""" + + +# ── Helper ─────────────────────────────────────────────────────────── + + +def _write_model(tmp_path: Path, source: str, filename: str = "model.py") -> Path: + """Write *source* to a temp file and return its path.""" + p = tmp_path / filename + p.write_text(source) + return p + + +# ── Tests: simple model ───────────────────────────────────────────── + + +class TestSimpleModel: + """Introspect a model with basic fields (string, text, boolean).""" + + def test_class_name(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + assert entity.name == "User" + + def test_base_class(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + assert entity.base == "FullAuditedEntity" + + def test_table_name(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + assert entity.table_name == "users" + + def test_field_count(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + assert len(entity.fields) == 4 + + def test_email_unique(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + email = next(f for f in entity.fields if f.name == "email") + assert email.unique is True + assert email.type == "string" + assert email.required is True + + def test_name_field(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + name = next(f for f in entity.fields if f.name == "name") + assert name.type == "string" + assert name.required is True + assert name.unique is False + + def test_bio_nullable(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + bio = next(f for f in entity.fields if f.name == "bio") + assert bio.required is False + + def test_is_active_default(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + active = next(f for f in entity.fields if f.name == "is_active") + assert active.type == "boolean" + assert active.default == "True" + + def test_no_relationships(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SIMPLE)) + assert entity.relationships == [] + + +# ── Tests: FK and relationship ─────────────────────────────────────── + + +class TestForeignKeyModel: + """Introspect a model with a FK and explicit relationship().""" + + def test_class_name(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + assert entity.name == "Post" + + def test_base_class(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + assert entity.base == "AuditedEntity" + + def test_user_id_field_type(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + uid = next(f for f in entity.fields if f.name == "user_id") + assert uid.type == "uuid" + + def test_user_id_references(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + uid = next(f for f in entity.fields if f.name == "user_id") + assert uid.references == "User" + + def test_has_relationship(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + assert len(entity.relationships) >= 1 + + def test_relationship_type(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + rel = next(r for r in entity.relationships if r.target_entity == "User") + assert rel.type == "many_to_one" + + def test_relationship_back_populates(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + rel = next(r for r in entity.relationships if r.target_entity == "User") + assert rel.back_populates == "posts" + + def test_not_self_referential(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_FK)) + rel = next(r for r in entity.relationships if r.target_entity == "User") + assert rel.type == "many_to_one" # not self_referential + + +# ── Tests: enum ────────────────────────────────────────────────────── + + +class TestEnumModel: + """Introspect a model that uses a ``str, enum.Enum`` class.""" + + def test_skips_enum_class(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_ENUM)) + assert entity.name == "Post" # not PostStatus + + def test_status_field_type(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_ENUM)) + status = next(f for f in entity.fields if f.name == "status") + assert status.type == "enum" + + def test_enum_values(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_ENUM)) + status = next(f for f in entity.fields if f.name == "status") + assert status.enum_values == ["draft", "published", "archived"] + + def test_enum_default(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_ENUM)) + status = next(f for f in entity.fields if f.name == "status") + assert status.default == "PostStatus.DRAFT" + + def test_field_count(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_WITH_ENUM)) + assert len(entity.fields) == 2 # title + status + + +# ── Tests: self-referential ────────────────────────────────────────── + + +class TestSelfReferentialModel: + """Introspect a model with a self-referential FK.""" + + def test_class_name(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SELF_REF)) + assert entity.name == "Category" + + def test_parent_id_nullable(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SELF_REF)) + pid = next(f for f in entity.fields if f.name == "parent_id") + assert pid.required is False + + def test_self_referential_relationship(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SELF_REF)) + rel = next(r for r in entity.relationships if r.target_entity == "Category") + assert rel.type == "self_referential" + + def test_relationship_back_populates(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SELF_REF)) + rel = next(r for r in entity.relationships if r.field_name == "parent") + assert rel.back_populates == "children" + + def test_parent_id_references_self(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_SELF_REF)) + pid = next(f for f in entity.fields if f.name == "parent_id") + assert pid.references == "Category" + + +# ── Tests: one-to-many ─────────────────────────────────────────────── + + +class TestOneToManyModel: + """Introspect a model with a ``list["Post"]`` relationship.""" + + def test_relationship_type(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_ONE_TO_MANY)) + rel = next(r for r in entity.relationships if r.target_entity == "Post") + assert rel.type == "one_to_many" + + def test_relationship_back_populates(self, tmp_path: Path) -> None: + entity = introspect_model(_write_model(tmp_path, MODEL_ONE_TO_MANY)) + rel = next(r for r in entity.relationships if r.target_entity == "Post") + assert rel.back_populates == "user" + + +# ── Tests: multiple types ──────────────────────────────────────────── + + +class TestMultipleTypes: + """Verify that all supported annotation types map correctly.""" + + @pytest.fixture() + def entity(self, tmp_path: Path): + return introspect_model(_write_model(tmp_path, MODEL_MULTIPLE_TYPES)) + + def test_string_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "name") + assert f.type == "string" + + def test_integer_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "quantity") + assert f.type == "integer" + + def test_float_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "price") + assert f.type == "float" + + def test_decimal_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "cost") + assert f.type == "decimal" + + def test_datetime_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "created_at") + assert f.type == "datetime" + + def test_date_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "launch_date") + assert f.type == "date" + + def test_json_type(self, entity) -> None: + f = next(f for f in entity.fields if f.name == "metadata_") + assert f.type == "json" + + def test_table_name(self, entity) -> None: + assert entity.table_name == "products" + + def test_field_count(self, entity) -> None: + assert len(entity.fields) == 7 + + +# ── Tests: error cases ─────────────────────────────────────────────── + + +class TestErrorCases: + """Edge cases and error handling.""" + + def test_no_model_raises(self, tmp_path: Path) -> None: + source = "x = 1\n" + with pytest.raises(ValueError, match="No model class found"): + introspect_model(_write_model(tmp_path, source)) + + def test_enum_only_raises(self, tmp_path: Path) -> None: + source = """\ +import enum + +class Status(str, enum.Enum): + A = "a" +""" + with pytest.raises(ValueError, match="No model class found"): + introspect_model(_write_model(tmp_path, source)) diff --git a/tests/test_cli/test_yaml_parser.py b/tests/test_cli/test_yaml_parser.py new file mode 100644 index 0000000..fac8f72 --- /dev/null +++ b/tests/test_cli/test_yaml_parser.py @@ -0,0 +1,305 @@ +"""Tests for cli.yaml_parser — YAML entity definition parsing.""" + +from __future__ import annotations + +from pathlib import Path +from textwrap import dedent + +import pytest + +from cli.yaml_parser import ( + EntityDefinition, + FieldDefinition, + RelationshipDefinition, + parse_entities_yaml, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +SAMPLE_YAML = dedent("""\ + entities: + User: + base: FullAuditedEntity + fields: + email: {type: string, unique: true, required: true} + name: {type: string, required: true} + role: {type: enum, values: [admin, editor, viewer], default: '"viewer"'} + bio: {type: text} + searchable: [email, name] + Post: + base: AuditedEntity + fields: + title: {type: string, required: true} + content: {type: text} + status: {type: enum, values: [draft, published, archived], default: '"draft"'} + tags: {type: array, items: string} + metadata: {type: jsonb} + user_id: {type: uuid, references: User} + searchable: [title] + Category: + base: AuditedEntity + fields: + name: {type: string, required: true} + parent_id: {type: uuid, references: self} +""") + + +@pytest.fixture() +def sample_yaml_path(tmp_path: Path) -> Path: + """Write the sample YAML to a temp file and return the path.""" + p = tmp_path / "entities.yaml" + p.write_text(SAMPLE_YAML, encoding="utf-8") + return p + + +@pytest.fixture() +def parsed_entities(sample_yaml_path: Path) -> list[EntityDefinition]: + """Parse the sample YAML and return entity definitions.""" + return parse_entities_yaml(sample_yaml_path) + + +# --------------------------------------------------------------------------- +# Basic structure tests +# --------------------------------------------------------------------------- + + +class TestEntityCounts: + def test_parses_three_entities(self, parsed_entities): + assert len(parsed_entities) == 3 + + def test_entity_names(self, parsed_entities): + names = [e.name for e in parsed_entities] + assert names == ["User", "Post", "Category"] + + +class TestFieldCounts: + def test_user_has_4_fields(self, parsed_entities): + user = parsed_entities[0] + assert user.name == "User" + assert len(user.fields) == 4 + + def test_post_has_6_fields(self, parsed_entities): + post = parsed_entities[1] + assert post.name == "Post" + assert len(post.fields) == 6 + + def test_category_has_2_fields(self, parsed_entities): + category = parsed_entities[2] + assert category.name == "Category" + assert len(category.fields) == 2 + + +# --------------------------------------------------------------------------- +# Base class +# --------------------------------------------------------------------------- + + +class TestBaseClass: + def test_user_base(self, parsed_entities): + assert parsed_entities[0].base == "FullAuditedEntity" + + def test_post_base(self, parsed_entities): + assert parsed_entities[1].base == "AuditedEntity" + + def test_category_base(self, parsed_entities): + assert parsed_entities[2].base == "AuditedEntity" + + +# --------------------------------------------------------------------------- +# Relationships +# --------------------------------------------------------------------------- + + +class TestRelationships: + def test_post_has_many_to_one_to_user(self, parsed_entities): + post = parsed_entities[1] + assert len(post.relationships) == 1 + rel = post.relationships[0] + assert rel.type == "many_to_one" + assert rel.target_entity == "User" + assert rel.field_name == "user_id" + assert rel.back_populates == "posts" + + def test_category_has_self_referential(self, parsed_entities): + category = parsed_entities[2] + assert len(category.relationships) == 1 + rel = category.relationships[0] + assert rel.type == "self_referential" + assert rel.target_entity == "Category" + assert rel.field_name == "parent_id" + assert rel.back_populates == "children" + + def test_user_has_no_relationships(self, parsed_entities): + user = parsed_entities[0] + assert len(user.relationships) == 0 + + +# --------------------------------------------------------------------------- +# Enum fields +# --------------------------------------------------------------------------- + + +class TestEnumFields: + def test_user_role_enum_values(self, parsed_entities): + user = parsed_entities[0] + role_field = next(f for f in user.fields if f.name == "role") + assert role_field.type == "enum" + assert role_field.enum_values == ["admin", "editor", "viewer"] + assert role_field.default == '"viewer"' + + def test_post_status_enum_values(self, parsed_entities): + post = parsed_entities[1] + status_field = next(f for f in post.fields if f.name == "status") + assert status_field.type == "enum" + assert status_field.enum_values == ["draft", "published", "archived"] + assert status_field.default == '"draft"' + + +# --------------------------------------------------------------------------- +# Array fields +# --------------------------------------------------------------------------- + + +class TestArrayFields: + def test_post_tags_has_items(self, parsed_entities): + post = parsed_entities[1] + tags_field = next(f for f in post.fields if f.name == "tags") + assert tags_field.type == "array" + assert tags_field.items == "string" + + +# --------------------------------------------------------------------------- +# Searchable fields +# --------------------------------------------------------------------------- + + +class TestSearchableFields: + def test_user_searchable(self, parsed_entities): + user = parsed_entities[0] + assert user.searchable == ["email", "name"] + + def test_post_searchable(self, parsed_entities): + post = parsed_entities[1] + assert post.searchable == ["title"] + + def test_category_no_searchable(self, parsed_entities): + category = parsed_entities[2] + assert category.searchable == [] + + +# --------------------------------------------------------------------------- +# Table name pluralization +# --------------------------------------------------------------------------- + + +class TestTableNames: + def test_user_table_name(self, parsed_entities): + assert parsed_entities[0].table_name == "users" + + def test_post_table_name(self, parsed_entities): + assert parsed_entities[1].table_name == "posts" + + def test_category_table_name(self, parsed_entities): + assert parsed_entities[2].table_name == "categories" + + +# --------------------------------------------------------------------------- +# Field properties +# --------------------------------------------------------------------------- + + +class TestFieldProperties: + def test_email_required_and_unique(self, parsed_entities): + user = parsed_entities[0] + email = next(f for f in user.fields if f.name == "email") + assert email.required is True + assert email.unique is True + + def test_bio_optional(self, parsed_entities): + user = parsed_entities[0] + bio = next(f for f in user.fields if f.name == "bio") + assert bio.required is False + assert bio.unique is False + + def test_user_id_references(self, parsed_entities): + post = parsed_entities[1] + user_id = next(f for f in post.fields if f.name == "user_id") + assert user_id.type == "uuid" + assert user_id.references == "User" + + def test_parent_id_references_self(self, parsed_entities): + category = parsed_entities[2] + parent_id = next(f for f in category.fields if f.name == "parent_id") + assert parent_id.type == "uuid" + assert parent_id.references == "self" + + def test_metadata_jsonb(self, parsed_entities): + post = parsed_entities[1] + meta = next(f for f in post.fields if f.name == "metadata") + assert meta.type == "jsonb" + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + + +class TestErrorHandling: + def test_unknown_reference_raises_value_error(self, tmp_path): + bad_yaml = dedent("""\ + entities: + Post: + fields: + author_id: {type: uuid, references: Author} + """) + p = tmp_path / "bad.yaml" + p.write_text(bad_yaml, encoding="utf-8") + with pytest.raises(ValueError, match="unknown entity 'Author'"): + parse_entities_yaml(p) + + def test_missing_file_raises_file_not_found(self, tmp_path): + p = tmp_path / "nonexistent.yaml" + with pytest.raises(FileNotFoundError): + parse_entities_yaml(p) + + def test_missing_entities_key_raises_value_error(self, tmp_path): + bad_yaml = "some_key: value\n" + p = tmp_path / "bad.yaml" + p.write_text(bad_yaml, encoding="utf-8") + with pytest.raises(ValueError, match="entities"): + parse_entities_yaml(p) + + +# --------------------------------------------------------------------------- +# Dataclass defaults +# --------------------------------------------------------------------------- + + +class TestDataclassDefaults: + def test_field_definition_defaults(self): + fd = FieldDefinition(name="test", type="string") + assert fd.required is False + assert fd.unique is False + assert fd.default is None + assert fd.references is None + assert fd.on_delete == "SET NULL" + assert fd.enum_values == [] + assert fd.items is None + + def test_relationship_definition_defaults(self): + rd = RelationshipDefinition( + field_name="user_id", + type="many_to_one", + target_entity="User", + ) + assert rd.back_populates is None + + def test_entity_definition_defaults(self): + ed = EntityDefinition(name="Test") + assert ed.base == "FullAuditedEntity" + assert ed.fields == [] + assert ed.relationships == [] + assert ed.searchable == [] + assert ed.table_name == ""