606 lines
18 KiB
Python
606 lines
18 KiB
Python
"""
|
|
Modelgen - Generic Model Generation Tool
|
|
|
|
Generates typed models from various sources to various formats.
|
|
|
|
Input sources:
|
|
- from-config: Configuration files (soleprint config.json style)
|
|
- from-schema: Python dataclasses in schema/ folder
|
|
- extract: Existing codebases (Django, SQLAlchemy)
|
|
- from-db: A live database, any SQLAlchemy dialect
|
|
- from-openapi: An OpenAPI 3.x / Swagger 2.0 document
|
|
- from-tabular: A directory of CSV/TSV/ODS spreadsheets
|
|
|
|
Output formats:
|
|
- pydantic: Pydantic BaseModel classes
|
|
- django: Django ORM models
|
|
- typescript: TypeScript interfaces
|
|
- protobuf: Protocol Buffer definitions
|
|
- prisma: Prisma schema
|
|
- schema: graphgen-compatible schema.json
|
|
- datagen: BaseDataGenerator subclass for station's datagen tool
|
|
|
|
Usage:
|
|
python -m soleprint.station.tools.modelgen --help
|
|
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-openapi -s api.yaml -o out/ -t pydantic,schema
|
|
python -m soleprint.station.tools.modelgen from-tabular -s ./sheets -o out/ -t pydantic,datagen
|
|
python -m soleprint.station.tools.modelgen generate --config schema/modelgen.json
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from .generator import GENERATORS
|
|
|
|
|
|
def cmd_from_config(args):
|
|
"""Generate models from a configuration file (soleprint config.json style)."""
|
|
from .loader import load_config
|
|
from .model_generator import ModelGenerator
|
|
|
|
config_path = Path(args.config)
|
|
if not config_path.exists():
|
|
print(f"Error: Config file not found: {config_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
output_path = Path(args.output)
|
|
|
|
print(f"Loading config: {config_path}")
|
|
config = load_config(config_path)
|
|
|
|
print(f"Generating {args.format} models to: {output_path}")
|
|
generator = ModelGenerator(
|
|
config=config,
|
|
output_path=output_path,
|
|
output_format=args.format,
|
|
)
|
|
result_path = generator.generate()
|
|
|
|
print(f"Models generated: {result_path}")
|
|
|
|
|
|
def cmd_from_schema(args):
|
|
"""Generate models from Python dataclasses in schema/ folder."""
|
|
from .loader import load_schema
|
|
from .writer import write_file
|
|
|
|
# Determine schema path
|
|
schema_path = Path(args.schema) if args.schema else Path.cwd() / "schema"
|
|
|
|
if not schema_path.exists():
|
|
print(f"Error: Schema folder not found: {schema_path}", file=sys.stderr)
|
|
print(
|
|
"Create a schema/ folder with Python dataclasses and an __init__.py",
|
|
file=sys.stderr,
|
|
)
|
|
print("that exports DATACLASSES and ENUMS lists.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Parse include groups
|
|
include = None
|
|
if args.include:
|
|
include = {g.strip() for g in args.include.split(",")}
|
|
|
|
print(f"Loading schema: {schema_path}")
|
|
schema = load_schema(schema_path, include=include)
|
|
|
|
loaded = []
|
|
if schema.models:
|
|
loaded.append(f"{len(schema.models)} models")
|
|
if schema.enums:
|
|
loaded.append(f"{len(schema.enums)} enums")
|
|
if schema.api_models:
|
|
loaded.append(f"{len(schema.api_models)} api models")
|
|
if schema.grpc_messages:
|
|
loaded.append(f"{len(schema.grpc_messages)} grpc messages")
|
|
print(f"Found {', '.join(loaded)}")
|
|
|
|
# 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(schema, output_file)
|
|
|
|
print("Done!")
|
|
|
|
|
|
def cmd_extract(args):
|
|
"""Extract models from existing codebase."""
|
|
from .loader.extract import EXTRACTORS
|
|
|
|
source_path = Path(args.source)
|
|
if not source_path.exists():
|
|
print(f"Error: Source path not found: {source_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Auto-detect or use specified framework
|
|
framework = args.framework
|
|
extractor = None
|
|
|
|
if framework == "auto":
|
|
for name, extractor_cls in EXTRACTORS.items():
|
|
ext = extractor_cls(source_path)
|
|
if ext.detect():
|
|
framework = name
|
|
extractor = ext
|
|
print(f"Detected framework: {framework}")
|
|
break
|
|
|
|
if not extractor:
|
|
print("Error: Could not auto-detect framework", file=sys.stderr)
|
|
print(f"Available frameworks: {list(EXTRACTORS.keys())}", file=sys.stderr)
|
|
sys.exit(1)
|
|
else:
|
|
if framework not in EXTRACTORS:
|
|
print(f"Error: Unknown framework: {framework}", file=sys.stderr)
|
|
print(f"Available: {list(EXTRACTORS.keys())}", file=sys.stderr)
|
|
sys.exit(1)
|
|
extractor = EXTRACTORS[framework](source_path)
|
|
|
|
print(f"Extracting from: {source_path}")
|
|
models, enums = extractor.extract()
|
|
|
|
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_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 _emit(payload, targets_arg: str, output: str) -> None:
|
|
"""Run one extraction result through every requested target.
|
|
|
|
The three older commands each carry their own copy of this loop; the two
|
|
below share it, because they also have to pass datasets through and a
|
|
fourth copy would be a fourth place to keep in step.
|
|
"""
|
|
targets = [t.strip() for t in targets_arg.split(",") if t.strip()]
|
|
output_dir = Path(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 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(payload, output_file)
|
|
|
|
print("Done!")
|
|
|
|
|
|
def cmd_from_openapi(args):
|
|
"""Generate models from an OpenAPI 3.x / Swagger 2.0 document."""
|
|
from .loader.extract.openapi import OpenAPIExtractor
|
|
|
|
spec_path = Path(args.spec)
|
|
if not spec_path.exists():
|
|
print(f"Error: Spec not found: {spec_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
extractor = OpenAPIExtractor(spec_path)
|
|
print(f"Reading spec: {spec_path}")
|
|
try:
|
|
models, enums = extractor.extract()
|
|
except (RuntimeError, ValueError) as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
endpoints = extractor.endpoints()
|
|
print(
|
|
f"Extracted {len(models)} models, {len(enums)} enums, "
|
|
f"{len(endpoints)} endpoints"
|
|
)
|
|
|
|
_emit((models, enums), args.targets, args.output)
|
|
|
|
|
|
def cmd_from_tabular(args):
|
|
"""Generate models from a directory of CSV/TSV/ODS spreadsheets."""
|
|
from .loader.extract.tabular import TabularExtractor
|
|
|
|
source_path = Path(args.source)
|
|
if not source_path.exists():
|
|
print(f"Error: Source not found: {source_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
extractor = TabularExtractor(source_path)
|
|
print(f"Reading sheets: {source_path}")
|
|
try:
|
|
models, enums = extractor.extract()
|
|
except ValueError as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
datasets = extractor.datasets()
|
|
rows = sum(len(d.rows) for d in datasets)
|
|
print(f"Extracted {len(models)} models, {rows} rows")
|
|
|
|
# Datasets ride along as a third element: the datagen target seeds from
|
|
# them, and every other target ignores the extra slot.
|
|
_emit((models, enums, datasets), args.targets, args.output)
|
|
|
|
|
|
def cmd_generate(args):
|
|
"""Generate all targets from a JSON config file."""
|
|
import json
|
|
from .loader import load_schema
|
|
|
|
config_path = Path(args.config)
|
|
if not config_path.exists():
|
|
print(f"Error: Config file not found: {config_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
with open(config_path) as f:
|
|
config = json.load(f)
|
|
|
|
# Resolve paths relative to current working directory
|
|
schema_path = Path(config["schema"])
|
|
if not schema_path.exists():
|
|
print(f"Error: Schema folder not found: {schema_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"Loading schema: {schema_path}")
|
|
|
|
for target_conf in config["targets"]:
|
|
target = target_conf["target"]
|
|
output = Path(target_conf["output"])
|
|
include = set(target_conf.get("include", []))
|
|
name_map = target_conf.get("name_map", {})
|
|
|
|
if target not in GENERATORS:
|
|
print(f"Warning: Unknown target '{target}', skipping", file=sys.stderr)
|
|
continue
|
|
|
|
# Load schema with this target's include filter
|
|
schema = load_schema(schema_path, include=include or None)
|
|
|
|
generator = GENERATORS[target](name_map=name_map)
|
|
print(f"Generating {target} to: {output}")
|
|
generator.generate(schema, output)
|
|
|
|
print("Done!")
|
|
|
|
|
|
def cmd_list_formats(args):
|
|
"""List available output formats."""
|
|
print("Available output formats:")
|
|
for fmt in GENERATORS.keys():
|
|
print(f" - {fmt}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Modelgen - Generic Model Generation Tool",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# Available formats for help text
|
|
formats = list(GENERATORS.keys())
|
|
formats_str = ", ".join(formats)
|
|
|
|
# from-config command
|
|
config_parser = subparsers.add_parser(
|
|
"from-config",
|
|
help="Generate models from soleprint configuration file",
|
|
)
|
|
config_parser.add_argument(
|
|
"--config",
|
|
"-c",
|
|
type=str,
|
|
required=True,
|
|
help="Path to configuration file (e.g., config.json)",
|
|
)
|
|
config_parser.add_argument(
|
|
"--output",
|
|
"-o",
|
|
type=str,
|
|
required=True,
|
|
help="Output path (file or directory)",
|
|
)
|
|
config_parser.add_argument(
|
|
"--format",
|
|
"-f",
|
|
type=str,
|
|
default="pydantic",
|
|
choices=["pydantic"], # Only pydantic for config mode
|
|
help="Output format (default: pydantic)",
|
|
)
|
|
config_parser.set_defaults(func=cmd_from_config)
|
|
|
|
# from-schema command
|
|
schema_parser = subparsers.add_parser(
|
|
"from-schema",
|
|
help="Generate models from Python dataclasses in schema/ folder",
|
|
)
|
|
schema_parser.add_argument(
|
|
"--schema",
|
|
"-s",
|
|
type=str,
|
|
default=None,
|
|
help="Path to schema folder (default: ./schema)",
|
|
)
|
|
schema_parser.add_argument(
|
|
"--output",
|
|
"-o",
|
|
type=str,
|
|
required=True,
|
|
help="Output path (file or directory)",
|
|
)
|
|
schema_parser.add_argument(
|
|
"--targets",
|
|
"-t",
|
|
type=str,
|
|
default="pydantic",
|
|
help=f"Comma-separated output targets ({formats_str})",
|
|
)
|
|
schema_parser.add_argument(
|
|
"--include",
|
|
type=str,
|
|
default=None,
|
|
help="Comma-separated model groups to include (dataclasses,enums,api,grpc). Default: all.",
|
|
)
|
|
schema_parser.set_defaults(func=cmd_from_schema)
|
|
|
|
# extract command
|
|
extract_parser = subparsers.add_parser(
|
|
"extract",
|
|
help="Extract models from existing codebase",
|
|
)
|
|
extract_parser.add_argument(
|
|
"--source",
|
|
"-s",
|
|
type=str,
|
|
required=True,
|
|
help="Path to source codebase",
|
|
)
|
|
extract_parser.add_argument(
|
|
"--framework",
|
|
"-f",
|
|
type=str,
|
|
choices=["django", "sqlalchemy", "prisma", "auto"],
|
|
default="auto",
|
|
help="Source framework (default: auto-detect)",
|
|
)
|
|
extract_parser.add_argument(
|
|
"--output",
|
|
"-o",
|
|
type=str,
|
|
required=True,
|
|
help="Output path (file or directory)",
|
|
)
|
|
extract_parser.add_argument(
|
|
"--targets",
|
|
"-t",
|
|
type=str,
|
|
default="pydantic",
|
|
help=f"Comma-separated output targets ({formats_str})",
|
|
)
|
|
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)
|
|
|
|
# from-openapi command (service contract -> models)
|
|
openapi_parser = subparsers.add_parser(
|
|
"from-openapi",
|
|
help="Generate models from an OpenAPI 3.x / Swagger 2.0 document",
|
|
)
|
|
openapi_parser.add_argument(
|
|
"--spec",
|
|
"-s",
|
|
type=str,
|
|
required=True,
|
|
help="Path to the spec (.json, .yaml or .yml; YAML needs PyYAML)",
|
|
)
|
|
openapi_parser.add_argument(
|
|
"--output",
|
|
"-o",
|
|
type=str,
|
|
required=True,
|
|
help="Output path (file or directory)",
|
|
)
|
|
openapi_parser.add_argument(
|
|
"--targets",
|
|
"-t",
|
|
type=str,
|
|
default="pydantic",
|
|
help=f"Comma-separated output targets ({formats_str})",
|
|
)
|
|
openapi_parser.set_defaults(func=cmd_from_openapi)
|
|
|
|
# from-tabular command (spreadsheets -> models + rows)
|
|
tabular_parser = subparsers.add_parser(
|
|
"from-tabular",
|
|
help="Generate models from a directory of CSV/TSV/ODS spreadsheets",
|
|
)
|
|
tabular_parser.add_argument(
|
|
"--source",
|
|
"-s",
|
|
type=str,
|
|
required=True,
|
|
help="Directory of sheets (or a single .csv/.tsv/.ods file)",
|
|
)
|
|
tabular_parser.add_argument(
|
|
"--output",
|
|
"-o",
|
|
type=str,
|
|
required=True,
|
|
help="Output path (file or directory)",
|
|
)
|
|
tabular_parser.add_argument(
|
|
"--targets",
|
|
"-t",
|
|
type=str,
|
|
default="pydantic",
|
|
help=f"Comma-separated output targets ({formats_str}). "
|
|
"The datagen target seeds from the imported rows.",
|
|
)
|
|
tabular_parser.set_defaults(func=cmd_from_tabular)
|
|
|
|
# generate command (config-driven multi-target)
|
|
gen_parser = subparsers.add_parser(
|
|
"generate",
|
|
help="Generate all targets from a JSON config file",
|
|
)
|
|
gen_parser.add_argument(
|
|
"--config",
|
|
"-c",
|
|
type=str,
|
|
required=True,
|
|
help="Path to generation config file (e.g., schema/modelgen.json)",
|
|
)
|
|
gen_parser.set_defaults(func=cmd_generate)
|
|
|
|
# list-formats command
|
|
formats_parser = subparsers.add_parser(
|
|
"list-formats",
|
|
help="List available output formats",
|
|
)
|
|
formats_parser.set_defaults(func=cmd_list_formats)
|
|
|
|
args = parser.parse_args()
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|