spb/company-atlas

1"""`catlas` — Company Atlas operations CLI.23Command groups live in `companyatlas.commands.<module>`; each module exposes `register(app: typer.Typer) -> None` and is4auto-discovered here, so crawl, intelligence, API and seed commands can evolve independently.5"""6from __future__ import annotations78import asyncio9import importlib10import logging11import pkgutil12from pathlib import Path13from typing import Annotated1415import typer16from rich.console import Console1718from companyatlas.config import settings19from companyatlas.logging import setup_logging2021app = typer.Typer(name="catlas", help="Company Atlas — the live atlas of global companies.", no_args_is_help=True, add_completion=False)22console = Console(stderr=True)23out = Console()242526@app.callback()27def _main(verbose: Annotated[bool, typer.Option("--verbose", "-v")] = False) -> None:28    setup_logging(level=logging.DEBUG if verbose else logging.INFO, service="catlas")29    settings.ensure_dirs()303132def run_async(coro):  # type: ignore[no-untyped-def]33    """Run a coroutine and dispose the shared DB engine afterwards (for CLI commands)."""34    from companyatlas.db import dispose3536    async def wrapper():  # type: ignore[no-untyped-def]37        try:38            return await coro39        finally:40            await dispose()4142    return asyncio.run(wrapper())434445@app.command()46def migrate(revision: str = "head") -> None:47    """Apply database migrations (forward-only)."""48    from alembic import command49    from alembic.config import Config5051    root = Path(__file__).resolve().parents[2]52    cfg = Config(str(root / "alembic.ini"))53    cfg.set_main_option("script_location", str(root / "migrations"))54    command.upgrade(cfg, revision)55    out.print("[green]migrations applied[/]")565758@app.command()59def api(host: str | None = None, port: int | None = None, workers: int = 1, reload: bool = False) -> None:60    """Serve the FastAPI application (development)."""61    import uvicorn6263    uvicorn.run("companyatlas.api.main:app", host=host or settings.api_host, port=port or settings.api_port, workers=workers, reload=reload,64                proxy_headers=True, access_log=False)656667@app.command()68def version() -> None:69    import companyatlas7071    out.print(companyatlas.__version__)727374def _discover_commands() -> None:75    try:76        import companyatlas.commands as pkg77    except ImportError:78        return79    for mod in pkgutil.iter_modules(pkg.__path__):80        if mod.name.startswith("_"):81            continue82        module = importlib.import_module(f"companyatlas.commands.{mod.name}")83        register = getattr(module, "register", None)84        if callable(register):85            register(app)868788_discover_commands()8990__all__ = ["app", "console", "out", "run_async"]91