Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions ATTRIBUTIONS.md
Original file line number Diff line number Diff line change
Expand Up @@ -6247,12 +6247,13 @@ THE SOFTWARE.

```

## requests (2.32.5) - Apache Software License
## requests (2.33.0) - Apache Software License

Python HTTP for Humans.

* URL: https://requests.readthedocs.io
* Author(s): Kenneth Reitz
* URL: https://github.com/psf/requests
* Author(s): Kenneth Reitz <me@kennethreitz.org>
* Maintainer(s): Ian Stapleton Cordasco <graffatcolmingov@gmail.com>, Nate Prewitt <nate.prewitt@gmail.com>

### License Text

Expand Down Expand Up @@ -6435,6 +6436,14 @@ Python HTTP for Humans.

```

### Notice

```
Requests
Copyright 2019 Kenneth Reitz

```

## rfc3339-validator (0.1.4) - MIT License

A pure python RFC3339 validator
Expand Down
1 change: 1 addition & 0 deletions noxfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ def audit(session: nox.Session) -> None:
# pip-audit to check for vulnerabilities
ignore_vulns = [
"CVE-2025-53000", # No fix for nbconvert yet
"CVE-2026-4539", # No fix available
]
try:
session.run(
Expand Down
17 changes: 17 additions & 0 deletions src/aignostics_foundry_core/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ This file provides an overview of all modules in `aignostics_foundry_core`, thei
| Module | Purpose | Description |
|--------|---------|-------------|
| **console** | Themed terminal output | Module-level `console` object (Rich `Console`) with colour theme and `_get_console()` factory |
| **di** | Dependency injection | `locate_subclasses`, `locate_implementations`, `load_modules`, `discover_plugin_packages`, `clear_caches`, `PLUGIN_ENTRY_POINT_GROUP` for plugin and subclass discovery |
| **health** | Service health checks | `Health` model and `HealthStatus` enum for tree-structured health status |
| **settings** | Pydantic settings loading | `OpaqueSettings`, `load_settings`, `strip_to_none_before_validator`, `UNHIDE_SENSITIVE_INFO` for env-based settings with secret masking and user-friendly validation errors |

Expand Down Expand Up @@ -42,6 +43,22 @@ This file provides an overview of all modules in `aignostics_foundry_core`, thei
- **Location**: `aignostics_foundry_core/settings.py`
- **Dependencies**: `pydantic>=2`, `pydantic-settings>=2`, `rich>=14`

### di

**Plugin and subclass discovery for dependency injection**

- **Purpose**: Provides reusable infrastructure for dynamically discovering plugin packages, class implementations, and subclasses across a project and its registered plugins
- **Key Features**:
- `PLUGIN_ENTRY_POINT_GROUP: str` — `"aignostics.plugins"` entry-point group constant
- `discover_plugin_packages()` — discovers plugin packages registered via `[project.entry-points."aignostics.plugins"]`; LRU-cached
- `load_modules(project_name)` — imports all top-level submodules of the given package
- `locate_implementations(_class, project_name)` — finds all instances of `_class` via shallow plugin scan + deep project scan; cached per `(_class, project_name)` to prevent cross-project pollution
- `locate_subclasses(_class, project_name)` — finds all subclasses of `_class` via shallow plugin scan + deep project scan; cached per `(_class, project_name)`
- `clear_caches()` — resets all module-level caches (`_implementation_cache`, `_subclass_cache`, `discover_plugin_packages` LRU cache)
- Two internal scan helpers: `_scan_packages_shallow` (plugin top-level exports only) and `_scan_packages_deep` (full submodule walk for the main project)
- **Location**: `aignostics_foundry_core/di.py`
- **Dependencies**: Python stdlib only (`importlib`, `pkgutil`, `importlib.metadata`)

### health

**Tree-structured health status for service health checks**
Expand Down
200 changes: 200 additions & 0 deletions src/aignostics_foundry_core/di.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,200 @@
"""Dependency injection using dynamic import and discovery of implementations and subclasses."""

import importlib
import pkgutil
from collections.abc import Callable
from functools import lru_cache
from importlib.metadata import entry_points
from inspect import isclass
from typing import Any

_implementation_cache: dict[tuple[Any, str], list[Any]] = {}
_subclass_cache: dict[tuple[Any, str], list[Any]] = {}

PLUGIN_ENTRY_POINT_GROUP = "aignostics.plugins"


@lru_cache(maxsize=1)
def discover_plugin_packages() -> tuple[str, ...]:
"""Discover plugin packages using entry points.

Plugins register themselves in their pyproject.toml:

[project.entry-points."aignostics.plugins"]
my_plugin = "my_plugin"

Results are cached after the first call.

Returns:
Tuple of discovered plugin package names.
"""
eps = entry_points(group=PLUGIN_ENTRY_POINT_GROUP)
return tuple(ep.value for ep in eps)


def load_modules(project_name: str) -> None:
"""Import all top-level submodules of the given project package.

Args:
project_name: The importable package name to scan (e.g. ``"bridge"``).
"""
package = importlib.import_module(project_name)
for _, name, _ in pkgutil.iter_modules(package.__path__):
importlib.import_module(f"{project_name}.{name}")


def _scan_packages_deep(
package_name: str,
predicate: Callable[[Any], bool],
) -> list[Any]:
"""Deep-scan a single package by walking all submodules via pkgutil.iter_modules.

Used for the main project package. Imports each submodule discovered via
``pkgutil.iter_modules`` and examines every name in ``dir(module)`` against
*predicate*. Silently skips submodules that raise ``ImportError``.

Args:
package_name: The package to scan (e.g. ``"bridge"``).
predicate: Called with each member; members where this returns ``True``
are included in the result.

Returns:
All members from submodules of *package_name* that satisfy *predicate*.
"""
results: list[Any] = []
try:
package = importlib.import_module(package_name)
except ImportError:
return results
for _, name, _ in pkgutil.iter_modules(package.__path__):
try:
module = importlib.import_module(f"{package_name}.{name}")
for member_name in dir(module):
member = getattr(module, member_name)
if predicate(member):
results.append(member)
except ImportError:
continue
return results


def _scan_packages_shallow(
package_names: tuple[str, ...],
predicate: Callable[[Any], bool],
) -> list[Any]:
"""Shallow-scan the top-level exports of each plugin package.

For each plugin package, imports only the top-level package and examines
``dir(package)`` for matches. Does **not** walk submodules via
``pkgutil.iter_modules``.

This prevents nested objects from plugin submodules (e.g.
``stargate.demeter.cli``) from being discovered alongside the intended
top-level export (``stargate.cli``). Only what the plugin's ``__init__.py``
explicitly exports is considered.

Silently skips packages that raise ``ImportError``.

Args:
package_names: Plugin package names to scan.
predicate: Called with each member; members where this returns ``True``
are included in the result.

Returns:
All members from the top-level namespace of each plugin that satisfy
*predicate*.
"""
results: list[Any] = []
for package_name in package_names:
try:
package = importlib.import_module(package_name)
except ImportError:
continue
for member_name in dir(package):
member = getattr(package, member_name)
if predicate(member):
results.append(member)
return results


def locate_implementations(_class: type[Any], project_name: str) -> list[Any]:
"""Dynamically discover all instances of some class.

Searches plugin top-level exports first (shallow scan), then deep-scans all
submodules of the main project package. Plugins are registered via entry
points; only their top-level ``__init__.py`` exports are examined (submodules
are not walked). The main package retains full deep-scan behaviour.

Cache keys include *project_name* to avoid cross-project cache pollution when
multiple projects share this library.

Args:
_class: Class to search for.
project_name: Importable package name of the calling project
(e.g. ``"bridge"``). Used as the deep-scan root and as part of the
cache key.

Returns:
List of discovered instances of the given class.
"""
cache_key = (_class, project_name)
if cache_key in _implementation_cache:
return _implementation_cache[cache_key]

def predicate(member: object) -> bool:
return isinstance(member, _class)

results = [
*_scan_packages_shallow(discover_plugin_packages(), predicate),
*_scan_packages_deep(project_name, predicate),
]
_implementation_cache[cache_key] = results
return results


def locate_subclasses(_class: type[Any], project_name: str) -> list[Any]:
"""Dynamically discover all classes that are subclasses of some type.

Searches plugin top-level exports first (shallow scan), then deep-scans all
submodules of the main project package. Plugins are registered via entry
points; only their top-level ``__init__.py`` exports are examined (submodules
are not walked). The main package retains full deep-scan behaviour.

Cache keys include *project_name* to avoid cross-project cache pollution when
multiple projects share this library.

Args:
_class: Parent class of subclasses to search for.
project_name: Importable package name of the calling project
(e.g. ``"bridge"``). Used as the deep-scan root and as part of the
cache key.

Returns:
List of discovered subclasses of the given class.
"""
cache_key = (_class, project_name)
if cache_key in _subclass_cache:
return _subclass_cache[cache_key]

def predicate(member: object) -> bool:
return isclass(member) and issubclass(member, _class) and member != _class

results = [
*_scan_packages_shallow(discover_plugin_packages(), predicate),
*_scan_packages_deep(project_name, predicate),
]
_subclass_cache[cache_key] = results
return results


def clear_caches() -> None:
"""Reset all module-level discovery caches.

Clears ``_implementation_cache``, ``_subclass_cache``, and the
``discover_plugin_packages`` LRU cache so that subsequent calls to
``locate_implementations``, ``locate_subclasses``, and
``discover_plugin_packages`` perform fresh discovery.
"""
_implementation_cache.clear()
_subclass_cache.clear()
discover_plugin_packages.cache_clear()
Loading
Loading