API Reference

Runtime support library for pyrig.

Contains code that a project needs at runtime, such as CLI invocation.

core

Infrastructure for cross-package dependency and subclass discovery.

constants

Constants used throughout the project.

dependencies

Cross-package subclass discovery driven by the installed package dependency graph.

Supports plugin hierarchies in dependent projects by automatically locating all subclass implementations defined across the installed dependency graph, without requiring explicit registration.

discovery

Subclass and module discovery scoped across installed package dependents.

deps_depending_on_dep(dependency) cached

Return every installed package that depends on dependency.

The result is cached per unique dependency argument.

Parameters:

Name	Type	Description	Default
dependency	ModuleType	Package whose dependents should be discovered.	required

Returns:

Type	Description
ModuleType	Tuple of imported module objects for every package that depends on
...	dependency directly or transitively, in dependency order.
tuple[ModuleType, ...]	Does not include dependency itself.

Source code in src/pyrig_runtime/core/dependencies/discovery.py

@cache
def deps_depending_on_dep(dependency: ModuleType) -> tuple[ModuleType, ...]:
    """Return every installed package that depends on `dependency`.

    The result is cached per unique `dependency` argument.

    Args:
        dependency: Package whose dependents should be discovered.

    Returns:
        Tuple of imported module objects for every package that depends on
        `dependency` directly or transitively, in dependency order.
        Does not include `dependency` itself.
    """
    return tuple(
        import_modules(
            DependencyGraph.cached(root=dependency.__name__).sorted_ancestors(
                dependency.__name__
            )
        )
    )

discover_equivalent_modules_across_dependents(module)

Yield the equivalent module from every dependent of module's root package.

For each installed package that depends on the root of module, locates the module at the same sub-path within that dependent and yields it if it can be imported. The root package itself is excluded from results.

Parameters:

Name	Type	Description	Default
module	ModuleType	Module whose root determines which dependents to search and whose sub-path within that root is used to locate the corresponding module in each dependent.	required

Yields:

Type	Description
ModuleType	Successfully imported module objects in dependency order. Dependents
ModuleType	that have no module at the equivalent sub-path are silently skipped.

Source code in src/pyrig_runtime/core/dependencies/discovery.py

def discover_equivalent_modules_across_dependents(
    module: ModuleType,
) -> Iterator[ModuleType]:
    """Yield the equivalent module from every dependent of `module`'s root package.

    For each installed package that depends on the root of `module`, locates
    the module at the same sub-path within that dependent and yields it if
    it can be imported. The root package itself is excluded from results.

    Args:
        module: Module whose root determines which dependents to search and
            whose sub-path within that root is used to locate the corresponding
            module in each dependent.

    Yields:
        Successfully imported module objects in dependency order. Dependents
        that have no module at the equivalent sub-path are silently skipped.
    """
    dependency = root_module(module)
    logger.debug(
        "Discovering modules equivalent to %s in packages depending on %s",
        module.__name__,
        dependency.__name__,
    )

    for package in deps_depending_on_dep(dependency):
        package_module = replace_root_module(module, package.__name__, default=None)
        if package_module is not None:
            yield package_module

discover_subclasses_across_dependencies(cls, module)

Yield subclasses of cls found in module and in dependent packages.

Parameters:

Name	Type	Description	Default
cls	type[T]	Base class whose subclasses should be discovered.	required
module	ModuleType	Module to search first, also used to determine which root package to scan for dependents.	required

Yields:

Type	Description
type[T]	Subclass types of cls, with module searched first, then
type[T]	dependent packages in dependency order.

Source code in src/pyrig_runtime/core/dependencies/discovery.py

def discover_subclasses_across_dependencies[T](
    cls: type[T],
    module: ModuleType,
) -> Iterator[type[T]]:
    """Yield subclasses of `cls` found in `module` and in dependent packages.

    Args:
        cls: Base class whose subclasses should be discovered.
        module: Module to search first, also used to determine which root
            package to scan for dependents.

    Yields:
        Subclass types of `cls`, with `module` searched first, then
        dependent packages in dependency order.
    """
    logger.debug(
        "Discovering subclasses of %s from modules in packages depending on %s",
        cls.__name__,
        module.__name__,
    )

    return (
        subclass
        for mod in chain(
            (module,),
            discover_equivalent_modules_across_dependents(module=module),
        )
        for subclass in discover_subclasses_across_module(
            cls,
            module=mod,
        )
    )

graph

Directed graph of installed Python package dependency relationships.

DependencyGraph

Bases: DiGraph

Directed graph of installed Python package dependencies.

Nodes are package names; an edge A → B means "A depends on B". The graph is built at instantiation by scanning all installed distributions.

When a root package is given, the graph is pruned to retain only that package and every package that depends on it directly or transitively.

Source code in src/pyrig_runtime/core/dependencies/graph.py

class DependencyGraph(DiGraph):
    """Directed graph of installed Python package dependencies.

    Nodes are package names; an edge A → B means "A depends on B".
    The graph is built at instantiation by scanning all installed
    distributions.

    When a `root` package is given, the graph is pruned to retain only
    that package and every package that depends on it directly or
    transitively.
    """

    def __init__(self, root: str | None = None) -> None:
        """Initialize the dependency graph rooted at the given package.

        Args:
            root: Name of the root package. Accepts either the installed name
                (`some-package`) or the import name (`some_package`).
        """
        super().__init__(
            root=dependency_requirement_as_module_name(root) if root else None
        )

    def build(self) -> None:
        """Build the graph from installed Python distributions."""
        for dist in importlib.metadata.distributions():
            name, deps = self.parse_name_and_deps(dist)
            self.add_node(name)
            for dep in deps:
                self.add_edge(name, dep)

    def parse_name_and_deps(
        self, dist: importlib.metadata.Distribution
    ) -> tuple[str, Iterator[str]]:
        """Extract the package name and dependencies from a distribution.

        Both the package name and every dependency name are normalized to an
        importable module name; version specifiers and extras in requirement
        strings are stripped. Dots are preserved for namespace packages
        (e.g. ``zope.interface`` remains ``zope.interface``). The dependency
        iterator is exhausted once consumed; it yields nothing when the
        distribution declares no dependencies.

        Args:
            dist: Distribution to extract metadata from.

        Returns:
            A two-tuple `(name, deps)` where `name` is the normalized package
            name and `deps` is an iterator over the normalized name of each
            declared dependency.
        """
        return dependency_requirement_as_module_name(dist.name), (
            dependency_requirement_as_module_name(req) for req in (dist.requires or [])
        )

__init__(root=None)

Initialize the dependency graph rooted at the given package.

Parameters:

Name	Type	Description	Default
root	str | None	Name of the root package. Accepts either the installed name (some-package) or the import name (some_package).	None

Source code in src/pyrig_runtime/core/dependencies/graph.py

def __init__(self, root: str | None = None) -> None:
    """Initialize the dependency graph rooted at the given package.

    Args:
        root: Name of the root package. Accepts either the installed name
            (`some-package`) or the import name (`some_package`).
    """
    super().__init__(
        root=dependency_requirement_as_module_name(root) if root else None
    )

add_edge(source, target)

Add a directed edge from source to target.

Creates both nodes if they do not already exist.

Parameters:

Name	Type	Description	Default
source	str	Edge origin node.	required
target	str	Edge destination node.	required

Source code in src/pyrig_runtime/core/graph.py

def add_edge(self, source: str, target: str) -> None:
    """Add a directed edge from source to target.

    Creates both nodes if they do not already exist.

    Args:
        source: Edge origin node.
        target: Edge destination node.
    """
    self.add_node(source)
    self.add_node(target)
    self.edges[source].add(target)
    self.reverse_edges[target].add(source)

add_node(node)

Add a node to the graph. No-op if the node already exists.

Source code in src/pyrig_runtime/core/graph.py

def add_node(self, node: str) -> None:
    """Add a node to the graph. No-op if the node already exists."""
    self.nodes.add(node)
    if node not in self.edges:
        self.edges[node] = set()
    if node not in self.reverse_edges:
        self.reverse_edges[node] = set()

ancestors(target)

Find all nodes that have a directed path to the target node.

The target itself is excluded from the result.

Parameters:

Name	Type	Description	Default
target	str	Node to find ancestors for.	required

Returns:

Type	Description
set[str]	Set of all nodes with a directed path to the target, excluding the
set[str]	target itself.

Raises:

Type	Description
KeyError	If the target node is not in the graph.

Source code in src/pyrig_runtime/core/graph.py

def ancestors(self, target: str) -> set[str]:
    """Find all nodes that have a directed path to the target node.

    The target itself is excluded from the result.

    Args:
        target: Node to find ancestors for.

    Returns:
        Set of all nodes with a directed path to the target, excluding the
        target itself.

    Raises:
        KeyError: If the target node is not in the graph.
    """
    visited: set[str] = set()
    queue: deque[str] = deque(self.reverse_edges[target])

    while queue:
        node = queue.popleft()
        if node not in visited:
            visited.add(node)
            for neighbor in self.reverse_edges[node]:
                if neighbor not in visited:
                    queue.append(neighbor)

    return visited

build()

Build the graph from installed Python distributions.

Source code in src/pyrig_runtime/core/dependencies/graph.py

def build(self) -> None:
    """Build the graph from installed Python distributions."""
    for dist in importlib.metadata.distributions():
        name, deps = self.parse_name_and_deps(dist)
        self.add_node(name)
        for dep in deps:
            self.add_edge(name, dep)

cached(*args, **kwargs) cached classmethod

Return a cached instance, constructing it only on the first call.

Repeated calls with identical arguments return the same instance without rebuilding the graph.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments forwarded to the constructor.	()
**kwargs	Any	Keyword arguments forwarded to the constructor.	{}

Returns:

Type	Description
Self	The cached graph instance for the given arguments.

Source code in src/pyrig_runtime/core/graph.py

@classmethod
@cache
def cached(cls, *args: Any, **kwargs: Any) -> Self:
    """Return a cached instance, constructing it only on the first call.

    Repeated calls with identical arguments return the same instance
    without rebuilding the graph.

    Args:
        *args: Positional arguments forwarded to the constructor.
        **kwargs: Keyword arguments forwarded to the constructor.

    Returns:
        The cached graph instance for the given arguments.
    """
    return cls(*args, **kwargs)

parse_name_and_deps(dist)

Extract the package name and dependencies from a distribution.

Both the package name and every dependency name are normalized to an importable module name; version specifiers and extras in requirement strings are stripped. Dots are preserved for namespace packages (e.g. zope.interface remains zope.interface). The dependency iterator is exhausted once consumed; it yields nothing when the distribution declares no dependencies.

Parameters:

Name	Type	Description	Default
dist	Distribution	Distribution to extract metadata from.	required

Returns:

Type	Description
str	A two-tuple (name, deps) where name is the normalized package
Iterator[str]	name and deps is an iterator over the normalized name of each
tuple[str, Iterator[str]]	declared dependency.

Source code in src/pyrig_runtime/core/dependencies/graph.py

def parse_name_and_deps(
    self, dist: importlib.metadata.Distribution
) -> tuple[str, Iterator[str]]:
    """Extract the package name and dependencies from a distribution.

    Both the package name and every dependency name are normalized to an
    importable module name; version specifiers and extras in requirement
    strings are stripped. Dots are preserved for namespace packages
    (e.g. ``zope.interface`` remains ``zope.interface``). The dependency
    iterator is exhausted once consumed; it yields nothing when the
    distribution declares no dependencies.

    Args:
        dist: Distribution to extract metadata from.

    Returns:
        A two-tuple `(name, deps)` where `name` is the normalized package
        name and `deps` is an iterator over the normalized name of each
        declared dependency.
    """
    return dependency_requirement_as_module_name(dist.name), (
        dependency_requirement_as_module_name(req) for req in (dist.requires or [])
    )

