pyvelm.cli

cli

pyvelm command-line entry points.

A single pyvelm command dispatches subcommands:

pyvelm cron                Background cron + mail-dispatcher worker.
pyvelm init <name>         Scaffold a new pyvelm project.
pyvelm new <module>        Drop a runnable module skeleton into a project.
pyvelm db diff <module>    Print the schema delta for a module.
pyvelm db autogen <module> Write an additive migration file.
pyvelm migrate               Upgrade installed modules (deploy hook).
pyvelm db migrate-fresh    Same as migrate, with plan + prod confirmation.
pyvelm migrate:fresh       DEV ONLY — drop schema, then ``migrate``.
pyvelm migrate:reset       DEV ONLY — drop schema (same wipe as ``db nuke``).
pyvelm db nuke             DEV ONLY — drop schema + reinstall every module.
pyvelm db status           Installed vs on-disk module versions.
pyvelm list                List core and module commands.
pyvelm make:module …       Scaffold a module (see docs/console.md).

The legacy pyvelm-cron entry point keeps working — it's a thin alias for pyvelm cron so existing docker-compose files and systemd units don't need editing during upgrades.

Configuration is env-driven (CLI flags override). Most apps set these in their .env:

PYVELM_DSN              Postgres DSN. Required for ``cron``/``db``.
PYVELM_MODULE_ROOTS     Extra module dirs (comma or colon separated).
                        ``cron``/``db`` also auto-detect ``modules_root``
                        from ``pyvelm.toml`` in cwd or a parent.
PYVELM_CRON_INTERVAL    Seconds between cron ticks. Default 60.

cron_loop

cron_loop(*, dsn: str, roots: list[Path], interval: float) -> None

Boot the registry against dsn + roots, then loop forever.

SIGTERM / SIGINT flip a shutdown flag and the loop exits cleanly after the current tick — useful for graceful container restarts.

Source code in pyvelm/cli.py

def cron_loop(*, dsn: str, roots: list[Path], interval: float) -> None:
    """Boot the registry against `dsn` + `roots`, then loop forever.

    SIGTERM / SIGINT flip a shutdown flag and the loop exits cleanly
    after the current tick — useful for graceful container restarts.
    """
    from .database import create_database_from_dsn, normalize_dsn

    registry = Registry()
    database = create_database_from_dsn(normalize_dsn(dsn), pool_size=2)
    with database.connect() as conn:
        env = Environment(conn, registry=registry)
        loader.load_and_install(roots, env)
        log.info("loaded modules; cron runner ready")

    pool = database.pool

    shutdown = False

    def _sig(_signum, _frame):
        nonlocal shutdown
        shutdown = True
        log.info("shutdown signal received")

    signal.signal(signal.SIGTERM, _sig)
    signal.signal(signal.SIGINT, _sig)

    try:
        while not shutdown:
            try:
                _tick(pool, registry)
            except Exception:  # noqa: BLE001
                # Don't kill the runner on a single failed tick — log
                # and try again next interval.
                log.exception("cron tick failed")
            # Sleep in 1-second slices so the signal handler can pre-empt.
            elapsed = 0.0
            while elapsed < interval and not shutdown:
                time.sleep(min(1.0, interval - elapsed))
                elapsed += 1.0
    finally:
        database.dispose()

bootstrap_command_env

bootstrap_command_env(ctx) -> None

Load registry + DB env for commands with requires_db=True.

Source code in pyvelm/cli.py

def bootstrap_command_env(ctx) -> None:
    """Load registry + DB env for commands with ``requires_db=True``."""
    from .database import create_database_from_dsn, normalize_dsn
    from .env import Environment
    from .registry import Registry

    dsn = os.environ.get("PYVELM_DSN")
    if not dsn:
        sys.exit("PYVELM_DSN not set (required for this command)")
    specs = loader.discover(ctx.roots)
    ordered = loader.resolve_order(specs)
    registry = Registry()
    for spec in ordered:
        loader._load_models(spec, registry)
    db = create_database_from_dsn(normalize_dsn(dsn))
    conn = db.open_connection()
    ctx.registry = registry
    ctx.env = Environment(conn, registry=registry)

main

main() -> None

pyvelm entry point — subcommand dispatch.

Source code in pyvelm/cli.py

def main() -> None:
    """``pyvelm`` entry point — subcommand dispatch."""
    _load_dotenv()
    argv = sys.argv[1:]
    if argv and _try_dispatch_module_command(argv):
        return
    parser = _build_parser()
    args = parser.parse_args(argv)
    if args.command is None:
        parser.print_help()
        print("\nMore commands (run `pyvelm list` for the full list):")
        reg = _command_registry()
        for cmd in reg.all()[:8]:
            print(f"  {cmd.name:<20} {cmd.description or ''}")
        if len(reg.all()) > 8:
            print(f"  … and {len(reg.all()) - 8} more")
        sys.exit(0)
    args.func(args)

cron_main

cron_main() -> None

pyvelm-cron legacy entry point.

Same shape as the old pyvelm-cron command (--interval, --roots), still works without subcommand prefix. Kept so existing docker-compose files and systemd units survive a pyvelm upgrade without edits.

Source code in pyvelm/cli.py

def cron_main() -> None:
    """``pyvelm-cron`` legacy entry point.

    Same shape as the old ``pyvelm-cron`` command (``--interval``,
    ``--roots``), still works without subcommand prefix. Kept so
    existing docker-compose files and systemd units survive a
    pyvelm upgrade without edits.
    """
    _load_dotenv()
    parser = argparse.ArgumentParser(
        prog="pyvelm-cron",
        description=(
            "Legacy entry; same as `pyvelm cron`. Kept for "
            "backward-compat with existing deployment configs."
        ),
    )
    _add_cron_args(parser)
    args = parser.parse_args()
    _run_cron(args)