API¶

@cache
def dependency_graph() -> DependencyGraph:
    """Return the dependency graph of `pyrig_runtime` and its dependents.

    Built once and cached. Pruned to `pyrig_runtime` and every package that
    depends on it, directly or transitively; packages it depends on are not
    included.

    Returns:
        Directed graph rooted at `pyrig_runtime`, containing only its
        ancestors.

    Note:
        The returned instance is shared across all callers. Do not mutate it.
    """
    graph = DependencyGraph()
    graph.prune(root=pyrig_runtime.__name__)
    return graph

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

    The result is cached per unique `package` argument.

    Args:
        package: Package whose dependents should be discovered.

    Returns:
        Tuple of imported module objects for every package that depends on
        `package`, directly or transitively, in dependency order
        (dependencies before dependents). Does not include `package`
        itself.

    Raises:
        KeyError: If `package` is not `pyrig_runtime` or one of its
            dependents.
    """
    graph = dependency_graph()
    return tuple(import_modules(graph.sorted_ancestors(package.__name__)))

def equivalent_modules_across_dependencies(
    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 the import succeeds. 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 locates the corresponding
            module in each dependent.

    Yields:
        Successfully imported module objects, in dependency order.
        Dependents are silently skipped whenever importing the equivalent
        module fails, whether because no module exists at that sub-path
        or because the import itself raises.
    """
    for package in dependent_packages(root_module(module)):
        package_module = replace_root_module(module, package.__name__, default=None)
        if package_module is not None:
            yield package_module

def subclasses_across_dependencies[T](
    cls: type[T],
    module: ModuleType,
) -> Iterator[type[T]]:
    """Yield subclasses of `cls` defined at the same sub-path as `module`.

    The search covers `module` itself, every sub-module if `module` is a
    package, and the equivalently-located module (and its own sub-modules,
    if a package) in every installed package that depends on `module`'s
    root package.

    Args:
        cls: Base class whose subclasses should be discovered.
        module: Module or package that scopes the search, and whose root
            package determines which dependents are searched.

    Yields:
        Subclass types of `cls` found anywhere in the search scope. The
        order is stable across calls but reflects discovery order, not
        any deliberate priority.

    Note:
        Every module within the search scope is imported as a side effect,
        executing any module-level code it contains.
    """
    for package in filter(
        is_package,
        chain(
            (module,),
            equivalent_modules_across_dependencies(module=module),
        ),
    ):
        register_package_modules(package)

    module_name = module.__name__
    root_name = root_module_name(module_name)
    for subclass in discover_subclasses(cls):
        if replace_root_module_name(
            subclass.__module__,
            root_name,
        ).startswith(module_name):
            yield subclass

def distribution_header(metadata: str) -> str:
    """Return the header portion of a distribution's metadata.

    The header is the metadata content before the first blank line,
    containing the single-line RFC 822 header fields (e.g. `Name`,
    `Requires-Dist`). If no blank line is found, the entire metadata
    text is returned.

    Args:
        metadata: The full metadata of an installed distribution.

    Returns:
        The header portion of the distribution's metadata.
    """
    end = metadata.find("\n\n")
    return metadata[:end] if end != -1 else metadata

def distribution_header_value_pattern(field_name: str) -> re.Pattern[str]:
    """Compile a regex that matches a single-line metadata header field.

    Matches lines of the form `field_name: value`. Use `findall` on the
    result to collect every value when the header repeats.

    Args:
        field_name: Name of the header field to match, exactly as it
            appears in the metadata (e.g. `Name`).

    Returns:
        A case-sensitive pattern with each match's value captured in the
        first group.
    """
    return re.compile(rf"^{field_name}:[ \t]*(.*)$", re.MULTILINE)

def distribution_metadata(dist: Distribution) -> str | None:
    """Return the full metadata text of a distribution, or `None` if it has none."""
    return dist.read_text("METADATA")

