pyvelm.loader

loader

Module discovery, dependency resolution, and install/migrate.

A pyvelm module is a Python package containing a __pyvelm__.py manifest. Preferred style (velmphp-like fluent builder)::

from pyvelm.manifest import Manifest

manifest = (
    Manifest.make("partners")
    .version(0, 1, 0)
    .depends("base")
    .data("views/partner.py")
)

Legacy module-level constants remain supported::

NAME = "partners"
VERSION = (0, 1, 0)
DEPENDS = ["base"]

Optionally:

MODELS_PACKAGE = "myapp.partners.models"   # defaults to <pkg>.models
INSTALL_HOOK = "myapp.partners.hooks:install"
# Optional — default: discover seeders/SEEDERS from seeders/__init__.py
SYNC_HOOK = "myapp.partners.hooks:sync"   # runs on Apps Sync (re-install path)
WEB_ROUTES = "myapp.partners.web:register_routes"  # optional FastAPI routes
MIGRATIONS_PACKAGE = "myapp.partners.migrations"  # defaults to <pkg>.migrations

The loader

1. discovers manifests under given roots,
2. resolves dependency order (topo sort, cycle detection),
3. imports each module's models under an active registry,
4. runs install/migrate per module inside one transaction,
5. records installed versions in ir_module.

parse_module_roots_env

parse_module_roots_env(value: str) -> list[Path]

Parse PYVELM_MODULE_ROOTS — comma- or colon-separated paths.

Source code in pyvelm/loader.py

def parse_module_roots_env(value: str) -> list[Path]:
    """Parse ``PYVELM_MODULE_ROOTS`` — comma- or colon-separated paths."""
    import re

    parts = re.split(r"[:,]", value or "")
    return [Path(p.strip()) for p in parts if p.strip()]

module_display_name

module_display_name(name: str, explicit: str | None = None) -> str

Readable Apps label. NAME stays the technical id (geo_data).

Source code in pyvelm/loader.py

def module_display_name(name: str, explicit: str | None = None) -> str:
    """Readable Apps label. ``NAME`` stays the technical id (``geo_data``)."""
    if explicit and str(explicit).strip():
        return str(explicit).strip()
    return " ".join(part.capitalize() for part in name.split("_") if part)

discover_bootstrap_module_names

discover_bootstrap_module_names() -> frozenset[str]

Bundled modules auto-installed on a fresh database.

Modules with BOOTSTRAP = False in their manifest (or Manifest.bootstrap(False)) are discovered but opt-in via Apps.

Source code in pyvelm/loader.py

def discover_bootstrap_module_names() -> frozenset[str]:
    """Bundled modules auto-installed on a fresh database.

    Modules with ``BOOTSTRAP = False`` in their manifest (or
    ``Manifest.bootstrap(False)``) are discovered but opt-in via Apps.
    """
    if not _BUILTIN_MODULES_ROOT.is_dir():
        return frozenset({"base", "admin"})
    names: list[str] = []
    for pkg_path in _BUILTIN_MODULES_ROOT.iterdir():
        manifest_path = pkg_path / "__pyvelm__.py"
        if not pkg_path.is_dir() or not manifest_path.is_file():
            continue
        mod = _exec_manifest_module(pkg_path)
        data = _manifest_dict_from_module(mod, manifest_path)
        if data.get("BOOTSTRAP", True):
            names.append(pkg_path.name)
    return frozenset(names)

discover

discover(roots: list[Path | str] | None = None) -> dict[str, ModuleSpec]

Walk module roots for directories containing a __pyvelm__.py manifest.

Always scans :data:pyvelm.BUILTIN_MODULE_ROOTS first so bundled modules such as contacts are visible even when the app only passes custom addon paths (mirrors pyvelm-cron / CLI behaviour).

Source code in pyvelm/loader.py

def discover(roots: list[Path | str] | None = None) -> dict[str, ModuleSpec]:
    """Walk module roots for directories containing a ``__pyvelm__.py`` manifest.

    Always scans :data:`pyvelm.BUILTIN_MODULE_ROOTS` first so bundled modules
    such as ``contacts`` are visible even when the app only passes custom
  addon paths (mirrors ``pyvelm-cron`` / CLI behaviour).
    """
    specs: dict[str, ModuleSpec] = {}
    for rootp in _discovery_roots(roots):
        # Make the root importable so `partners.models.res_partner` etc. resolves.
        rootp_str = str(rootp)
        if rootp_str not in sys.path:
            sys.path.insert(0, rootp_str)
        for sub in sorted(rootp.iterdir()):
            if not sub.is_dir():
                continue
            spec = _read_manifest(sub)
            if spec is None:
                continue
            if spec.name in specs:
                raise ValueError(
                    f"Duplicate module name {spec.name!r} "
                    f"({specs[spec.name].package_path} and {sub})"
                )
            specs[spec.name] = spec
    return specs