prune(root)

Retain only the given root and its ancestors, removing all other nodes.

Keeps root and all its ancestors (nodes with a directed path to root). All other nodes and their associated edges are removed.

Parameters:

Name	Type	Description	Default
root	str	The node to keep, along with all nodes that have a directed path to it.	required

Source code in src/pyrig_runtime/core/graph.py

def prune(self, root: str) -> None:
    """Retain only the given root and its ancestors, removing all other nodes.

    Keeps `root` and all its ancestors (nodes with a directed path to
    `root`). All other nodes and their associated edges are removed.

    Args:
        root: The node to keep, along with all nodes that have a directed
            path to it.
    """
    keep = self.ancestors(root) | {root}
    self.nodes = keep
    self.edges = {n: self.edges[n] & keep for n in keep}
    self.reverse_edges = {n: self.reverse_edges[n] & keep for n in keep}

sorted_ancestors(target)

Return all ancestors of the target node in topological order.

Ancestors are nodes that have a directed path to the target. The result is ordered so that each ancestor appears after all ancestors it has outgoing edges to.

Parameters:

Name	Type	Description	Default
target	str	Node to find ancestors of.	required

Returns:

Type	Description
list[str]	List of ancestor node identifiers in topological order.
list[str]	Returns an empty list if the target has no ancestors.

Raises:

Type	Description
KeyError	If the target node is not in the graph.
RuntimeError	If the ancestor subgraph contains a cycle, making topological sorting impossible.

Source code in src/pyrig_runtime/core/graph.py

def sorted_ancestors(self, target: str) -> list[str]:
    """Return all ancestors of the target node in topological order.

    Ancestors are nodes that have a directed path to the target. The
    result is ordered so that each ancestor appears after all ancestors
    it has outgoing edges to.

    Args:
        target: Node to find ancestors of.

    Returns:
        List of ancestor node identifiers in topological order.
        Returns an empty list if the target has no ancestors.

    Raises:
        KeyError: If the target node is not in the graph.
        RuntimeError: If the ancestor subgraph contains a cycle, making
            topological sorting impossible.
    """
    return self.topological_sort_subgraph(self.ancestors(target))

topological_sort_subgraph(nodes)

Sort a subset of nodes in topological order.

If there is an edge from A to B, B appears before A in the result. The ordering is deterministic: when multiple nodes are eligible to be emitted at the same step, they are emitted in ascending lexicographic order.

Only edges whose both endpoints are in nodes are considered; edges to or from nodes outside the subset are ignored.

Parameters:

Name	Type	Description	Default
nodes	set[str]	The subset of nodes to sort.	required

Returns:

Type	Description
list[str]	List of nodes in topological order, with each node appearing
list[str]	after all nodes it has outgoing edges to.

Raises:

Type	Description
RuntimeError	If the subgraph contains a cycle, making topological sorting impossible.

Source code in src/pyrig_runtime/core/graph.py

def topological_sort_subgraph(self, nodes: set[str]) -> list[str]:
    """Sort a subset of nodes in topological order.

    If there is an edge from A to B, B appears before A in the result.
    The ordering is deterministic: when multiple nodes are eligible to be
    emitted at the same step, they are emitted in ascending lexicographic
    order.

    Only edges whose both endpoints are in `nodes` are considered; edges
    to or from nodes outside the subset are ignored.

    Args:
        nodes: The subset of nodes to sort.

    Returns:
        List of nodes in topological order, with each node appearing
        after all nodes it has outgoing edges to.

    Raises:
        RuntimeError: If the subgraph contains a cycle, making topological
            sorting impossible.
    """
    out_degree: dict[str, int] = dict.fromkeys(nodes, 0)

    for node in nodes:
        for dependency in self.edges[node]:
            if dependency in nodes:
                out_degree[node] += 1

    heap: list[str] = [node for node in nodes if out_degree[node] == 0]
    heapq.heapify(heap)
    result: list[str] = []

    while heap:
        node = heapq.heappop(heap)
        result.append(node)

        for dependent in self.reverse_edges[node]:
            if dependent in nodes:
                out_degree[dependent] -= 1
                if out_degree[dependent] == 0:
                    heapq.heappush(heap, dependent)

    if len(result) != len(nodes):
        msg = (
            "Graph contains a cycle; topological sort not possible. "
            "This indicates a circular dependency among the following nodes: "
            f"{set(nodes) - set(result)}"
        )
        raise RuntimeError(msg)

    return result

subclass

Abstract base for cross-package subclass discovery without explicit registration.

DependencySubclass

Abstract base enabling plugin-style subclass discovery across installed packages.

Subclasses declare a discovery scope by overriding the discovery hook, and the base class automatically finds all subclass implementations defined in that scope across every installed package that depends on the root package. The scope may be a single module, to keep discovery narrow, or a whole sub-package, to widen it to a full hierarchy. No explicit registration is required.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

class DependencySubclass(metaclass=DependencySubclassMeta):
    """Abstract base enabling plugin-style subclass discovery across installed packages.

    Subclasses declare a discovery scope by overriding the discovery hook,
    and the base class automatically finds all subclass implementations defined
    in that scope across every installed package that depends on the root
    package. The scope may be a single module, to keep discovery narrow, or a
    whole sub-package, to widen it to a full hierarchy. No explicit
    registration is required.
    """

    def __str__(self) -> str:
        """Return the fully qualified class name of this instance."""
        return str(self.__class__)

    @classmethod
    @abstractmethod
    def discovery_module(cls) -> ModuleType:
        """Return the module or package that scopes discovery of this class.

        Every concrete subclass must override this to declare where its
        implementation classes live. Returning a package widens discovery to
        that package's whole module hierarchy; returning a plain module keeps
        discovery narrow to that single module.

        The base implementation returns `pyrig_runtime.rig`.

        Returns:
            The module or package that scopes the search for concrete
            implementations of this class.
        """
        return rig

    @classmethod
    def sort_key(cls) -> Any:
        """Return the sort key used to order this class relative to peer subclasses.

        Override to sort by priority, numeric position, or any other criterion.
        The default returns the class name, giving alphabetical ordering.

        Returns:
            A value comparable with `<` against the sort keys of other
            subclasses.
        """
        return cls.__name__

    @classmethod
    def leaf(cls) -> type[Self]:
        """Return the single leaf subclass found across dependent packages.

        If no subclasses are found, the class itself is returned.

        Returns:
            The single leaf subclass type, or the class itself if no
            subclasses are found. May be abstract.

        Raises:
            RuntimeError: If more than one leaf subclass is discovered across
                the dependent packages.
        """
        subclasses = cls.subclasses()
        leaf = next(subclasses, cls)
        second = next(subclasses, None)
        if second is None:
            return leaf

        subclasses_dump = json.dumps(
            [fully_qualified_name(subcls) for subcls in (leaf, second, *subclasses)],
            indent=4,
        )
        msg = f"""Multiple leaf subclasses found for {cls}.
Defining multiple leaf subclasses is ambiguous.
This can happen if more than one leaf subclass is defined
across all the dependent packages.

Found subclasses:
{subclasses_dump}"""
        raise RuntimeError(msg)

    @classmethod
    def concrete_subclasses(cls) -> Iterator[type[Self]]:
        """Yield all concrete leaf subclasses discovered across dependent packages.

        Yields:
            Non-abstract leaf subclass types.
        """
        return discard_abstract_classes(cls.subclasses())

    @classmethod
    def subclasses(cls) -> Iterator[type[Self]]:
        """Yield all subclasses discovered across installed dependent packages.

        Only leaf-level subclasses are yielded; any intermediate parent classes
        that also appear in the result set are omitted.

        Yields:
            Leaf subclass types.
        """
        return discard_parent_classes(
            discover_subclasses_across_dependencies(
                cls,
                module=cls.discovery_module(),
            )
        )

    @classmethod
    def subclasses_sorted(cls, subclasses: Iterable[type[Self]]) -> list[type[Self]]:
        """Sort the given subclasses using each subclass's `sort_key()`.

        Does not perform any discovery.

        Args:
            subclasses: Subclass types to sort.

        Returns:
            The same subclass types sorted by their `sort_key()`.
        """
        return sorted(subclasses, key=lambda subclass: subclass.sort_key())

__str__()

Return the fully qualified class name of this instance.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

def __str__(self) -> str:
    """Return the fully qualified class name of this instance."""
    return str(self.__class__)

concrete_subclasses() classmethod

Yield all concrete leaf subclasses discovered across dependent packages.

Yields:

Type	Description
type[Self]	Non-abstract leaf subclass types.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def concrete_subclasses(cls) -> Iterator[type[Self]]:
    """Yield all concrete leaf subclasses discovered across dependent packages.

    Yields:
        Non-abstract leaf subclass types.
    """
    return discard_abstract_classes(cls.subclasses())

discovery_module() abstractmethod classmethod

Return the module or package that scopes discovery of this class.

Every concrete subclass must override this to declare where its implementation classes live. Returning a package widens discovery to that package's whole module hierarchy; returning a plain module keeps discovery narrow to that single module.

The base implementation returns pyrig_runtime.rig.

Returns:

Type	Description
ModuleType	The module or package that scopes the search for concrete
ModuleType	implementations of this class.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
@abstractmethod
def discovery_module(cls) -> ModuleType:
    """Return the module or package that scopes discovery of this class.

    Every concrete subclass must override this to declare where its
    implementation classes live. Returning a package widens discovery to
    that package's whole module hierarchy; returning a plain module keeps
    discovery narrow to that single module.

    The base implementation returns `pyrig_runtime.rig`.

    Returns:
        The module or package that scopes the search for concrete
        implementations of this class.
    """
    return rig

leaf() classmethod

Return the single leaf subclass found across dependent packages.

If no subclasses are found, the class itself is returned.

Returns:

Type	Description
type[Self]	The single leaf subclass type, or the class itself if no
type[Self]	subclasses are found. May be abstract.

Raises:

Type	Description
RuntimeError	If more than one leaf subclass is discovered across the dependent packages.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

    @classmethod
    def leaf(cls) -> type[Self]:
        """Return the single leaf subclass found across dependent packages.

        If no subclasses are found, the class itself is returned.

        Returns:
            The single leaf subclass type, or the class itself if no
            subclasses are found. May be abstract.

        Raises:
            RuntimeError: If more than one leaf subclass is discovered across
                the dependent packages.
        """
        subclasses = cls.subclasses()
        leaf = next(subclasses, cls)
        second = next(subclasses, None)
        if second is None:
            return leaf

        subclasses_dump = json.dumps(
            [fully_qualified_name(subcls) for subcls in (leaf, second, *subclasses)],
            indent=4,
        )
        msg = f"""Multiple leaf subclasses found for {cls}.
Defining multiple leaf subclasses is ambiguous.
This can happen if more than one leaf subclass is defined
across all the dependent packages.

Found subclasses:
{subclasses_dump}"""
        raise RuntimeError(msg)

sort_key() classmethod

Return the sort key used to order this class relative to peer subclasses.

Override to sort by priority, numeric position, or any other criterion. The default returns the class name, giving alphabetical ordering.

Returns:

Type	Description
Any	A value comparable with < against the sort keys of other
Any	subclasses.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def sort_key(cls) -> Any:
    """Return the sort key used to order this class relative to peer subclasses.

    Override to sort by priority, numeric position, or any other criterion.
    The default returns the class name, giving alphabetical ordering.

    Returns:
        A value comparable with `<` against the sort keys of other
        subclasses.
    """
    return cls.__name__

