263 lines
9.2 KiB
Python
263 lines
9.2 KiB
Python
"""
|
|
Schema Loader
|
|
|
|
Loads Python dataclasses from a schema/ folder.
|
|
Expects the folder to have an __init__.py that exports:
|
|
- DATACLASSES: List of dataclass types to generate
|
|
- ENUMS: List of Enum types to include
|
|
- API_MODELS: (optional) List of API request/response types
|
|
- GRPC_MESSAGES: (optional) List of gRPC message types
|
|
- GRPC_SERVICE: (optional) gRPC service definition dict
|
|
"""
|
|
|
|
import dataclasses as dc
|
|
import importlib.util
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Type, get_type_hints
|
|
|
|
|
|
@dataclass
|
|
class FieldDefinition:
|
|
"""Represents a model field."""
|
|
|
|
name: str
|
|
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
|
|
# True when foreign_key points at many rows rather than one — an array of
|
|
# $ref, a m2m table. graphgen renders the two differently, so losing the
|
|
# distinction would draw every collection as a single edge.
|
|
many: bool = False
|
|
|
|
|
|
@dataclass
|
|
class ModelDefinition:
|
|
"""Represents a model/dataclass."""
|
|
|
|
name: str
|
|
fields: List[FieldDefinition]
|
|
docstring: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class EnumDefinition:
|
|
"""Represents an enum."""
|
|
|
|
name: str
|
|
values: List[tuple[str, str]] # (name, value) pairs
|
|
|
|
|
|
@dataclass
|
|
class GrpcServiceDefinition:
|
|
"""Represents a gRPC service."""
|
|
|
|
package: str
|
|
name: str
|
|
methods: List[Dict[str, Any]]
|
|
|
|
|
|
@dataclass
|
|
class EndpointDefinition:
|
|
"""Represents one HTTP operation on a service.
|
|
|
|
Models describe the shapes a service passes around; endpoints describe how
|
|
it is called. Loaders that read a service contract (OpenAPI) or infer one
|
|
(tabular) emit these alongside the models, and shuntgen turns them into
|
|
routes. Loaders that only see shapes — dataclasses, a Django app — emit
|
|
none, which is why this is not part of the BaseExtractor contract.
|
|
"""
|
|
|
|
method: str # GET, POST, ...
|
|
path: str # /pets/{petId}
|
|
operation_id: Optional[str] = None
|
|
summary: Optional[str] = None
|
|
# "collection" returns/accepts many, "item" one, "action" neither.
|
|
kind: str = "item"
|
|
model: Optional[str] = None # what this operation is about
|
|
request_model: Optional[str] = None
|
|
response_model: Optional[str] = None
|
|
response_is_list: bool = False
|
|
# Set when the collection arrives wrapped — {"items": [...], "total": n}
|
|
# rather than a bare array. Answering a wrapped endpoint with an array is
|
|
# the kind of mismatch a client only discovers at parse time.
|
|
envelope_key: Optional[str] = None
|
|
status: int = 200
|
|
path_params: List[str] = field(default_factory=list)
|
|
example: Any = None # response example carried through from the source
|
|
|
|
|
|
@dataclass
|
|
class DatasetDefinition:
|
|
"""Concrete rows harvested next to a model.
|
|
|
|
Importing a spreadsheet yields both a shape and the data that shaped it.
|
|
Throwing the rows away would mean generating a service that answers with
|
|
invented values when the real ones were right there.
|
|
"""
|
|
|
|
model: str
|
|
rows: List[Dict[str, Any]] = field(default_factory=list)
|
|
source: Optional[str] = None # file the rows came from
|
|
collection: Optional[str] = None # url-facing name, e.g. "customers"
|
|
|
|
|
|
class SchemaLoader:
|
|
"""Loads model definitions from Python dataclasses in schema/ folder."""
|
|
|
|
def __init__(self, schema_path: Path):
|
|
self.schema_path = Path(schema_path)
|
|
self.models: List[ModelDefinition] = []
|
|
self.api_models: List[ModelDefinition] = []
|
|
self.enums: List[EnumDefinition] = []
|
|
self.grpc_messages: List[ModelDefinition] = []
|
|
self.grpc_service: Optional[GrpcServiceDefinition] = None
|
|
|
|
def load(self, include: Optional[set] = None) -> "SchemaLoader":
|
|
"""Load schema definitions from the schema folder.
|
|
|
|
Args:
|
|
include: Set of groups to load (dataclasses, enums, api, grpc).
|
|
None means load all groups.
|
|
"""
|
|
init_path = self.schema_path / "__init__.py"
|
|
|
|
if not init_path.exists():
|
|
raise FileNotFoundError(f"Schema folder must have __init__.py: {init_path}")
|
|
|
|
# Import the schema module
|
|
module = self._import_module(init_path)
|
|
|
|
load_all = include is None
|
|
|
|
# Extract DATACLASSES
|
|
if load_all or "dataclasses" in include:
|
|
dataclasses = getattr(module, "DATACLASSES", [])
|
|
for cls in dataclasses:
|
|
self.models.append(self._parse_dataclass(cls))
|
|
|
|
# Extract API_MODELS (request/response types)
|
|
if load_all or "api" in include:
|
|
api_models = getattr(module, "API_MODELS", [])
|
|
for cls in api_models:
|
|
self.api_models.append(self._parse_dataclass(cls))
|
|
|
|
# Extract ENUMS
|
|
if load_all or "enums" in include:
|
|
enums = getattr(module, "ENUMS", [])
|
|
for enum_cls in enums:
|
|
self.enums.append(self._parse_enum(enum_cls))
|
|
|
|
# Extract VIEWS (view/event projections)
|
|
if load_all or "views" in include:
|
|
views = getattr(module, "VIEWS", [])
|
|
for cls in views:
|
|
self.api_models.append(self._parse_dataclass(cls))
|
|
|
|
# Extract GRPC_MESSAGES (optional)
|
|
if load_all or "grpc" in include:
|
|
grpc_messages = getattr(module, "GRPC_MESSAGES", [])
|
|
for cls in grpc_messages:
|
|
self.grpc_messages.append(self._parse_dataclass(cls))
|
|
|
|
# Extract GRPC_SERVICE (optional)
|
|
if load_all or "grpc" in include:
|
|
grpc_service = getattr(module, "GRPC_SERVICE", None)
|
|
if grpc_service:
|
|
self.grpc_service = GrpcServiceDefinition(
|
|
package=grpc_service.get("package", "service"),
|
|
name=grpc_service.get("name", "Service"),
|
|
methods=grpc_service.get("methods", []),
|
|
)
|
|
|
|
# Generic group loader: any include group not handled above
|
|
# is looked up as UPPER_CASE attribute on the module.
|
|
# e.g. include "detect_views" → module.DETECT_VIEWS
|
|
if include:
|
|
known_groups = {"dataclasses", "enums", "api", "views", "grpc"}
|
|
for group in include - known_groups:
|
|
attr_name = group.upper()
|
|
items = getattr(module, attr_name, [])
|
|
for cls in items:
|
|
if isinstance(cls, type) and dc.is_dataclass(cls):
|
|
self.api_models.append(self._parse_dataclass(cls))
|
|
elif isinstance(cls, type) and issubclass(cls, Enum):
|
|
self.enums.append(self._parse_enum(cls))
|
|
|
|
return self
|
|
|
|
def _import_module(self, path: Path):
|
|
"""Import a Python module from a file path."""
|
|
spec = importlib.util.spec_from_file_location("schema", path)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError(f"Could not load module from {path}")
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules["schema"] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
def _parse_dataclass(self, cls: Type) -> ModelDefinition:
|
|
"""Parse a dataclass into a ModelDefinition."""
|
|
hints = get_type_hints(cls)
|
|
fields_info = {f.name: f for f in dc.fields(cls)}
|
|
|
|
fields = []
|
|
for name, type_hint in hints.items():
|
|
if name.startswith("_"):
|
|
continue
|
|
|
|
field_info = fields_info.get(name)
|
|
default = dc.MISSING
|
|
if field_info:
|
|
if field_info.default is not dc.MISSING:
|
|
default = field_info.default
|
|
elif field_info.default_factory is not dc.MISSING:
|
|
default = field_info.default_factory
|
|
|
|
# Check if optional (Union with None)
|
|
optional = self._is_optional(type_hint)
|
|
|
|
fields.append(
|
|
FieldDefinition(
|
|
name=name,
|
|
type_hint=type_hint,
|
|
default=default,
|
|
optional=optional,
|
|
)
|
|
)
|
|
|
|
return ModelDefinition(
|
|
name=cls.__name__,
|
|
fields=fields,
|
|
docstring=cls.__doc__,
|
|
)
|
|
|
|
def _parse_enum(self, enum_cls: Type[Enum]) -> EnumDefinition:
|
|
"""Parse an Enum into an EnumDefinition."""
|
|
values = [(m.name, m.value) for m in enum_cls]
|
|
return EnumDefinition(name=enum_cls.__name__, values=values)
|
|
|
|
def _is_optional(self, type_hint: Any) -> bool:
|
|
"""Check if a type hint is Optional (Union with None)."""
|
|
from typing import Union, get_args, get_origin
|
|
|
|
origin = get_origin(type_hint)
|
|
if origin is Union:
|
|
args = get_args(type_hint)
|
|
return type(None) in args
|
|
return False
|
|
|
|
|
|
def load_schema(schema_path: str | Path, include: Optional[set] = None) -> SchemaLoader:
|
|
"""Load schema definitions from folder."""
|
|
loader = SchemaLoader(schema_path)
|
|
return loader.load(include=include)
|