resolve_order

resolve_order(specs: dict[str, ModuleSpec]) -> list[ModuleSpec]

Topological sort by DEPENDS; raises on missing deps or cycles.

Source code in pyvelm/loader.py

def resolve_order(specs: dict[str, ModuleSpec]) -> list[ModuleSpec]:
    """Topological sort by DEPENDS; raises on missing deps or cycles."""
    WHITE, GRAY, BLACK = 0, 1, 2
    color: dict[str, int] = {n: WHITE for n in specs}
    order: list[ModuleSpec] = []

    def visit(n: str, stack: list[str]):
        if color[n] == BLACK:
            return
        if color[n] == GRAY:
            cycle = stack[stack.index(n):] + [n]
            raise ValueError(f"Module dependency cycle: {' -> '.join(cycle)}")
        color[n] = GRAY
        for dep in specs[n].depends:
            if dep not in specs:
                raise ValueError(
                    f"Module {n!r} depends on {dep!r} which was not discovered"
                )
            visit(dep, stack + [n])
        color[n] = BLACK
        order.append(specs[n])

    for n in specs:
        visit(n, [])
    return order

reload_models

reload_models(spec: ModuleSpec, registry: Registry) -> None

Re-import a module's models package (upgrade / dev reload).

Refreshes Python class definitions on the live registry without requiring a full process restart.

Source code in pyvelm/loader.py

def reload_models(spec: ModuleSpec, registry: Registry) -> None:
    """Re-import a module's models package (upgrade / dev reload).

    Refreshes Python class definitions on the live registry without
    requiring a full process restart.
    """
    if not _has_models_package(spec):
        return
    with registry.activate():
        before_models: dict[str, type] = dict(registry._models)
        _ensure_top_level_package(spec)
        pkg = importlib.import_module(spec.models_package)
        importlib.reload(pkg)
        prefix = spec.models_package + "."
        for mod_name in sorted(
            k for k in sys.modules if k.startswith(prefix)
        ):
            importlib.reload(sys.modules[mod_name])
        _sync_models_from_package(spec, registry, before_models)

reload_installed_models

reload_installed_models(env: Environment, specs: dict[str, ModuleSpec]) -> None

Re-import models for every installed module in dependency order.

Reloading a single module overwrites _inherit merges on shared models (e.g. upgrading base alone drops geo_data fields on res.country while res.continent still references them).

Source code in pyvelm/loader.py

def reload_installed_models(env: Environment, specs: dict[str, ModuleSpec]) -> None:
    """Re-import models for every installed module in dependency order.

    Reloading a single module overwrites ``_inherit`` merges on shared
    models (e.g. upgrading ``base`` alone drops ``geo_data`` fields on
    ``res.country`` while ``res.continent`` still references them).
    """
    _ensure_ir_module(env)
    rows = env.conn.execute(
        f'SELECT "name" FROM "{IR_MODULE_TABLE}"',
    ).fetchall()
    installed = {r[0] for r in rows}
    subset = {k: v for k, v in specs.items() if k in installed}
    if not subset:
        return
    for spec in resolve_order(subset):
        reload_models(spec, env.registry)

specs_to_install

specs_to_install(env: Environment, ordered: list[ModuleSpec], *, install_all: bool = False) -> list[ModuleSpec]

Return the subset of ordered specs to load and install on this pass.

By default (app/cron boot): on a fresh database every bundled module in pyvelm/modules/ (BOOTSTRAP_MODULES) is installed; otherwise only rows already present in ir_module. Pass install_all=True to also install discovered addons outside the bundled tree (migrate --all, demo scripts, integration tests).

Source code in pyvelm/loader.py