subclasses() classmethod

Yield all subclasses discovered across installed dependent packages.

Only leaf-level subclasses are yielded; any intermediate parent classes that also appear in the result set are omitted.

Yields:

Type	Description
type[Self]	Leaf subclass types.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def subclasses(cls) -> Iterator[type[Self]]:
    """Yield all subclasses discovered across installed dependent packages.

    Only leaf-level subclasses are yielded; any intermediate parent classes
    that also appear in the result set are omitted.

    Yields:
        Leaf subclass types.
    """
    return discard_parent_classes(
        discover_subclasses_across_dependencies(
            cls,
            module=cls.discovery_module(),
        )
    )

subclasses_sorted(subclasses) classmethod

Sort the given subclasses using each subclass's sort_key().

Does not perform any discovery.

Parameters:

Name	Type	Description	Default
subclasses	Iterable[type[Self]]	Subclass types to sort.	required

Returns:

Type	Description
list[type[Self]]	The same subclass types sorted by their sort_key().

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def subclasses_sorted(cls, subclasses: Iterable[type[Self]]) -> list[type[Self]]:
    """Sort the given subclasses using each subclass's `sort_key()`.

    Does not perform any discovery.

    Args:
        subclasses: Subclass types to sort.

    Returns:
        The same subclass types sorted by their `sort_key()`.
    """
    return sorted(subclasses, key=lambda subclass: subclass.sort_key())

DependencySubclassMeta

Bases: ABCMeta

Metaclass backing DependencySubclass with the cached I/L properties.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

class DependencySubclassMeta(ABCMeta):
    """Metaclass backing `DependencySubclass` with the cached `I`/`L` properties."""

    def __str__(cls) -> str:
        """Return the fully qualified name of this class."""
        return fully_qualified_name(cls)

    @property
    def I[C: DependencySubclass](cls: type[C]) -> C:  # noqa: E743, N802
        """Return a cached instance of the leaf subclass.

        The instance is created once per class and reused on every subsequent
        access.

        Returns:
            An instance of the leaf subclass, or of the class itself if no
            subclasses exist.

        Raises:
            RuntimeError: If more than one leaf subclass is found.
        """
        if "_instance" not in cls.__dict__:
            cls._instance = cls.L()
        return cls._instance

    @property
    def L[C: DependencySubclass](cls: type[C]) -> type[C]:  # noqa: N802
        """Return the cached leaf subclass type.

        The result is cached per class and reused on every subsequent access.

        Returns:
            The single leaf subclass type, or the class itself if no
            subclasses exist. May be abstract.

        Raises:
            RuntimeError: If more than one leaf subclass is found.
        """
        if "_leaf" not in cls.__dict__:
            cls._leaf = cls.leaf()
        return cls._leaf

I property

Return a cached instance of the leaf subclass.

The instance is created once per class and reused on every subsequent access.

Returns:

Type	Description
C	An instance of the leaf subclass, or of the class itself if no
C	subclasses exist.

Raises:

Type	Description
RuntimeError	If more than one leaf subclass is found.

L property

Return the cached leaf subclass type.

The result is cached per class and reused on every subsequent access.

Returns:

Type	Description
type[C]	The single leaf subclass type, or the class itself if no
type[C]	subclasses exist. May be abstract.

Raises:

Type	Description
RuntimeError	If more than one leaf subclass is found.

__str__()

Return the fully qualified name of this class.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

def __str__(cls) -> str:
    """Return the fully qualified name of this class."""
    return fully_qualified_name(cls)

graph

Abstract directed graph foundation with forward and reverse edge traversal.

DiGraph

Bases: ABC

Abstract directed graph with forward and reverse adjacency tracking.

Subclasses implement build to populate nodes and edges. When a root node is given, the graph is pruned to contain only that node and every node that transitively points to it.

Attributes:

Name	Type	Description
root		The root node passed at construction, or None if none was given.
nodes	set[str]	Set of all node identifiers currently in the graph.
edges	dict[str, set[str]]	Forward adjacency map from each node to its outgoing neighbors.
reverse_edges	dict[str, set[str]]	Reverse adjacency map from each node to its incoming neighbors.

Source code in src/pyrig_runtime/core/graph.py

class DiGraph(ABC):
    """Abstract directed graph with forward and reverse adjacency tracking.

    Subclasses implement `build` to populate nodes and edges. When a `root`
    node is given, the graph is pruned to contain only that node and every node
    that transitively points to it.

    Attributes:
        root: The root node passed at construction, or `None` if none was given.
        nodes: Set of all node identifiers currently in the graph.
        edges: Forward adjacency map from each node to its outgoing neighbors.
        reverse_edges: Reverse adjacency map from each node to its incoming neighbors.
    """

    @classmethod
    @cache
    def cached(cls, *args: Any, **kwargs: Any) -> Self:
        """Return a cached instance, constructing it only on the first call.

        Repeated calls with identical arguments return the same instance
        without rebuilding the graph.

        Args:
            *args: Positional arguments forwarded to the constructor.
            **kwargs: Keyword arguments forwarded to the constructor.

        Returns:
            The cached graph instance for the given arguments.
        """
        return cls(*args, **kwargs)

    def __init__(self, root: str | None = None) -> None:
        """Initialize the directed graph, optionally pruned to the given root."""
        self.root = root
        self.nodes: set[str] = set()
        self.edges: dict[str, set[str]] = {}
        self.reverse_edges: dict[str, set[str]] = {}
        self.build()
        if self.root is not None:
            self.prune(self.root)

    @abstractmethod
    def build(self) -> None:
        """Populate the graph with nodes and edges.

        Called during construction before any pruning occurs. Subclasses
        must add all nodes and edges that belong to the graph.
        """

    def prune(self, root: str) -> None:
        """Retain only the given root and its ancestors, removing all other nodes.

        Keeps `root` and all its ancestors (nodes with a directed path to
        `root`). All other nodes and their associated edges are removed.

        Args:
            root: The node to keep, along with all nodes that have a directed
                path to it.
        """
        keep = self.ancestors(root) | {root}
        self.nodes = keep
        self.edges = {n: self.edges[n] & keep for n in keep}
        self.reverse_edges = {n: self.reverse_edges[n] & keep for n in keep}

    def add_edge(self, source: str, target: str) -> None:
        """Add a directed edge from source to target.

        Creates both nodes if they do not already exist.

        Args:
            source: Edge origin node.
            target: Edge destination node.
        """
        self.add_node(source)
        self.add_node(target)
        self.edges[source].add(target)
        self.reverse_edges[target].add(source)

    def add_node(self, node: str) -> None:
        """Add a node to the graph. No-op if the node already exists."""
        self.nodes.add(node)
        if node not in self.edges:
            self.edges[node] = set()
        if node not in self.reverse_edges:
            self.reverse_edges[node] = set()

    def sorted_ancestors(self, target: str) -> list[str]:
        """Return all ancestors of the target node in topological order.

        Ancestors are nodes that have a directed path to the target. The
        result is ordered so that each ancestor appears after all ancestors
        it has outgoing edges to.

        Args:
            target: Node to find ancestors of.

        Returns:
            List of ancestor node identifiers in topological order.
            Returns an empty list if the target has no ancestors.

        Raises:
            KeyError: If the target node is not in the graph.
            RuntimeError: If the ancestor subgraph contains a cycle, making
                topological sorting impossible.
        """
        return self.topological_sort_subgraph(self.ancestors(target))

    def ancestors(self, target: str) -> set[str]:
        """Find all nodes that have a directed path to the target node.

        The target itself is excluded from the result.

        Args:
            target: Node to find ancestors for.

        Returns:
            Set of all nodes with a directed path to the target, excluding the
            target itself.

        Raises:
            KeyError: If the target node is not in the graph.
        """
        visited: set[str] = set()
        queue: deque[str] = deque(self.reverse_edges[target])

        while queue:
            node = queue.popleft()
            if node not in visited:
                visited.add(node)
                for neighbor in self.reverse_edges[node]:
                    if neighbor not in visited:
                        queue.append(neighbor)

        return visited

    def topological_sort_subgraph(self, nodes: set[str]) -> list[str]:
        """Sort a subset of nodes in topological order.

        If there is an edge from A to B, B appears before A in the result.
        The ordering is deterministic: when multiple nodes are eligible to be
        emitted at the same step, they are emitted in ascending lexicographic
        order.

        Only edges whose both endpoints are in `nodes` are considered; edges
        to or from nodes outside the subset are ignored.

        Args:
            nodes: The subset of nodes to sort.

        Returns:
            List of nodes in topological order, with each node appearing
            after all nodes it has outgoing edges to.

        Raises:
            RuntimeError: If the subgraph contains a cycle, making topological
                sorting impossible.
        """
        out_degree: dict[str, int] = dict.fromkeys(nodes, 0)

        for node in nodes:
            for dependency in self.edges[node]:
                if dependency in nodes:
                    out_degree[node] += 1

        heap: list[str] = [node for node in nodes if out_degree[node] == 0]
        heapq.heapify(heap)
        result: list[str] = []

        while heap:
            node = heapq.heappop(heap)
            result.append(node)

            for dependent in self.reverse_edges[node]:
                if dependent in nodes:
                    out_degree[dependent] -= 1
                    if out_degree[dependent] == 0:
                        heapq.heappush(heap, dependent)

        if len(result) != len(nodes):
            msg = (
                "Graph contains a cycle; topological sort not possible. "
                "This indicates a circular dependency among the following nodes: "
                f"{set(nodes) - set(result)}"
            )
            raise RuntimeError(msg)

        return result

__init__(root=None)

Initialize the directed graph, optionally pruned to the given root.

Source code in src/pyrig_runtime/core/graph.py

def __init__(self, root: str | None = None) -> None:
    """Initialize the directed graph, optionally pruned to the given root."""
    self.root = root
    self.nodes: set[str] = set()
    self.edges: dict[str, set[str]] = {}
    self.reverse_edges: dict[str, set[str]] = {}
    self.build()
    if self.root is not None:
        self.prune(self.root)

add_edge(source, target)

Add a directed edge from source to target.

Creates both nodes if they do not already exist.

Parameters:

Name	Type	Description	Default
source	str	Edge origin node.	required
target	str	Edge destination node.	required

Source code in src/pyrig_runtime/core/graph.py

def add_edge(self, source: str, target: str) -> None:
    """Add a directed edge from source to target.

    Creates both nodes if they do not already exist.

    Args:
        source: Edge origin node.
        target: Edge destination node.
    """
    self.add_node(source)
    self.add_node(target)
    self.edges[source].add(target)
    self.reverse_edges[target].add(source)

add_node(node)

Add a node to the graph. No-op if the node already exists.

Source code in src/pyrig_runtime/core/graph.py

def add_node(self, node: str) -> None:
    """Add a node to the graph. No-op if the node already exists."""
    self.nodes.add(node)
    if node not in self.edges:
        self.edges[node] = set()
    if node not in self.reverse_edges:
        self.reverse_edges[node] = set()

ancestors(target)

Find all nodes that have a directed path to the target node.

The target itself is excluded from the result.

Parameters:

Name	Type	Description	Default
target	str	Node to find ancestors for.	required

Returns:

Type	Description
set[str]	Set of all nodes with a directed path to the target, excluding the
set[str]	target itself.

Raises:

Type	Description
KeyError	If the target node is not in the graph.

Source code in src/pyrig_runtime/core/graph.py