def distribution_name(metadata: str) -> str:
    """Return the name of a distribution from its metadata.

    Args:
        metadata: The full metadata of an installed distribution.

    Returns:
        The name of the distribution.
    """
    return regex_find(DISTRIBUTION_NAME_PATTERN, metadata)

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

    Version specifiers, extras, and anything else after the package name
    are discarded. Hyphens are normalized to underscores; dots are kept, so
    namespace packages remain dotted.

    Args:
        req: A dependency requirement string, e.g. `my-package[extra]>=1.0`.

    Returns:
        The package name in snake_case, e.g. `my_package`.

    Example:
        >>> distribution_requirement_as_module_name("my-package[extra]>=1.0.0")
        'my_package'
    """
    return kebab_to_snake_case(regex_find(REQUIRES_DIST_NAME_PATTERN, req))

def distribution_requirements(metadata: str) -> list[str]:
    """Return the list of dependency requirements from a distribution's metadata."""
    return DISTRIBUTION_REQUIRES_DIST_PATTERN.findall(metadata)

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

    Args:
        metadata: The full metadata of an installed distribution.

    Returns:
        The distribution's summary description.

    Raises:
        LookupError: If the metadata has no `Summary` field.
    """
    return regex_find(DISTRIBUTION_SUMMARY_PATTERN, metadata)

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

    Each subclass appears exactly once, even when reachable through multiple
    inheritance paths. 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:
        Tuple of all transitive subclass types, excluding `cls` itself.
    """
    visited: dict[type[T], None] = {}
    stack = cls.__subclasses__()
    while stack:
        subclass = stack.pop()
        if subclass in visited:
            continue
        visited[subclass] = None
        stack.extend(subclass.__subclasses__())
    return tuple(visited)

def filter_concrete_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, in the same order
        as `classes`.
    """
    return filterfalse(inspect.isabstract, classes)

def filter_leaf_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, in
        the same order as `classes`.
    """
    classes = tuple(classes)
    parents = {parent for cls in classes for parent in cls.__mro__[1:]}
    return filterfalse(parents.__contains__, classes)

def generate_class[T](
    name: str,
    bases: tuple[type[T], ...],
    methods: Iterable[FunctionType] = (),
    namespace: dict[str, Any] | None = None,
) -> type[T]:
    """Dynamically create a class from base classes, methods, and attributes.

    Args:
        name: Name of the new class, used as its `__name__`.
        bases: Base classes the new class inherits from.
        methods: Functions to add to the class, each under its own `__name__`.
        namespace: Extra attributes for the class body, keyed by name. Mutated
            in place with the `methods` entries added on top, so a method
            whose name matches a key here overrides it. Defaults to a new,
            empty dict when omitted.

    Returns:
        The newly created class.

    Raises:
        TypeError: If `bases` cannot be combined into a class with a
            consistent method resolution order.

    Note:
        The generated class's `__module__` is whatever the underlying
        `type()` call infers from the calling context, which is not
        necessarily the caller's own module. Pass `"__module__"` in
        `namespace` to set it explicitly.
    """
    if namespace is None:
        namespace = {}
    for method in methods:
        namespace[method.__name__] = method
    return cast(
        "type[T]",
        type(
            name,
            bases,
            namespace,
        ),
    )

def filter_module_functions(
    module: ModuleType,
    members: Iterable[Any],
) -> Iterator[Callable[..., Any]]:
    """Yield functions from `members` defined directly in `module`.

    Args:
        module: Module to filter functions for.
        members: Iterable of candidate members to filter.

    Yields:
        Each function defined in `module` from `members`.
    """
    for member in members:
        unwrapped_member = unwrap_obj(member)
        if (
            inspect.isfunction(unwrapped_member)
            and unwrapped_member.__module__ == module.__name__
        ):
            yield member

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`.
    """
    yield from filter_module_functions(module, obj_members(module))

def obj_members(
    obj: object,
    predicate: Callable[[Any], bool] | None = None,
) -> Iterator[Any]:
    """Yield the values of an object's members 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.

    Yields:
        Each matching member's value.
    """
    excluded = {"__annotate__", "__annotate_func__"}
    return (
        value
        for member, value in inspect.getmembers_static(obj, predicate=predicate)
        if member not in excluded
    )