def specs_to_install(
    env: Environment,
    ordered: list[ModuleSpec],
    *,
    install_all: bool = False,
) -> list[ModuleSpec]:
    """Return the subset of *ordered* specs to load and install on this pass.

    By default (app/cron boot): on a fresh database every bundled module in
    ``pyvelm/modules/`` (``BOOTSTRAP_MODULES``) is installed; otherwise only
    rows already present in ``ir_module``. Pass ``install_all=True`` to also
    install discovered addons outside the bundled tree (``migrate --all``,
    demo scripts, integration tests).
    """
    if install_all:
        return ordered
    installed = _installed_module_names(env)
    if not installed:
        return [s for s in ordered if s.name in BOOTSTRAP_MODULES]
    return [s for s in ordered if s.name in installed]

install

install(specs: list[ModuleSpec], env: Environment) -> list[dict]

Install or upgrade each module, in specs order, atomically per module. Models must already be loaded into env.registry.

Returns one result dict per spec with keys name, schema, views, menus (human-readable summaries for the Apps UI).

Source code in pyvelm/loader.py

def install(specs: list[ModuleSpec], env: Environment) -> list[dict]:
    """Install or upgrade each module, in `specs` order, atomically per
    module. Models must already be loaded into `env.registry`.

    Returns one result dict per spec with keys ``name``, ``schema``,
    ``views``, ``menus`` (human-readable summaries for the Apps UI).
    """
    from . import db_autogen

    with env.transaction():
        _ensure_ir_module(env)
    results: list[dict] = []
    for spec in specs:
        with env.transaction():
            current = _installed_version(env, spec.name)
            schema_note = ""
            if current is None:
                _setup_module_schema(spec, env)
                applied = db_autogen.apply_schema_diff(env, spec.name)
                schema_note = applied.summary()
                if spec.install_hook is not None:
                    spec.install_hook(env)
                from pyvelm.database import _conn_capabilities, now_sql

                cap = _conn_capabilities(env.conn)
                env.conn.execute(
                    f'INSERT INTO "{IR_MODULE_TABLE}" '
                    f'("name", "version", "installed_at") '
                    f'VALUES (%s, %s, {now_sql(cap)})',
                    [spec.name, spec.version_str],
                )
            else:
                _setup_module_schema(spec, env)
                if current < spec.version:
                    _run_migrations(spec, env, current, spec.version)
                # Sync hook before schema apply: idempotent backfills and
                # orphan cleanup (runs every upgrade/Sync, not only on bump).
                if spec.sync_hook is not None:
                    spec.sync_hook(env)
                applied = db_autogen.apply_schema_diff(env, spec.name)
                schema_note = applied.summary()
                from pyvelm.database import _conn_capabilities, now_sql

                cap = _conn_capabilities(env.conn)
                env.conn.execute(
                    f'UPDATE "{IR_MODULE_TABLE}" SET "version" = %s, '
                    f'"installed_at" = {now_sql(cap)} WHERE "name" = %s',
                    [spec.version_str, spec.name],
                )
            _run_module_seeders(spec, env)
            # Load data files (views, menus) from disk — always reload
            # so Upgrade/Sync picks up new DATA without reinstall.
            _load_data_files(spec)
            view_count = len(spec.views)
            inherit_count = len(spec.view_inherits)
            menu_count = len(spec.menus)
            _sync_views(spec, env)
            _sync_view_inherits(spec, env)
            _sync_menus(spec, env)
            results.append(
                {
                    "name": spec.name,
                    "schema": schema_note,
                    "views": (
                        f"{view_count} view(s)"
                        + (
                            f", {inherit_count} inherit(s)"
                            if inherit_count
                            else ""
                        )
                    ),
                    "menus": f"{menu_count} menu(s)",
                }
            )

    # Build cross-model indexes once everything's loaded.
    env.registry._build_o2m_inverse_index()
    env.registry._build_m2o_referrers_index()
    env.registry._build_m2m_relation_index()
    env.registry._build_compute_graph()
    for cls in env.registry._models.values():
        cls._validate_relations(env.registry)
    return results

load_and_install

load_and_install(roots: list[Path | str], env: Environment, *, install_all: bool = False) -> list[ModuleSpec]

End-to-end: discover, resolve, load models, install/sync.

By default every bundled module under pyvelm/modules/ is installed on a fresh database; other discovered addons stay available in Apps until installed. Pass install_all=True to install every discovered module (used by pyvelm migrate --all and demo scripts). Returns the specs that were loaded and installed/synced.

Source code in pyvelm/loader.py