def ancestors(self, target: str) -> set[str]:
    """Find all nodes that have a directed path to the target node.

    The target itself is excluded from the result.

    Args:
        target: Node to find ancestors for.

    Returns:
        Set of all nodes with a directed path to the target, excluding the
        target itself.

    Raises:
        KeyError: If the target node is not in the graph.
    """
    visited: set[str] = set()
    queue: deque[str] = deque(self.reverse_edges[target])

    while queue:
        node = queue.popleft()
        if node not in visited:
            visited.add(node)
            for neighbor in self.reverse_edges[node]:
                if neighbor not in visited:
                    queue.append(neighbor)

    return visited

build() abstractmethod

Populate the graph with nodes and edges.

Called during construction before any pruning occurs. Subclasses must add all nodes and edges that belong to the graph.

Source code in src/pyrig_runtime/core/graph.py

@abstractmethod
def build(self) -> None:
    """Populate the graph with nodes and edges.

    Called during construction before any pruning occurs. Subclasses
    must add all nodes and edges that belong to the graph.
    """

cached(*args, **kwargs) cached classmethod

Return a cached instance, constructing it only on the first call.

Repeated calls with identical arguments return the same instance without rebuilding the graph.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments forwarded to the constructor.	()
**kwargs	Any	Keyword arguments forwarded to the constructor.	{}

Returns:

Type	Description
Self	The cached graph instance for the given arguments.

Source code in src/pyrig_runtime/core/graph.py

@classmethod
@cache
def cached(cls, *args: Any, **kwargs: Any) -> Self:
    """Return a cached instance, constructing it only on the first call.

    Repeated calls with identical arguments return the same instance
    without rebuilding the graph.

    Args:
        *args: Positional arguments forwarded to the constructor.
        **kwargs: Keyword arguments forwarded to the constructor.

    Returns:
        The cached graph instance for the given arguments.
    """
    return cls(*args, **kwargs)

prune(root)

Retain only the given root and its ancestors, removing all other nodes.

Keeps root and all its ancestors (nodes with a directed path to root). All other nodes and their associated edges are removed.

Parameters:

Name	Type	Description	Default
root	str	The node to keep, along with all nodes that have a directed path to it.	required

Source code in src/pyrig_runtime/core/graph.py

def prune(self, root: str) -> None:
    """Retain only the given root and its ancestors, removing all other nodes.

    Keeps `root` and all its ancestors (nodes with a directed path to
    `root`). All other nodes and their associated edges are removed.

    Args:
        root: The node to keep, along with all nodes that have a directed
            path to it.
    """
    keep = self.ancestors(root) | {root}
    self.nodes = keep
    self.edges = {n: self.edges[n] & keep for n in keep}
    self.reverse_edges = {n: self.reverse_edges[n] & keep for n in keep}

sorted_ancestors(target)

Return all ancestors of the target node in topological order.

Ancestors are nodes that have a directed path to the target. The result is ordered so that each ancestor appears after all ancestors it has outgoing edges to.

Parameters:

Name	Type	Description	Default
target	str	Node to find ancestors of.	required

Returns:

Type	Description
list[str]	List of ancestor node identifiers in topological order.
list[str]	Returns an empty list if the target has no ancestors.

Raises:

Type	Description
KeyError	If the target node is not in the graph.
RuntimeError	If the ancestor subgraph contains a cycle, making topological sorting impossible.

Source code in src/pyrig_runtime/core/graph.py

def sorted_ancestors(self, target: str) -> list[str]:
    """Return all ancestors of the target node in topological order.

    Ancestors are nodes that have a directed path to the target. The
    result is ordered so that each ancestor appears after all ancestors
    it has outgoing edges to.

    Args:
        target: Node to find ancestors of.

    Returns:
        List of ancestor node identifiers in topological order.
        Returns an empty list if the target has no ancestors.

    Raises:
        KeyError: If the target node is not in the graph.
        RuntimeError: If the ancestor subgraph contains a cycle, making
            topological sorting impossible.
    """
    return self.topological_sort_subgraph(self.ancestors(target))

topological_sort_subgraph(nodes)

Sort a subset of nodes in topological order.

If there is an edge from A to B, B appears before A in the result. The ordering is deterministic: when multiple nodes are eligible to be emitted at the same step, they are emitted in ascending lexicographic order.

Only edges whose both endpoints are in nodes are considered; edges to or from nodes outside the subset are ignored.

Parameters:

Name	Type	Description	Default
nodes	set[str]	The subset of nodes to sort.	required

Returns:

Type	Description
list[str]	List of nodes in topological order, with each node appearing
list[str]	after all nodes it has outgoing edges to.

Raises:

Type	Description
RuntimeError	If the subgraph contains a cycle, making topological sorting impossible.

Source code in src/pyrig_runtime/core/graph.py

def topological_sort_subgraph(self, nodes: set[str]) -> list[str]:
    """Sort a subset of nodes in topological order.

    If there is an edge from A to B, B appears before A in the result.
    The ordering is deterministic: when multiple nodes are eligible to be
    emitted at the same step, they are emitted in ascending lexicographic
    order.

    Only edges whose both endpoints are in `nodes` are considered; edges
    to or from nodes outside the subset are ignored.

    Args:
        nodes: The subset of nodes to sort.

    Returns:
        List of nodes in topological order, with each node appearing
        after all nodes it has outgoing edges to.

    Raises:
        RuntimeError: If the subgraph contains a cycle, making topological
            sorting impossible.
    """
    out_degree: dict[str, int] = dict.fromkeys(nodes, 0)

    for node in nodes:
        for dependency in self.edges[node]:
            if dependency in nodes:
                out_degree[node] += 1

    heap: list[str] = [node for node in nodes if out_degree[node] == 0]
    heapq.heapify(heap)
    result: list[str] = []

    while heap:
        node = heapq.heappop(heap)
        result.append(node)

        for dependent in self.reverse_edges[node]:
            if dependent in nodes:
                out_degree[dependent] -= 1
                if out_degree[dependent] == 0:
                    heapq.heappush(heap, dependent)

    if len(result) != len(nodes):
        msg = (
            "Graph contains a cycle; topological sort not possible. "
            "This indicates a circular dependency among the following nodes: "
            f"{set(nodes) - set(result)}"
        )
        raise RuntimeError(msg)

    return result

introspection

Runtime introspection primitives for classes, callables, modules, and packages.

classes

Utilities for introspecting and filtering Python classes.

discard_abstract_classes(classes)

Filter out abstract classes from a collection.

A class is considered abstract when it has one or more unimplemented abstract methods and therefore cannot be instantiated directly.

Parameters:

Name	Type	Description	Default
classes	Iterable[type[T]]	Iterable of class types to filter.	required

Yields:

Type	Description
type[T]	Concrete (non-abstract) classes from the input.

Source code in src/pyrig_runtime/core/introspection/classes.py

def discard_abstract_classes[T](classes: Iterable[type[T]]) -> Iterator[type[T]]:
    """Filter out abstract classes from a collection.

    A class is considered abstract when it has one or more unimplemented
    abstract methods and therefore cannot be instantiated directly.

    Args:
        classes: Iterable of class types to filter.

    Yields:
        Concrete (non-abstract) classes from the input.
    """
    return (cls for cls in classes if not inspect.isabstract(cls))

discard_parent_classes(classes)

Yield only leaf classes, removing any ancestors present in the collection.

A class is kept only when no other class in the collection is a strict subclass of it. The original iterable is not modified.

Parameters:

Name	Type	Description	Default
classes	Iterable[type[T]]	Iterable of class types to filter.	required

Yields:

Type	Description
type[T]	Classes that have no subclasses present in the same collection.

Source code in src/pyrig_runtime/core/introspection/classes.py

def discard_parent_classes[T](
    classes: Iterable[type[T]],
) -> Iterator[type[T]]:
    """Yield only leaf classes, removing any ancestors present in the collection.

    A class is kept only when no other class in the collection is a strict
    subclass of it. The original iterable is not modified.

    Args:
        classes: Iterable of class types to filter.

    Yields:
        Classes that have no subclasses present in the same collection.
    """
    classes = set(classes)
    return (
        cls
        for cls in classes
        if not any(
            issubclass(candidate, cls) for candidate in classes if candidate is not cls
        )
    )

discover_subclasses(cls)

Discover all transitive subclasses of cls currently loaded in memory.

Does not trigger any imports, so only subclasses from already-imported modules are included in the result.

Parameters:

Name	Type	Description	Default
cls	type[T]	Base class to find subclasses of.	required

Returns:

Type	Description
set[type[T]]	Set of all transitive subclass types, excluding cls itself.

Source code in src/pyrig_runtime/core/introspection/classes.py

def discover_subclasses[T](cls: type[T]) -> set[type[T]]:
    """Discover all transitive subclasses of `cls` currently loaded in memory.

    Does not trigger any imports, so only subclasses from already-imported
    modules are included in the result.

    Args:
        cls: Base class to find subclasses of.

    Returns:
        Set of all transitive subclass types, excluding `cls` itself.
    """
    direct = cls.__subclasses__()
    subclasses = set(direct)
    for subclass in direct:
        subclasses.update(discover_subclasses(subclass))
    return subclasses

functions

Utilities for Python functions.

module_functions(module)

Yield all functions defined directly in a module.

Excludes objects imported from other modules.

Parameters:

Name	Type	Description	Default
module	ModuleType	Module to inspect.	required

Yields:

Type	Description
Callable[..., Any]	Each function defined in module.

Source code in src/pyrig_runtime/core/introspection/functions.py

def module_functions(
    module: ModuleType,
) -> Iterator[Callable[..., Any]]:
    """Yield all functions defined directly in a module.

    Excludes objects imported from other modules.

    Args:
        module: Module to inspect.

    Yields:
        Each function defined in `module`.
    """
    for _name, func in obj_members(module):
        unwrapped_func = unwrap_obj(func)
        if (
            inspect.isfunction(unwrapped_func)
            and inspect.getmodule(unwrapped_func) is module
        ):
            yield func

inspection

Utilities for inspecting Python objects.

obj_members(obj, predicate=None)

Yield the members of an object as name-value pairs without invoking descriptors.

Members are read statically, so properties with side effects are not triggered. __annotate__ and __annotate_func__ are always excluded from the result.

Parameters:

Name	Type	Description	Default
obj	Any	Object to inspect (class, module, or any Python object).	required
predicate	Callable[[Any], bool] | None	Optional filter. When given, only members for which it returns True are included.	None

Returns:

Type	Description
Iterator[tuple[str, Any]]	(name, value) pairs for the matching members of obj.

Source code in src/pyrig_runtime/core/introspection/inspection.py

def obj_members(
    obj: Any, predicate: Callable[[Any], bool] | None = None
) -> Iterator[tuple[str, Any]]:
    """Yield the members of an object as name-value pairs without invoking descriptors.

    Members are read statically, so properties with side effects are not
    triggered. `__annotate__` and `__annotate_func__` are always excluded
    from the result.

    Args:
        obj: Object to inspect (class, module, or any Python object).
        predicate: Optional filter. When given, only members for which it
            returns `True` are included.

    Returns:
        `(name, value)` pairs for the matching members of `obj`.
    """
    return (
        (member, value)
        for member, value in inspect.getmembers_static(obj, predicate=predicate)
        if member not in {"__annotate__", "__annotate_func__"}
    )

unwrap_obj(obj)

Unwrap a Python object to its innermost underlying object.

Recognizes properties, bound methods, classmethods, staticmethods, and functools.wraps-style decorator chains. All layers are removed, not just the outermost one.

Parameters:

Name	Type	Description	Default
obj	Any	Python object to unwrap.	required

