You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import importlib.resources
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import click
|
|
|
|
|
|
@click.group()
|
|
def iti_system_cli() -> None:
|
|
"""iTi-System commands."""
|
|
|
|
|
|
@iti_system_cli.group()
|
|
def seed() -> None:
|
|
"""Seed system-domain data."""
|
|
|
|
|
|
@seed.command("system")
|
|
@click.argument("app_import")
|
|
def seed_system(app_import: str) -> None:
|
|
app = _load_app(app_import)
|
|
registry = getattr(app.state, "iti_modules", None)
|
|
factory = getattr(app.state, "db_sessionmaker", None)
|
|
if factory is None:
|
|
raise click.ClickException("app database is not configured")
|
|
|
|
from .seeds.system import seed_system_data
|
|
|
|
with factory() as db:
|
|
summary = seed_system_data(db, registry)
|
|
for name, result in summary.items():
|
|
click.echo(
|
|
f"{name}: created {result['created']}, updated {result['updated']}, skipped {result['skipped']}"
|
|
)
|
|
|
|
|
|
@iti_system_cli.group()
|
|
def migrations() -> None:
|
|
"""System migration helpers."""
|
|
|
|
|
|
@migrations.command("sync")
|
|
@click.option(
|
|
"--target",
|
|
default="migrations/versions",
|
|
show_default=True,
|
|
help="业务项目 migration versions 目录",
|
|
)
|
|
def sync_migrations(target: str) -> None:
|
|
target_dir = Path(target)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
synced: list[str] = []
|
|
skipped: list[str] = []
|
|
source_root = importlib.resources.files("iti_system.migrations").joinpath("versions")
|
|
for source in source_root.iterdir():
|
|
if not source.name.endswith(".py") or source.name == "__init__.py":
|
|
continue
|
|
target_file = target_dir / source.name
|
|
if target_file.exists():
|
|
skipped.append(source.name)
|
|
continue
|
|
with importlib.resources.as_file(source) as source_path:
|
|
shutil.copy2(source_path, target_file)
|
|
synced.append(source.name)
|
|
|
|
for name in synced:
|
|
click.echo(f"synced: {name}")
|
|
for name in skipped:
|
|
click.echo(f"skipped: {name}")
|
|
click.echo(f"summary: synced {len(synced)}, skipped {len(skipped)}")
|
|
|
|
|
|
def _load_app(app_import: str):
|
|
module_name, _, attr_name = app_import.partition(":")
|
|
if not module_name or not attr_name:
|
|
raise click.ClickException("app import must use module:attribute")
|
|
module = importlib.import_module(module_name)
|
|
app = getattr(module, attr_name)
|
|
return app() if callable(app) else app
|