def load_and_install(
    roots: list[Path | str],
    env: Environment,
    *,
    install_all: bool = False,
) -> list[ModuleSpec]:
    """End-to-end: discover, resolve, load models, install/sync.

    By default every bundled module under ``pyvelm/modules/`` is installed on
    a fresh database; other discovered addons stay available in **Apps** until
    installed. Pass ``install_all=True`` to install every discovered module
    (used by ``pyvelm migrate --all`` and demo scripts). Returns the specs
    that were loaded and installed/synced.
    """
    from pyvelm.policies import register_builtin_policies

    register_builtin_policies()
    specs = discover(roots)
    ordered = resolve_order(specs)
    to_install = specs_to_install(env, ordered, install_all=install_all)
    for spec in to_install:
        _load_models(spec, env.registry)
    install(to_install, env)
    return to_install

register_web_routes

register_web_routes(app, roots: list[Path | str], *, only: set[str] | frozenset[str] | None = None) -> None

Mount each discovered module's WEB_ROUTES registrar on app.

Modules declare WEB_ROUTES = "pkg.web:register_routes" in __pyvelm__.py. The callable receives the FastAPI app (with app.state.registry and app.state.pool already set) and should attach routes, static mounts, or routers. Registrars run in dependency order after core create_app routes are registered.

Only installed modules register routes — uninstalled addons stay visible in Apps but do not mount HTTP handlers until installed.

Pass only={...} after a live Apps install to mount routes for modules that were not present in ir_module when the process started. Each module's registrar runs at most once per process (tracked on app.state.registered_web_route_modules).

Source code in pyvelm/loader.py

def register_web_routes(
    app,
    roots: list[Path | str],
    *,
    only: set[str] | frozenset[str] | None = None,
) -> None:
    """Mount each discovered module's ``WEB_ROUTES`` registrar on *app*.

    Modules declare ``WEB_ROUTES = "pkg.web:register_routes"`` in
    ``__pyvelm__.py``. The callable receives the FastAPI app (with
    ``app.state.registry`` and ``app.state.pool`` already set) and should
    attach routes, static mounts, or routers. Registrars run in dependency
    order after core ``create_app`` routes are registered.

    Only **installed** modules register routes — uninstalled addons stay
    visible in **Apps** but do not mount HTTP handlers until installed.

    Pass ``only={...}`` after a live Apps install to mount routes for modules
    that were not present in ``ir_module`` when the process started. Each
    module's registrar runs at most once per process (tracked on
    ``app.state.registered_web_route_modules``).
    """
    from pyvelm import Environment

    registered: set[str] = getattr(app.state, "registered_web_route_modules", None)
    if registered is None:
        registered = set()
        app.state.registered_web_route_modules = registered

    specs = discover(roots)
    ordered = resolve_order(specs)
    installed: set[str] | None = None
    if getattr(app.state, "pool", None) is not None:
        with app.state.pool.connection() as conn:
            env = Environment(conn, registry=app.state.registry, uid=None)
            installed = _installed_module_names(env)
    for spec in ordered:
        if only is not None and spec.name not in only:
            continue
        if spec.name in registered:
            continue
        if installed is not None and spec.name not in installed:
            continue
        if not spec.web_routes:
            continue
        registrar = _import_attr(spec.web_routes)
        registrar(app)
        registered.add(spec.name)

discover_commands

discover_commands(roots: list[Path | str], registry: Any | None = None) -> Any

Discover and register console commands from all modules under roots.

Returns a :class:~pyvelm.console.CommandRegistry. Does not require a database connection unless a command sets requires_db=True.

Source code in pyvelm/loader.py

def discover_commands(
    roots: list[Path | str],
    registry: Any | None = None,
) -> Any:
    """Discover and register console commands from all modules under ``roots``.

    Returns a :class:`~pyvelm.console.CommandRegistry`. Does not require a
    database connection unless a command sets ``requires_db=True``.
    """
    from .console import CommandRegistry

    if registry is None:
        registry = CommandRegistry()
    specs = discover(roots)
    ordered = resolve_order(specs)
    for spec in ordered:
        for ref in spec.command_refs:
            cls = _import_attr(ref)
            if not isinstance(cls, type):
                raise TypeError(f"COMMANDS entry {ref!r} must be a Command class")
            registry.register(cls())
        for cls in _load_commands_from_package(spec):
            registry.register(cls())
    return registry