Returns:

Type	Description
Any	The innermost Python object after all wrapping layers have been removed.

Source code in src/pyrig_runtime/core/introspection/inspection.py

def unwrap_obj(obj: Any) -> Any:
    """Unwrap a Python object to its innermost underlying object.

    Recognizes properties, bound methods, classmethods, staticmethods, and
    `functools.wraps`-style decorator chains. All layers are removed, not just
    the outermost one.

    Args:
        obj: Python object to unwrap.

    Returns:
        The innermost Python object after all wrapping layers have been removed.
    """
    prev = None
    while prev is not obj:
        prev = obj
        if func := getattr(obj, "__func__", None):
            obj = func
        if fget := getattr(obj, "fget", None):
            obj = fget
        obj = inspect.unwrap(obj)
    return obj

modules

Utilities for Python modules.

import_modules(module_names)

Import multiple modules by name, lazily.

Modules are imported on demand as the result is iterated, not eagerly.

Parameters:

Name	Type	Description	Default
module_names	Iterable[str]	Dotted module names to import.	required

Yields:

Type	Description
ModuleType	Each imported module, in the order the names are iterated.

Source code in src/pyrig_runtime/core/introspection/modules.py

def import_modules(module_names: Iterable[str]) -> Iterator[ModuleType]:
    """Import multiple modules by name, lazily.

    Modules are imported on demand as the result is iterated, not eagerly.

    Args:
        module_names: Dotted module names to import.

    Yields:
        Each imported module, in the order the names are iterated.
    """
    return (import_module(name) for name in module_names)

iter_modules(package)

Import and yield each direct child of a package, in discovery order.

Only the immediate children are visited; nested sub-packages are not recursed into.

Parameters:

Name	Type	Description	Default
package	ModuleType	Package to iterate. Must have a __path__ attribute (i.e., must be a package, not a plain module).	required

Yields:

Type	Description
ModuleType	(module, is_package) pairs where module is the imported child and
bool	is_package is True when the child is itself a sub-package.

Note

Importing each child is a deliberate side effect — any module-level code in those children executes on demand as the iterator is consumed.

Source code in src/pyrig_runtime/core/introspection/modules.py

def iter_modules(package: ModuleType) -> Iterator[tuple[ModuleType, bool]]:
    """Import and yield each direct child of a package, in discovery order.

    Only the immediate children are visited; nested sub-packages are not
    recursed into.

    Args:
        package: Package to iterate. Must have a `__path__` attribute
            (i.e., must be a package, not a plain module).

    Yields:
        `(module, is_package)` pairs where `module` is the imported child and
        `is_package` is `True` when the child is itself a sub-package.

    Note:
        Importing each child is a deliberate side effect — any module-level
        code in those children executes on demand as the iterator is consumed.
    """
    for _finder, name, is_package in pkgutil_iter_modules(
        package.__path__, prefix=package.__name__ + "."
    ):
        mod = import_module(name)
        yield mod, is_package

replace_root_module(module, root, default=MISSING)

replace_root_module(
    module: ModuleType, root: str
) -> ModuleType

replace_root_module(
    module: ModuleType, root: str, default: T
) -> ModuleType | T

Import the equivalent module under a different root package.

Replaces the first dotted segment of module's name with root and attempts to import the resulting module name. Later segments are left untouched even if they happen to share the old root's name.

Parameters:

Name	Type	Description	Default
module	ModuleType	Module whose root segment should be swapped.	required
root	str	Root package name to substitute in.	required
default	Any	Value to return if the import fails. If not provided, the exception propagates unchanged.	MISSING

Returns:

Type	Description
ModuleType | Any	The imported module at the equivalent sub-path under root, or
ModuleType | Any	default if the import fails and default was provided.

Example

from some_package.subpackage import module replace_root_module(module, "other_package")

Source code in src/pyrig_runtime/core/introspection/modules.py

def replace_root_module(
    module: ModuleType, root: str, default: Any = MISSING
) -> ModuleType | Any:
    """Import the equivalent module under a different root package.

    Replaces the first dotted segment of `module`'s name with `root` and
    attempts to import the resulting module name. Later segments are left
    untouched even if they happen to share the old root's name.

    Args:
        module: Module whose root segment should be swapped.
        root: Root package name to substitute in.
        default: Value to return if the import fails. If not provided,
            the exception propagates unchanged.

    Returns:
        The imported module at the equivalent sub-path under `root`, or
        `default` if the import fails and `default` was provided.

    Example:
        >>> from some_package.subpackage import module
        >>> replace_root_module(module, "other_package")
        <module 'other_package.subpackage.module'
         from '/path/to/other_package/subpackage/module.py'>
    """
    return safe_import_module(replace_root_module_name(module, root), default=default)

replace_root_module_name(module, root)

Return the equivalent module name under a different root package.

Replaces the first dotted segment of module.__name__ with root. Later segments are left untouched even if they happen to share the old root's name.

Parameters:

Name	Type	Description	Default
module	ModuleType	Module whose root segment should be swapped.	required
root	str	Root package name to substitute in.	required

Returns:

Type	Description
str	The equivalent dotted module name under root.

Example

replace_root_module_name(module, "other_package") 'other_package.subpackage.module'

Source code in src/pyrig_runtime/core/introspection/modules.py

def replace_root_module_name(module: ModuleType, root: str) -> str:
    """Return the equivalent module name under a different root package.

    Replaces the first dotted segment of `module.__name__` with `root`. Later
    segments are left untouched even if they happen to share the old root's
    name.

    Args:
        module: Module whose root segment should be swapped.
        root: Root package name to substitute in.

    Returns:
        The equivalent dotted module name under `root`.

    Example:
        >>> replace_root_module_name(module, "other_package")
        'other_package.subpackage.module'
    """
    return module.__name__.replace(root_module_name(module), root, 1)

root_module(module)

Import and return the top-level package of the given module.

For a module named "package.subpackage.module", the module corresponding to "package" is returned. For a top-level module with no dots in its name, the module for that same name is returned.

Parameters:

Name	Type	Description	Default
module	ModuleType	Module to resolve the root package for.	required

Returns:

Type	Description
ModuleType	The module corresponding to the first segment of the dotted name.

Example

from some_package.subpackage import module root_module(module)

Source code in src/pyrig_runtime/core/introspection/modules.py

def root_module(module: ModuleType) -> ModuleType:
    """Import and return the top-level package of the given module.

    For a module named `"package.subpackage.module"`, the module corresponding
    to `"package"` is returned. For a top-level module with no dots in its name,
    the module for that same name is returned.

    Args:
        module: Module to resolve the root package for.

    Returns:
        The module corresponding to the first segment of the dotted name.

    Example:
        >>> from some_package.subpackage import module
        >>> root_module(module)
        <module 'some_package' from '/path/to/some_package/__init__.py'>
    """
    return import_module(root_module_name(module))

root_module_name(module)

Return the name of the top-level package of the given module.

For a module named "package.subpackage.module", the string "package" is returned. For a top-level module with no dots in its name, that same name is returned.

Parameters:

Name	Type	Description	Default
module	ModuleType	Module to resolve the root package name for.	required

Returns:

Type	Description
str	The first segment of the dotted module name.

Example

from some_package.subpackage import module root_module_name(module) 'some_package'

Source code in src/pyrig_runtime/core/introspection/modules.py

def root_module_name(module: ModuleType) -> str:
    """Return the name of the top-level package of the given module.

    For a module named `"package.subpackage.module"`, the string `"package"`
    is returned. For a top-level module with no dots in its name, that same
    name is returned.

    Args:
        module: Module to resolve the root package name for.

    Returns:
        The first segment of the dotted module name.

    Example:
        >>> from some_package.subpackage import module
        >>> root_module_name(module)
        'some_package'
    """
    return module.__name__.split(".", 1)[0]

safe_import_module(module_name, package=None, *, default=MISSING, exceptions=(Exception,))

safe_import_module(
    module_name: str,
    package: str | None = ...,
    *,
    exceptions: tuple[type[BaseException], ...] = ...,
) -> ModuleType

safe_import_module(
    module_name: str,
    package: str | None = ...,
    *,
    default: T,
    exceptions: tuple[type[BaseException], ...] = ...,
) -> ModuleType | T

Import a module by name, with an optional fallback on failure.

Any Exception raised during import — not just ImportError — is handled, so errors at module level (e.g., ValueError raised on import) are also covered.

Parameters:

Name	Type	Description	Default
module_name	str	Dotted module name (e.g., "package.subpackage.module").	required
package	str | None	Anchor package for relative imports, forwarded to import_module.	None
default	Any	Value to return if the import raises. If not provided, the exception propagates unchanged.	MISSING
exceptions	tuple[type[BaseException], ...]	Tuple of exception types to catch. Defaults to (Exception,).	(Exception,)

Returns:

Type	Description
ModuleType | Any	The imported module, or default if an exception is raised and
ModuleType | Any	default was provided.

Source code in src/pyrig_runtime/core/introspection/modules.py

def safe_import_module(
    module_name: str,
    package: str | None = None,
    *,
    default: Any = MISSING,
    exceptions: tuple[type[BaseException], ...] = (Exception,),
) -> ModuleType | Any:
    """Import a module by name, with an optional fallback on failure.

    Any `Exception` raised during import — not just `ImportError` — is
    handled, so errors at module level (e.g., `ValueError` raised on
    import) are also covered.

    Args:
        module_name: Dotted module name (e.g., `"package.subpackage.module"`).
        package: Anchor package for relative imports, forwarded to `import_module`.
        default: Value to return if the import raises. If not provided,
            the exception propagates unchanged.
        exceptions: Tuple of exception types to catch. Defaults to `(Exception,)`.

    Returns:
        The imported module, or `default` if an exception is raised and
        `default` was provided.
    """
    return safe_call(
        import_module,
        module_name,
        package,
        default=default,
        exceptions=exceptions,
    )

packages

Subclass discovery scoped by a module name prefix.

discover_subclasses_across_module(cls, module)

Discover all subclasses of cls whose module name starts with module.

When module is a package, its full module hierarchy is imported first, so subclasses are found in its sub-modules at any nesting depth, including ones not yet imported when this is called. When module is a plain module, no import walking happens and only subclasses already defined are considered.

A subclass matches when the dotted name of its defining module starts with module.__name__. This includes module itself, every sub-module of it, and any sibling module whose name shares that prefix (for example, scope pkg.foo also matches pkg.foobar). Choosing a plain module as the scope therefore narrows discovery to that one file, while choosing a package widens it to the whole hierarchy.

Parameters:

Name	Type	Description	Default
cls	type[T]	Base class whose subclasses should be discovered.	required
module	ModuleType	Module or package whose dotted name prefixes the discovery scope.	required

Returns:

Type	Description
set[type[T]]	Set of all subclass types of cls whose defining module name starts
set[type[T]]	with module.__name__, excluding cls itself.

Source code in src/pyrig_runtime/core/introspection/packages.py

def discover_subclasses_across_module[T](
    cls: type[T],
    module: ModuleType,
) -> set[type[T]]:
    """Discover all subclasses of `cls` whose module name starts with `module`.

    When `module` is a package, its full module hierarchy is imported first, so
    subclasses are found in its sub-modules at any nesting depth, including ones
    not yet imported when this is called. When `module` is a plain module, no
    import walking happens and only subclasses already defined are considered.

    A subclass matches when the dotted name of its defining module starts with
    `module.__name__`. This includes `module` itself, every sub-module of it,
    and any sibling module whose name shares that prefix (for example, scope
    `pkg.foo` also matches `pkg.foobar`). Choosing a plain module as the scope
    therefore narrows discovery to that one file, while choosing a package
    widens it to the whole hierarchy.

    Args:
        cls: Base class whose subclasses should be discovered.
        module: Module or package whose dotted name prefixes the discovery
            scope.

    Returns:
        Set of all subclass types of `cls` whose defining module name starts
        with `module.__name__`, excluding `cls` itself.
    """
    if is_package(module):
        register_package_modules(module)
    subclasses = discover_subclasses(cls)
    return {
        subclass
        for subclass in subclasses
        if subclass.__module__.startswith(module.__name__)
    }

