371 lines
11 KiB
Python
371 lines
11 KiB
Python
"""
|
|
Model Generator
|
|
|
|
Generic model generation from configuration files.
|
|
Supports multiple output formats and is extensible for bidirectional conversion.
|
|
|
|
Output formats:
|
|
- pydantic: Pydantic BaseModel classes
|
|
- django: Django ORM models (planned)
|
|
- prisma: Prisma schema (planned)
|
|
- sqlalchemy: SQLAlchemy models (planned)
|
|
|
|
Future: Extract models FROM existing codebases (reverse direction)
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from pathlib import Path
|
|
from typing import Dict, Type
|
|
|
|
from .config_loader import ConfigLoader
|
|
|
|
|
|
class BaseModelWriter(ABC):
|
|
"""Abstract base for model output writers."""
|
|
|
|
@abstractmethod
|
|
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
|
"""Write models to the specified path."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def file_extension(self) -> str:
|
|
"""Return the file extension for this format."""
|
|
pass
|
|
|
|
|
|
class PydanticWriter(BaseModelWriter):
|
|
"""Generates Pydantic model files."""
|
|
|
|
def file_extension(self) -> str:
|
|
return ".py"
|
|
|
|
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
|
"""Write Pydantic models to output_path."""
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
content = self._generate_content(config)
|
|
output_path.write_text(content)
|
|
|
|
def _generate_content(self, config: ConfigLoader) -> str:
|
|
"""Generate the Pydantic models file content."""
|
|
|
|
# Get component names from config
|
|
config_comp = config.get_shared_component("config")
|
|
data_comp = config.get_shared_component("data")
|
|
|
|
data_flow_sys = config.get_system("data_flow")
|
|
doc_sys = config.get_system("documentation")
|
|
exec_sys = config.get_system("execution")
|
|
|
|
connector_comp = config.get_component("data_flow", "connector")
|
|
pulse_comp = config.get_component("data_flow", "composed")
|
|
|
|
pattern_comp = config.get_component("documentation", "pattern")
|
|
doc_composed = config.get_component("documentation", "composed")
|
|
|
|
tool_comp = config.get_component("execution", "utility")
|
|
monitor_comp = config.get_component("execution", "watcher")
|
|
cabinet_comp = config.get_component("execution", "container")
|
|
exec_composed = config.get_component("execution", "composed")
|
|
|
|
return f'''"""
|
|
Pydantic models - Generated from {config.framework.name}.config.json
|
|
|
|
DO NOT EDIT MANUALLY - Regenerate from config
|
|
"""
|
|
|
|
from enum import Enum
|
|
from typing import List, Literal, Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class Status(str, Enum):
|
|
PENDING = "pending"
|
|
PLANNED = "planned"
|
|
BUILDING = "building"
|
|
DEV = "dev"
|
|
LIVE = "live"
|
|
READY = "ready"
|
|
|
|
|
|
class System(str, Enum):
|
|
{data_flow_sys.name.upper()} = "{data_flow_sys.name}"
|
|
{doc_sys.name.upper()} = "{doc_sys.name}"
|
|
{exec_sys.name.upper()} = "{exec_sys.name}"
|
|
|
|
|
|
class ToolType(str, Enum):
|
|
APP = "app"
|
|
CLI = "cli"
|
|
|
|
|
|
# === Shared Components ===
|
|
|
|
|
|
class {config_comp.title}(BaseModel):
|
|
"""{config_comp.description}. Shared across {data_flow_sys.name}, {exec_sys.name}."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
config_path: Optional[str] = None
|
|
|
|
|
|
class {data_comp.title}(BaseModel):
|
|
"""{data_comp.description}. Shared across all systems."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
source_template: Optional[str] = None
|
|
data_path: Optional[str] = None
|
|
|
|
|
|
# === System-Specific Components ===
|
|
|
|
|
|
class {connector_comp.title}(BaseModel):
|
|
"""{connector_comp.description} ({data_flow_sys.name})."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
system: Literal["{data_flow_sys.name}"] = "{data_flow_sys.name}"
|
|
mock: Optional[bool] = None
|
|
description: Optional[str] = None
|
|
|
|
|
|
class {pattern_comp.title}(BaseModel):
|
|
"""{pattern_comp.description} ({doc_sys.name})."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
template_path: Optional[str] = None
|
|
system: Literal["{doc_sys.name}"] = "{doc_sys.name}"
|
|
|
|
|
|
class {tool_comp.title}(BaseModel):
|
|
"""{tool_comp.description} ({exec_sys.name})."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
|
type: Optional[ToolType] = None
|
|
description: Optional[str] = None
|
|
path: Optional[str] = None
|
|
url: Optional[str] = None
|
|
cli: Optional[str] = None
|
|
|
|
|
|
class {monitor_comp.title}(BaseModel):
|
|
"""{monitor_comp.description} ({exec_sys.name})."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
|
|
|
|
|
class {cabinet_comp.title}(BaseModel):
|
|
"""{cabinet_comp.description} ({exec_sys.name})."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
tools: List[{tool_comp.title}] = Field(default_factory=list)
|
|
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
|
|
|
|
|
# === Composed Types ===
|
|
|
|
|
|
class {pulse_comp.title}(BaseModel):
|
|
"""{pulse_comp.description} ({data_flow_sys.name}). Formula: {pulse_comp.formula}."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
{connector_comp.name}: Optional[{connector_comp.title}] = None
|
|
{config_comp.name}: Optional[{config_comp.title}] = None
|
|
{data_comp.name}: Optional[{data_comp.title}] = None
|
|
system: Literal["{data_flow_sys.name}"] = "{data_flow_sys.name}"
|
|
|
|
|
|
class {doc_composed.title}(BaseModel):
|
|
"""{doc_composed.description} ({doc_sys.name}). Formula: {doc_composed.formula}."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
template: Optional[{pattern_comp.title}] = None
|
|
{data_comp.name}: Optional[{data_comp.title}] = None
|
|
output_{data_comp.name}: Optional[{data_comp.title}] = None
|
|
system: Literal["{doc_sys.name}"] = "{doc_sys.name}"
|
|
|
|
|
|
class {exec_composed.title}(BaseModel):
|
|
"""{exec_composed.description} ({exec_sys.name}). Formula: {exec_composed.formula}."""
|
|
|
|
name: str # Unique identifier
|
|
slug: str # URL-friendly identifier
|
|
title: str # Display title for UI
|
|
status: Optional[Status] = None
|
|
cabinet: Optional[{cabinet_comp.title}] = None
|
|
{config_comp.name}: Optional[{config_comp.title}] = None
|
|
{data_comp.plural}: List[{data_comp.title}] = Field(default_factory=list)
|
|
system: Literal["{exec_sys.name}"] = "{exec_sys.name}"
|
|
|
|
|
|
# === Collection wrappers for JSON files ===
|
|
|
|
|
|
class {config_comp.title}Collection(BaseModel):
|
|
items: List[{config_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {data_comp.title}Collection(BaseModel):
|
|
items: List[{data_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {connector_comp.title}Collection(BaseModel):
|
|
items: List[{connector_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {pattern_comp.title}Collection(BaseModel):
|
|
items: List[{pattern_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {tool_comp.title}Collection(BaseModel):
|
|
items: List[{tool_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {monitor_comp.title}Collection(BaseModel):
|
|
items: List[{monitor_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {cabinet_comp.title}Collection(BaseModel):
|
|
items: List[{cabinet_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {pulse_comp.title}Collection(BaseModel):
|
|
items: List[{pulse_comp.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {doc_composed.title}Collection(BaseModel):
|
|
items: List[{doc_composed.title}] = Field(default_factory=list)
|
|
|
|
|
|
class {exec_composed.title}Collection(BaseModel):
|
|
items: List[{exec_composed.title}] = Field(default_factory=list)
|
|
'''
|
|
|
|
|
|
class DjangoWriter(BaseModelWriter):
|
|
"""Generates Django model files (placeholder)."""
|
|
|
|
def file_extension(self) -> str:
|
|
return ".py"
|
|
|
|
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
|
raise NotImplementedError("Django model generation not yet implemented")
|
|
|
|
|
|
class PrismaWriter(BaseModelWriter):
|
|
"""Generates Prisma schema files (placeholder)."""
|
|
|
|
def file_extension(self) -> str:
|
|
return ".prisma"
|
|
|
|
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
|
raise NotImplementedError("Prisma schema generation not yet implemented")
|
|
|
|
|
|
class SQLAlchemyWriter(BaseModelWriter):
|
|
"""Generates SQLAlchemy model files (placeholder)."""
|
|
|
|
def file_extension(self) -> str:
|
|
return ".py"
|
|
|
|
def write(self, config: ConfigLoader, output_path: Path) -> None:
|
|
raise NotImplementedError("SQLAlchemy model generation not yet implemented")
|
|
|
|
|
|
# Registry of available writers
|
|
WRITERS: Dict[str, Type[BaseModelWriter]] = {
|
|
"pydantic": PydanticWriter,
|
|
"django": DjangoWriter,
|
|
"prisma": PrismaWriter,
|
|
"sqlalchemy": SQLAlchemyWriter,
|
|
}
|
|
|
|
|
|
class ModelGenerator:
|
|
"""
|
|
Generates typed models from configuration.
|
|
|
|
This is the main entry point for model generation.
|
|
Delegates to format-specific writers.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
config: ConfigLoader,
|
|
output_path: Path,
|
|
output_format: str = "pydantic",
|
|
):
|
|
"""
|
|
Initialize the generator.
|
|
|
|
Args:
|
|
config: Loaded configuration
|
|
output_path: Exact path where to write (file or directory depending on format)
|
|
output_format: Output format (pydantic, django, prisma, sqlalchemy)
|
|
"""
|
|
self.config = config
|
|
self.output_path = Path(output_path)
|
|
self.output_format = output_format
|
|
|
|
if output_format not in WRITERS:
|
|
raise ValueError(
|
|
f"Unknown output format: {output_format}. "
|
|
f"Available: {list(WRITERS.keys())}"
|
|
)
|
|
|
|
self.writer = WRITERS[output_format]()
|
|
|
|
def generate(self) -> Path:
|
|
"""
|
|
Generate models to the specified output path.
|
|
|
|
Returns:
|
|
Path to the generated file/directory
|
|
"""
|
|
# Determine output file path
|
|
if self.output_path.suffix:
|
|
# User specified a file path
|
|
output_file = self.output_path
|
|
else:
|
|
# User specified a directory, add default filename
|
|
output_file = self.output_path / f"__init__{self.writer.file_extension()}"
|
|
|
|
self.writer.write(self.config, output_file)
|
|
print(f"Generated {self.output_format} models: {output_file}")
|
|
return output_file
|
|
|
|
@classmethod
|
|
def available_formats(cls) -> list:
|
|
"""Return list of available output formats."""
|
|
return list(WRITERS.keys())
|