is_package(module)

Return True if module is a package rather than a plain module.

Source code in src/pyrig_runtime/core/introspection/packages.py

def is_package(module: ModuleType) -> bool:
    """Return `True` if `module` is a package rather than a plain module."""
    return hasattr(module, "__path__")

register_package_modules(package) cached

Ensure all modules in a package hierarchy are imported.

Parameters:

Name	Type	Description	Default
package	ModuleType	Root package whose entire module hierarchy will be imported.	required

Note

Cached per package — subsequent calls with the same package do nothing.

Source code in src/pyrig_runtime/core/introspection/packages.py

@cache
def register_package_modules(package: ModuleType) -> None:
    """Ensure all modules in a package hierarchy are imported.

    Args:
        package: Root package whose entire module hierarchy will be imported.

    Note:
        Cached per package — subsequent calls with the same package do
        nothing.
    """
    _ = tuple(walk_package(package))

walk_package(package)

Walk all modules in a package hierarchy, recursing into sub-packages.

Importing each visited module is a side effect of iteration. The root package itself is not yielded.

Parameters:

Name	Type	Description	Default
package	ModuleType	Root package to start traversal from.	required

Yields:

Type	Description
ModuleType	(module, is_package) pairs for each visited module, where
bool	is_package is True when the module is itself a sub-package.

Source code in src/pyrig_runtime/core/introspection/packages.py

def walk_package(package: ModuleType) -> Iterator[tuple[ModuleType, bool]]:
    """Walk all modules in a package hierarchy, recursing into sub-packages.

    Importing each visited module is a side effect of iteration. The root
    `package` itself is not yielded.

    Args:
        package: Root package to start traversal from.

    Yields:
        `(module, is_package)` pairs for each visited module, where
        `is_package` is `True` when the module is itself a sub-package.
    """
    for module, is_package in iter_modules(package):
        if is_package:
            yield module, True
            yield from walk_package(module)
        else:
            yield module, False

strings

String conversion utilities for Python package naming conventions.

dependency_requirement_as_module_name(dep_req)

Extract the importable module name from a dependency requirement string.

Version specifiers, extras notation, and any other non-name characters are stripped. Hyphens in the package name are normalized to underscores.

Parameters:

Name	Type	Description	Default
dep_req	str	A dependency requirement string ( e.g., "requests>=2.0,<3.0" or "my-package[extra]==1.0.0"or "some.package==1.0.0"	required

Returns:

Type	Description
str	The package name in snake_case ( e.g., "requests", "my_package", "some.package"
str	).

Source code in src/pyrig_runtime/core/strings.py

def dependency_requirement_as_module_name(dep_req: str) -> str:
    """Extract the importable module name from a dependency requirement string.

    Version specifiers, extras notation, and any other non-name characters are
    stripped. Hyphens in the package name are normalized to underscores.

    Args:
        dep_req: A dependency requirement string (
            e.g., `"requests>=2.0,<3.0"` or
            `"my-package[extra]==1.0.0"`or
            `"some.package==1.0.0"`
        ).

    Returns:
        The package name in snake_case (
            e.g., `"requests"`, `"my_package"`, `"some.package"`
        ).
    """
    return kebab_to_snake_case(
        dependency_requirement_split_pattern().split(dep_req, maxsplit=1)[0]
    )

dependency_requirement_split_pattern()

Return a compiled regex pattern matching characters outside a package name.

Returns:

Type	Description
Pattern[str]	A pattern matching any character that is not alphanumeric, an
Pattern[str]	underscore, a hyphen, or a period.

Source code in src/pyrig_runtime/core/strings.py

def dependency_requirement_split_pattern() -> re.Pattern[str]:
    """Return a compiled regex pattern matching characters outside a package name.

    Returns:
        A pattern matching any character that is not alphanumeric, an
        underscore, a hyphen, or a period.
    """
    return re.compile(r"[^a-zA-Z0-9_.-]")

distribution_summary(name)

Return the summary recorded in an installed distribution's metadata.

This function assumes that the package is installed and its metadata has a "Summary" field.

Parameters:

Name	Type	Description	Default
name	str	Name of an installed distribution (e.g. "requests").	required

Returns:

Type	Description
str	The distribution's summary description.

Source code in src/pyrig_runtime/core/strings.py

def distribution_summary(name: str) -> str:
    """Return the summary recorded in an installed distribution's metadata.

    This function assumes that the package is installed and its
    metadata has a "Summary" field.

    Args:
        name: Name of an installed distribution (e.g. `"requests"`).

    Returns:
        The distribution's summary description.
    """
    return metadata(name)["Summary"]

fully_qualified_name(obj)

Return the fully qualified name of a callable.

The returned name consists of the callable's module and qualified name, preserving any enclosing classes or functions. E.g., for a method foo in class Bar in module baz, the fully qualified name is "baz.Bar.foo".

Parameters:

Name	Type	Description	Default
obj	MethodType | FunctionType | type	The callable (function, method, or class).	required

Returns:

Type	Description
str	The callable's fully qualified name.

Source code in src/pyrig_runtime/core/strings.py

def fully_qualified_name(obj: MethodType | FunctionType | type) -> str:
    """Return the fully qualified name of a callable.

    The returned name consists of the callable's module and qualified name,
    preserving any enclosing classes or functions.
    E.g., for a method `foo` in class `Bar` in module `baz`, the fully qualified
    name is `"baz.Bar.foo"`.

    Args:
        obj: The callable (function, method, or class).

    Returns:
        The callable's fully qualified name.
    """
    return f"{obj.__module__}.{obj.__qualname__}"

kebab_to_snake_case(value)

Convert a kebab-case string to snake_case, replacing hyphens with underscores.

Source code in src/pyrig_runtime/core/strings.py

def kebab_to_snake_case(value: str) -> str:
    """Convert a kebab-case string to snake_case, replacing hyphens with underscores."""
    return value.replace("-", "_")

snake_to_kebab_case(value)

Convert a snake_case string to kebab-case, replacing underscores with hyphens.

Source code in src/pyrig_runtime/core/strings.py

def snake_to_kebab_case(value: str) -> str:
    """Convert a snake_case string to kebab-case, replacing underscores with hyphens."""
    return value.replace("_", "-")

wrappers

Wrapper utilities.

safe_call(func, *args, default=MISSING, exceptions=(Exception,), **kwargs)

safe_call(
    func: Callable[..., T],
    *args: Any,
    default: D,
    exceptions: tuple[type[BaseException], ...] = ...,
    **kwargs: Any,
) -> T | D

safe_call(
    func: Callable[..., T],
    *args: Any,
    exceptions: tuple[type[BaseException], ...] = ...,
    **kwargs: Any,
) -> T

Call func, returning default on failure or re-raising if none is given.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Callable to invoke.	required
*args	Any	Positional arguments forwarded to func.	()
default	Any	Value to return when a caught exception is raised. If not provided, any caught exception propagates unchanged.	MISSING
exceptions	tuple[type[BaseException], ...]	Exception types to catch. Defaults to (Exception,).	(Exception,)
**kwargs	Any	Keyword arguments forwarded to func.	{}

Returns:

Type	Description
Any	The return value of func(*args, **kwargs), or default if an
Any	exception is caught and default was provided.

Source code in src/pyrig_runtime/core/wrappers.py

def safe_call(
    func: Callable[..., Any],
    *args: Any,
    default: Any = MISSING,
    exceptions: tuple[type[BaseException], ...] = (Exception,),
    **kwargs: Any,
) -> Any:
    """Call `func`, returning `default` on failure or re-raising if none is given.

    Args:
        func: Callable to invoke.
        *args: Positional arguments forwarded to `func`.
        default: Value to return when a caught exception is raised. If not
            provided, any caught exception propagates unchanged.
        exceptions: Exception types to catch. Defaults to `(Exception,)`.
        **kwargs: Keyword arguments forwarded to `func`.

    Returns:
        The return value of `func(*args, **kwargs)`, or `default` if an
        exception is caught and `default` was provided.
    """
    try:
        return func(*args, **kwargs)
    except exceptions:
        if default is MISSING:
            raise
        return default

rig

Convention-based namespace that dependent projects mirror to extend behavior.

cli

CLI subsystem for pyrig-runtime-based projects.

Dependent projects inherit a runnable CLI entry point and can extend it with project-specific commands alongside shared built-in commands.

cli

Typer CLI application builder with cross-package command discovery.

CLI

Bases: DependencySubclass

Typer application builder for pyrig-runtime-based projects.

Builds and runs the command-line application for any project that depends on pyrig-runtime. A dependent project may subclass CLI to override any step of the build to fit its needs.

Source code in src/pyrig_runtime/rig/cli/cli.py

class CLI(DependencySubclass):
    """Typer application builder for pyrig-runtime-based projects.

    Builds and runs the command-line application for any project that depends on
    pyrig-runtime. A dependent project may subclass `CLI` to override any step of
    the build to fit its needs.
    """

    @classmethod
    def discovery_module(cls) -> ModuleType:
        """Return the `pyrig_runtime.rig.cli.cli` module."""
        return sys.modules[__name__]

    def run(self) -> None:
        """Build and invoke the Typer application."""
        self.app()()

    def app(self) -> typer.Typer:
        """Build a fully configured Typer application."""
        app = self.base_app()
        return self.build_app(app)

    def base_app(self) -> typer.Typer:
        """Create an empty base Typer application.

        Returns:
            A new Typer app configured to show help when invoked without
            arguments.
        """
        return typer.Typer(**self.base_app_kwargs())

    def base_app_kwargs(self) -> dict[str, Any]:
        """Return keyword arguments for creating the base Typer application.

        This base configuration makes sure that calling the CLI without
        any arguments will display the help message, and that the help text
        is the same as the description of the invoking project.

        Returns:
            A dictionary of keyword arguments to pass to `typer.Typer`.
        """
        return {
            "no_args_is_help": True,
            "help": distribution_summary(self.project_name()),
        }

    def build_app(self, app: typer.Typer) -> typer.Typer:
        """Register the callback and all commands onto the given app.

        Args:
            app: The Typer app to populate.

        Returns:
            The same app instance, now fully configured.
        """
        self.register_callback(app)
        self.register_subcommands(app)
        self.register_shared_subcommands(app)
        return app

    def register_callback(self, app: typer.Typer) -> None:
        """Attach the verbosity callback to the given app.

        Args:
            app: The Typer app to attach the callback to.
        """
        app.callback()(self.callback)

    def callback(
        self,
        verbose: int = typer.Option(
            0,
            "--verbose",
            "-v",
            count=True,
            help="Increase verbosity: -v (DEBUG), -vv (modules), -vvv (timestamps)",
        ),
        quiet: int = typer.Option(
            0,
            "--quiet",
            "-q",
            count=True,
            help="Decrease verbosity: -q (WARNING), -qq (ERROR), -qqq (CRITICAL)",
        ),
    ) -> None:
        """Apply the verbosity options for the current invocation.

        Args:
            verbose: Number of times verbosity was increased (e.g. via `-v`).
            quiet: Number of times verbosity was decreased (e.g. via `-q`).
        """
        self.configure_logging(verbose, quiet)

    def configure_logging(self, verbose: int, quiet: int) -> None:
        """Configure the logging level and format for the current invocation.

        Each increment of `verbose` lowers the log level by one step (toward
        DEBUG); each increment of `quiet` raises it by one step (toward
        CRITICAL). The format also expands at higher verbosity, adding module
        names at two increments and timestamps at three.

        The log level is intentionally unclamped to potentially support
        custom log levels.

        Args:
            verbose: Number of times verbosity was increased (e.g. via `-v`).
            quiet: Number of times verbosity was decreased (e.g. via `-q`).

        Note:
            Uses `logging.basicConfig` with `force=True` to ensure that the
            configuration is applied even if logging has already been configured
            by the calling project or other dependencies.
        """
        level = logging.INFO
        step = logging.INFO - logging.DEBUG
        level -= step * verbose
        level += step * quiet

        verbose_names = 1
        verbose_modules = verbose_names + 1
        verbose_timestamps = verbose_modules + 1

        if verbose >= verbose_timestamps:
            fmt = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
        elif verbose == verbose_modules:
            fmt = "%(levelname)s [%(name)s] %(message)s"
        elif verbose == verbose_names:
            fmt = "%(levelname)s: %(message)s"
        else:
            fmt = "%(message)s"

        logging.basicConfig(level=level, format=fmt, force=True)

    def register_subcommands(self, app: typer.Typer) -> None:
        """Discover and register project-specific commands for the invoking project.

        Any dependent project can define its own CLI commands by adding
        functions or `typer.Typer` groups to `<package>.rig.cli.subcommands`.

        Args:
            app: The Typer app to register the commands onto.

        Note:
            If the invoking project's subcommands module cannot be imported,
            registration is silently skipped.
        """
        subcommands_module = replace_root_module(
            subcommands, root=self.package_name(), default=None
        )

        if subcommands_module is None:
            return

        self.register_direct_subcommands(app=app, module=subcommands_module)
        self.register_subcommand_groups(app=app, module=subcommands_module)

    def register_shared_subcommands(self, app: typer.Typer) -> None:
        """Discover and register shared commands from pyrig-runtime and its dependents.

        Args:
            app: The Typer app to register the commands onto.

        Note:
            Commands are registered in dependency order (pyrig-runtime first,
            then dependent packages in topological order). When two packages
            define a command with the same name, the last registration takes
            precedence.
        """
        for shared_subcommands_module in chain(
            (shared_subcommands,),
            discover_equivalent_modules_across_dependents(
                shared_subcommands,
            ),
        ):
            self.register_direct_subcommands(app=app, module=shared_subcommands_module)
            self.register_subcommand_groups(app=app, module=shared_subcommands_module)

    def register_direct_subcommands(self, app: typer.Typer, module: ModuleType) -> None:
        """Register every function defined in a module as a top-level command.

        Adds each function found directly in `module` to `app` as a flat Typer
        command. Imported functions are excluded.

        Args:
            app: The Typer app to register the commands onto.
            module: The subcommands module to scan for command functions.
        """
        for func in module_functions(module):
            app.command()(func)

    def register_subcommand_groups(self, app: typer.Typer, module: ModuleType) -> None:
        """Register every `typer.Typer` instance in a module as a named command group.

        Attaches each `typer.Typer` found in the module's namespace to `app`,
        using the kebab-case form of the attribute name as the group name.

        Args:
            app: The Typer app to register the command groups onto.
            module: The subcommands module to scan for group instances.
        """
        for name, group in self.module_subcommand_groups(module).items():
            app.add_typer(group, name=name)

    def module_subcommand_groups(self, module: ModuleType) -> dict[str, typer.Typer]:
        """Return the Typer command groups found in a subcommands module.

        Scans the module's namespace for `typer.Typer` instances. Both natively
        defined and imported instances are included. The key for each entry is
        the kebab-case form of the attribute name it is bound to.

        Args:
            module: The subcommands module to scan.

        Returns:
            Mapping of kebab-case attribute name to the `typer.Typer` instance
            bound to it.
        """
        return {
            snake_to_kebab_case(name): obj
            for name, obj in vars(module).items()
            if isinstance(obj, typer.Typer)
        }

    def package_name(self) -> str:
        """Return the snake_case package name of the invoking project.

        For example, if the project is invoked as `uv run my-project`, the
        package name is `my_project`.
        """
        return kebab_to_snake_case(self.project_name())

    def project_name(self) -> str:
        """Return the stem of `sys.argv[0]` as the invoking project name.

        When a project is invoked through a registered console-script entry point
        (e.g. `uv run my-project`), `sys.argv[0]` is the path to that script, so
        its stem is the project name as it was registered.
        """
        return Path(sys.argv[0]).stem

__str__()

Return the fully qualified class name of this instance.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

def __str__(self) -> str:
    """Return the fully qualified class name of this instance."""
    return str(self.__class__)

app()

Build a fully configured Typer application.

Source code in src/pyrig_runtime/rig/cli/cli.py

def app(self) -> typer.Typer:
    """Build a fully configured Typer application."""
    app = self.base_app()
    return self.build_app(app)

base_app()

Create an empty base Typer application.

Returns:

Type	Description
Typer	A new Typer app configured to show help when invoked without
Typer	arguments.

Source code in src/pyrig_runtime/rig/cli/cli.py

def base_app(self) -> typer.Typer:
    """Create an empty base Typer application.

    Returns:
        A new Typer app configured to show help when invoked without
        arguments.
    """
    return typer.Typer(**self.base_app_kwargs())

base_app_kwargs()

Return keyword arguments for creating the base Typer application.

This base configuration makes sure that calling the CLI without any arguments will display the help message, and that the help text is the same as the description of the invoking project.

Returns:

Type	Description
dict[str, Any]	A dictionary of keyword arguments to pass to typer.Typer.

Source code in src/pyrig_runtime/rig/cli/cli.py

def base_app_kwargs(self) -> dict[str, Any]:
    """Return keyword arguments for creating the base Typer application.

    This base configuration makes sure that calling the CLI without
    any arguments will display the help message, and that the help text
    is the same as the description of the invoking project.

    Returns:
        A dictionary of keyword arguments to pass to `typer.Typer`.
    """
    return {
        "no_args_is_help": True,
        "help": distribution_summary(self.project_name()),
    }

build_app(app)

Register the callback and all commands onto the given app.

Parameters:

Name	Type	Description	Default
app	Typer	The Typer app to populate.	required

Returns:

Type	Description
Typer	The same app instance, now fully configured.

Source code in src/pyrig_runtime/rig/cli/cli.py

def build_app(self, app: typer.Typer) -> typer.Typer:
    """Register the callback and all commands onto the given app.

    Args:
        app: The Typer app to populate.

    Returns:
        The same app instance, now fully configured.
    """
    self.register_callback(app)
    self.register_subcommands(app)
    self.register_shared_subcommands(app)
    return app

callback(verbose=typer.Option(0, '--verbose', '-v', count=True, help='Increase verbosity: -v (DEBUG), -vv (modules), -vvv (timestamps)'), quiet=typer.Option(0, '--quiet', '-q', count=True, help='Decrease verbosity: -q (WARNING), -qq (ERROR), -qqq (CRITICAL)'))

Apply the verbosity options for the current invocation.

Parameters:

Name	Type	Description	Default
verbose	int	Number of times verbosity was increased (e.g. via -v).	Option(0, '--verbose', '-v', count=True, help='Increase verbosity: -v (DEBUG), -vv (modules), -vvv (timestamps)')
quiet	int	Number of times verbosity was decreased (e.g. via -q).	Option(0, '--quiet', '-q', count=True, help='Decrease verbosity: -q (WARNING), -qq (ERROR), -qqq (CRITICAL)')

Source code in src/pyrig_runtime/rig/cli/cli.py

def callback(
    self,
    verbose: int = typer.Option(
        0,
        "--verbose",
        "-v",
        count=True,
        help="Increase verbosity: -v (DEBUG), -vv (modules), -vvv (timestamps)",
    ),
    quiet: int = typer.Option(
        0,
        "--quiet",
        "-q",
        count=True,
        help="Decrease verbosity: -q (WARNING), -qq (ERROR), -qqq (CRITICAL)",
    ),
) -> None:
    """Apply the verbosity options for the current invocation.

    Args:
        verbose: Number of times verbosity was increased (e.g. via `-v`).
        quiet: Number of times verbosity was decreased (e.g. via `-q`).
    """
    self.configure_logging(verbose, quiet)

concrete_subclasses() classmethod

Yield all concrete leaf subclasses discovered across dependent packages.

Yields:

Type	Description
type[Self]	Non-abstract leaf subclass types.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def concrete_subclasses(cls) -> Iterator[type[Self]]:
    """Yield all concrete leaf subclasses discovered across dependent packages.

    Yields:
        Non-abstract leaf subclass types.
    """
    return discard_abstract_classes(cls.subclasses())

configure_logging(verbose, quiet)

Configure the logging level and format for the current invocation.

Each increment of verbose lowers the log level by one step (toward DEBUG); each increment of quiet raises it by one step (toward CRITICAL). The format also expands at higher verbosity, adding module names at two increments and timestamps at three.

The log level is intentionally unclamped to potentially support custom log levels.

Parameters:

Name	Type	Description	Default
verbose	int	Number of times verbosity was increased (e.g. via -v).	required
quiet	int	Number of times verbosity was decreased (e.g. via -q).	required

Note

Uses logging.basicConfig with force=True to ensure that the configuration is applied even if logging has already been configured by the calling project or other dependencies.

Source code in src/pyrig_runtime/rig/cli/cli.py

def configure_logging(self, verbose: int, quiet: int) -> None:
    """Configure the logging level and format for the current invocation.

    Each increment of `verbose` lowers the log level by one step (toward
    DEBUG); each increment of `quiet` raises it by one step (toward
    CRITICAL). The format also expands at higher verbosity, adding module
    names at two increments and timestamps at three.

    The log level is intentionally unclamped to potentially support
    custom log levels.

    Args:
        verbose: Number of times verbosity was increased (e.g. via `-v`).
        quiet: Number of times verbosity was decreased (e.g. via `-q`).

    Note:
        Uses `logging.basicConfig` with `force=True` to ensure that the
        configuration is applied even if logging has already been configured
        by the calling project or other dependencies.
    """
    level = logging.INFO
    step = logging.INFO - logging.DEBUG
    level -= step * verbose
    level += step * quiet

    verbose_names = 1
    verbose_modules = verbose_names + 1
    verbose_timestamps = verbose_modules + 1

    if verbose >= verbose_timestamps:
        fmt = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
    elif verbose == verbose_modules:
        fmt = "%(levelname)s [%(name)s] %(message)s"
    elif verbose == verbose_names:
        fmt = "%(levelname)s: %(message)s"
    else:
        fmt = "%(message)s"

    logging.basicConfig(level=level, format=fmt, force=True)

discovery_module() classmethod

Return the pyrig_runtime.rig.cli.cli module.

Source code in src/pyrig_runtime/rig/cli/cli.py

@classmethod
def discovery_module(cls) -> ModuleType:
    """Return the `pyrig_runtime.rig.cli.cli` module."""
    return sys.modules[__name__]

leaf() classmethod

Return the single leaf subclass found across dependent packages.

If no subclasses are found, the class itself is returned.

Returns:

Type	Description
type[Self]	The single leaf subclass type, or the class itself if no
type[Self]	subclasses are found. May be abstract.

Raises:

Type	Description
RuntimeError	If more than one leaf subclass is discovered across the dependent packages.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

    @classmethod
    def leaf(cls) -> type[Self]:
        """Return the single leaf subclass found across dependent packages.

        If no subclasses are found, the class itself is returned.

        Returns:
            The single leaf subclass type, or the class itself if no
            subclasses are found. May be abstract.

        Raises:
            RuntimeError: If more than one leaf subclass is discovered across
                the dependent packages.
        """
        subclasses = cls.subclasses()
        leaf = next(subclasses, cls)
        second = next(subclasses, None)
        if second is None:
            return leaf

        subclasses_dump = json.dumps(
            [fully_qualified_name(subcls) for subcls in (leaf, second, *subclasses)],
            indent=4,
        )
        msg = f"""Multiple leaf subclasses found for {cls}.
Defining multiple leaf subclasses is ambiguous.
This can happen if more than one leaf subclass is defined
across all the dependent packages.

Found subclasses:
{subclasses_dump}"""
        raise RuntimeError(msg)

module_subcommand_groups(module)

Return the Typer command groups found in a subcommands module.

Scans the module's namespace for typer.Typer instances. Both natively defined and imported instances are included. The key for each entry is the kebab-case form of the attribute name it is bound to.

Parameters:

Name	Type	Description	Default
module	ModuleType	The subcommands module to scan.	required

Returns:

Type	Description
dict[str, Typer]	Mapping of kebab-case attribute name to the typer.Typer instance
dict[str, Typer]	bound to it.

Source code in src/pyrig_runtime/rig/cli/cli.py

def module_subcommand_groups(self, module: ModuleType) -> dict[str, typer.Typer]:
    """Return the Typer command groups found in a subcommands module.

    Scans the module's namespace for `typer.Typer` instances. Both natively
    defined and imported instances are included. The key for each entry is
    the kebab-case form of the attribute name it is bound to.

    Args:
        module: The subcommands module to scan.

    Returns:
        Mapping of kebab-case attribute name to the `typer.Typer` instance
        bound to it.
    """
    return {
        snake_to_kebab_case(name): obj
        for name, obj in vars(module).items()
        if isinstance(obj, typer.Typer)
    }

package_name()

Return the snake_case package name of the invoking project.

For example, if the project is invoked as uv run my-project, the package name is my_project.

Source code in src/pyrig_runtime/rig/cli/cli.py

def package_name(self) -> str:
    """Return the snake_case package name of the invoking project.

    For example, if the project is invoked as `uv run my-project`, the
    package name is `my_project`.
    """
    return kebab_to_snake_case(self.project_name())

project_name()

Return the stem of sys.argv[0] as the invoking project name.

When a project is invoked through a registered console-script entry point (e.g. uv run my-project), sys.argv[0] is the path to that script, so its stem is the project name as it was registered.

Source code in src/pyrig_runtime/rig/cli/cli.py

def project_name(self) -> str:
    """Return the stem of `sys.argv[0]` as the invoking project name.

    When a project is invoked through a registered console-script entry point
    (e.g. `uv run my-project`), `sys.argv[0]` is the path to that script, so
    its stem is the project name as it was registered.
    """
    return Path(sys.argv[0]).stem

register_callback(app)

Attach the verbosity callback to the given app.

Parameters:

Name	Type	Description	Default
app	Typer	The Typer app to attach the callback to.	required

Source code in src/pyrig_runtime/rig/cli/cli.py

def register_callback(self, app: typer.Typer) -> None:
    """Attach the verbosity callback to the given app.

    Args:
        app: The Typer app to attach the callback to.
    """
    app.callback()(self.callback)

register_direct_subcommands(app, module)

Register every function defined in a module as a top-level command.

Adds each function found directly in module to app as a flat Typer command. Imported functions are excluded.

Parameters:

Name	Type	Description	Default
app	Typer	The Typer app to register the commands onto.	required
module	ModuleType	The subcommands module to scan for command functions.	required

Source code in src/pyrig_runtime/rig/cli/cli.py

def register_direct_subcommands(self, app: typer.Typer, module: ModuleType) -> None:
    """Register every function defined in a module as a top-level command.

    Adds each function found directly in `module` to `app` as a flat Typer
    command. Imported functions are excluded.

    Args:
        app: The Typer app to register the commands onto.
        module: The subcommands module to scan for command functions.
    """
    for func in module_functions(module):
        app.command()(func)

register_shared_subcommands(app)

Discover and register shared commands from pyrig-runtime and its dependents.

Parameters:

Name	Type	Description	Default
app	Typer	The Typer app to register the commands onto.	required

Note

Commands are registered in dependency order (pyrig-runtime first, then dependent packages in topological order). When two packages define a command with the same name, the last registration takes precedence.

Source code in src/pyrig_runtime/rig/cli/cli.py

def register_shared_subcommands(self, app: typer.Typer) -> None:
    """Discover and register shared commands from pyrig-runtime and its dependents.

    Args:
        app: The Typer app to register the commands onto.

    Note:
        Commands are registered in dependency order (pyrig-runtime first,
        then dependent packages in topological order). When two packages
        define a command with the same name, the last registration takes
        precedence.
    """
    for shared_subcommands_module in chain(
        (shared_subcommands,),
        discover_equivalent_modules_across_dependents(
            shared_subcommands,
        ),
    ):
        self.register_direct_subcommands(app=app, module=shared_subcommands_module)
        self.register_subcommand_groups(app=app, module=shared_subcommands_module)

register_subcommand_groups(app, module)

Register every typer.Typer instance in a module as a named command group.

Attaches each typer.Typer found in the module's namespace to app, using the kebab-case form of the attribute name as the group name.

Parameters:

Name	Type	Description	Default
app	Typer	The Typer app to register the command groups onto.	required
module	ModuleType	The subcommands module to scan for group instances.	required

Source code in src/pyrig_runtime/rig/cli/cli.py

def register_subcommand_groups(self, app: typer.Typer, module: ModuleType) -> None:
    """Register every `typer.Typer` instance in a module as a named command group.

    Attaches each `typer.Typer` found in the module's namespace to `app`,
    using the kebab-case form of the attribute name as the group name.

    Args:
        app: The Typer app to register the command groups onto.
        module: The subcommands module to scan for group instances.
    """
    for name, group in self.module_subcommand_groups(module).items():
        app.add_typer(group, name=name)

register_subcommands(app)

Discover and register project-specific commands for the invoking project.

Any dependent project can define its own CLI commands by adding functions or typer.Typer groups to <package>.rig.cli.subcommands.

Parameters:

Name	Type	Description	Default
app	Typer	The Typer app to register the commands onto.	required

Note

If the invoking project's subcommands module cannot be imported, registration is silently skipped.

Source code in src/pyrig_runtime/rig/cli/cli.py

def register_subcommands(self, app: typer.Typer) -> None:
    """Discover and register project-specific commands for the invoking project.

    Any dependent project can define its own CLI commands by adding
    functions or `typer.Typer` groups to `<package>.rig.cli.subcommands`.

    Args:
        app: The Typer app to register the commands onto.

    Note:
        If the invoking project's subcommands module cannot be imported,
        registration is silently skipped.
    """
    subcommands_module = replace_root_module(
        subcommands, root=self.package_name(), default=None
    )

    if subcommands_module is None:
        return

    self.register_direct_subcommands(app=app, module=subcommands_module)
    self.register_subcommand_groups(app=app, module=subcommands_module)

run()

Build and invoke the Typer application.

Source code in src/pyrig_runtime/rig/cli/cli.py

def run(self) -> None:
    """Build and invoke the Typer application."""
    self.app()()

sort_key() classmethod

Return the sort key used to order this class relative to peer subclasses.

Override to sort by priority, numeric position, or any other criterion. The default returns the class name, giving alphabetical ordering.

Returns:

Type	Description
Any	A value comparable with < against the sort keys of other
Any	subclasses.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def sort_key(cls) -> Any:
    """Return the sort key used to order this class relative to peer subclasses.

    Override to sort by priority, numeric position, or any other criterion.
    The default returns the class name, giving alphabetical ordering.

    Returns:
        A value comparable with `<` against the sort keys of other
        subclasses.
    """
    return cls.__name__

subclasses() classmethod

Yield all subclasses discovered across installed dependent packages.

Only leaf-level subclasses are yielded; any intermediate parent classes that also appear in the result set are omitted.

Yields:

Type	Description
type[Self]	Leaf subclass types.

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def subclasses(cls) -> Iterator[type[Self]]:
    """Yield all subclasses discovered across installed dependent packages.

    Only leaf-level subclasses are yielded; any intermediate parent classes
    that also appear in the result set are omitted.

    Yields:
        Leaf subclass types.
    """
    return discard_parent_classes(
        discover_subclasses_across_dependencies(
            cls,
            module=cls.discovery_module(),
        )
    )

subclasses_sorted(subclasses) classmethod

Sort the given subclasses using each subclass's sort_key().

Does not perform any discovery.

Parameters:

Name	Type	Description	Default
subclasses	Iterable[type[Self]]	Subclass types to sort.	required

Returns:

Type	Description
list[type[Self]]	The same subclass types sorted by their sort_key().

Source code in src/pyrig_runtime/core/dependencies/subclass.py

@classmethod
def subclasses_sorted(cls, subclasses: Iterable[type[Self]]) -> list[type[Self]]:
    """Sort the given subclasses using each subclass's `sort_key()`.

    Does not perform any discovery.

    Args:
        subclasses: Subclass types to sort.

    Returns:
        The same subclass types sorted by their `sort_key()`.
    """
    return sorted(subclasses, key=lambda subclass: subclass.sort_key())

commands

Implementations of CLI command actions for pyrig-runtime-based projects.

version

Version display command for pyrig-runtime-based project CLIs.

project_version()

Print the name and installed version of the project using this runtime.

The version belongs to the dependent project, not to pyrig-runtime itself. The project must be installed for its version to be available.

Source code in src/pyrig_runtime/rig/cli/commands/version.py

def project_version() -> None:
    """Print the name and installed version of the project using this runtime.

    The version belongs to the dependent project, not to pyrig-runtime itself.
    The project must be installed for its version to be available.
    """
    project_name = CLI.I.project_name()
    typer.echo(f"{project_name} {version(project_name)}")

main

Console-script entry point for pyrig-runtime-based projects.

main()

Run the CLI application.

Source code in src/pyrig_runtime/rig/cli/main.py

def main() -> None:
    """Run the CLI application."""
    CLI.I.run()

shared_subcommands

Shared CLI commands for all dependent packages.

In every installed pyrig-runtime based package functions defined directly in this module are registered as top-level CLI commands and module-level typer.Typer instances are registered as command groups, with each group's name derived from the kebab-case form of the variable name.

version()

Print the name and installed version.

Reports the version of whichever project's CLI entry point was used to invoke this command, not pyrig-runtime's own version. The project must be installed; an editable install is sufficient.

Example

$ uv run myproject version
myproject 0.4.1

Source code in src/pyrig_runtime/rig/cli/shared_subcommands.py

def version() -> None:
    """Print the name and installed version.

    Reports the version of whichever project's CLI entry point was used to
    invoke this command, not pyrig-runtime's own version. The project must be
    installed; an editable install is sufficient.

    Example:
        ```
        $ uv run myproject version
        myproject 0.4.1
        ```
    """
    from pyrig_runtime.rig.cli.commands.version import project_version  # noqa: PLC0415

    project_version()

subcommands

Project-specific CLI commands.

Functions defined directly in this module are registered as top-level CLI commands. Module-level typer.Typer instances are registered as command groups, with each group's name derived from the kebab-case form of the variable name.