Core API Reference

API reference for myfy-core package.

---

Application

Application

Application(
    settings_class=CoreSettings, auto_discover=True
)

The myfy application kernel.

Lifecycle: 1. Create application instance 2. Configure modules and providers 3. Initialize (compile DI, discover modules) 4. Start modules 5. Run 6. Stop modules gracefully

Usage

app = Application() app.add_module(WebModule()) await app.run()

Create a new application.

Parameters:

Name	Type	Description	Default
settings_class	type[BaseSettings]	Settings class to load	CoreSettings
auto_discover	bool	Automatically discover modules via entry points	True

Source code in packages/myfy-core/myfy/core/kernel/app.py

def __init__(
    self,
    settings_class: type[BaseSettings] = CoreSettings,
    auto_discover: bool = True,
):
    """
    Create a new application.

    Args:
        settings_class: Settings class to load
        auto_discover: Automatically discover modules via entry points
    """
    self.container = Container()
    self.settings = load_settings(settings_class)

    # Get shutdown timeout from settings or use default
    shutdown_timeout = getattr(self.settings, "shutdown_timeout", 10.0)
    self.lifecycle = LifecycleManager(timeout=shutdown_timeout)

    self._initialized = False
    self._modules: list[Module] = []
    self._auto_discover = auto_discover

add_module

add_module(module)

Register a module with the application.

Must be called before initialize().

Parameters:

Name	Type	Description	Default
module	type[T]	The module to add	required

Source code in packages/myfy-core/myfy/core/kernel/app.py

def add_module[T: ModuleType](self, module: type[T]) -> None:
    """
    Register a module with the application.

    Must be called before initialize().

    Args:
        module: The module to add
    """
    if self._initialized:
        raise RuntimeError(
            "Cannot add modules after initialization. "
            "Add modules before calling initialize() or run()."
        )
    self._modules.append(module)
    self.lifecycle.add_module(module)

get_module

get_module(module_type)

Get a module by type.

Parameters:

Name	Type	Description	Default
module_type	type[T]	The module class to find (e.g., WebModule)	required

Returns:

Type	Description
T	The module instance

Raises:

Type	Description
ModuleNotFoundError	If module not found

Example

web_module = app.get_module(WebModule)

Source code in packages/myfy-core/myfy/core/kernel/app.py

def get_module(self, module_type: type[T]) -> T:
    """
    Get a module by type.

    Args:
        module_type: The module class to find (e.g., WebModule)

    Returns:
        The module instance

    Raises:
        ModuleNotFoundError: If module not found

    Example:
        web_module = app.get_module(WebModule)
    """
    for module in self._modules:
        if isinstance(module, module_type):
            return module  # type: ignore
    raise MyfyModuleNotFoundError(module_type)

has_module

has_module(module_type)

Check if a module type is registered.

Parameters:

Name	Type	Description	Default
module_type	type	The module class to check	required

Returns:

Type	Description
bool	True if module is registered, False otherwise

Example

if app.has_module(FrontendModule): print("Frontend module is loaded")

Source code in packages/myfy-core/myfy/core/kernel/app.py

def has_module(self, module_type: type) -> bool:
    """
    Check if a module type is registered.

    Args:
        module_type: The module class to check

    Returns:
        True if module is registered, False otherwise

    Example:
        if app.has_module(FrontendModule):
            print("Frontend module is loaded")
    """
    return any(isinstance(m, module_type) for m in self._modules)

get_modules_implementing

get_modules_implementing(protocol)

Get all modules implementing a specific protocol.

Parameters:

Name	Type	Description	Default
protocol	type[T]	The protocol type to filter by (e.g., IWebExtension)	required

Returns:

Type	Description
list[T]	List of modules that declare they implement the protocol

Example

extensions = app.get_modules_implementing(IWebExtension) for ext in extensions: ext.extend_asgi_app(asgi_app)

Source code in packages/myfy-core/myfy/core/kernel/app.py

def get_modules_implementing(self, protocol: type[T]) -> list[T]:
    """
    Get all modules implementing a specific protocol.

    Args:
        protocol: The protocol type to filter by (e.g., IWebExtension)

    Returns:
        List of modules that declare they implement the protocol

    Example:
        extensions = app.get_modules_implementing(IWebExtension)
        for ext in extensions:
            ext.extend_asgi_app(asgi_app)
    """
    result = []
    for module in self._modules:
        provides = getattr(module, "provides", [])
        if protocol in provides:
            result.append(module)  # type: ignore
    return result

create_lifespan

create_lifespan()

Create lifespan context manager.

Returns a lifespan context that starts/stops all modules. This centralizes lifespan creation for use in CLI and factories.

Returns:

Type	Description
	lifespan context manager

Example

lifespan = app.create_lifespan() asgi_app = web_module.get_asgi_app(container, lifespan=lifespan)

Source code in packages/myfy-core/myfy/core/kernel/app.py

def create_lifespan(self):
    """
    Create lifespan context manager.

    Returns a lifespan context that starts/stops all modules.
    This centralizes lifespan creation for use in CLI and factories.

    Returns:
        lifespan context manager

    Example:
        lifespan = app.create_lifespan()
        asgi_app = web_module.get_asgi_app(container, lifespan=lifespan)
    """

    @asynccontextmanager
    async def lifespan(app):  # noqa: ARG001
        """lifespan that manages all module lifecycles."""
        await self.lifecycle.start_all()
        try:
            yield
        finally:
            await self.lifecycle.stop_all()

    return lifespan

initialize

initialize()

Initialize the application with multi-phase module setup.

Phases: 1. Discovery - Auto-discover modules via entry points 2. Dependency Validation - Validate module dependency graph 3. Configure - Modules register services in DI 4. Extend - Modules modify service registrations (optional) 5. Compile - DI container builds injection plans 6. Finalize - Modules configure singleton services (optional)

This must be called before start() or run().

Source code in packages/myfy-core/myfy/core/kernel/app.py

def initialize(self) -> None:
    """
    Initialize the application with multi-phase module setup.

    Phases:
    1. Discovery - Auto-discover modules via entry points
    2. Dependency Validation - Validate module dependency graph
    3. Configure - Modules register services in DI
    4. Extend - Modules modify service registrations (optional)
    5. Compile - DI container builds injection plans
    6. Finalize - Modules configure singleton services (optional)

    This must be called before start() or run().
    """
    if self._initialized:
        return

    # Phase 1: Discovery
    if self._auto_discover:
        self._discover_modules()

    # Phase 2: Validate dependencies (fail-fast)
    self._validate_dependencies()

    # Register core settings as singleton
    self.container.register(
        type_=type(self.settings),
        factory=lambda: self.settings,
        scope=SINGLETON,
    )

    # Also make CoreSettings available if using a custom settings class
    if not isinstance(self.settings, CoreSettings):
        self.container.register(
            type_=CoreSettings,
            factory=lambda: self.settings,  # type: ignore
            scope=SINGLETON,
        )

    # Register nested module settings (ADR-0007: Optional Nested Module Settings)
    # If user's settings class contains nested BaseSettings, register them too
    self._register_nested_settings()

    # Phase 3: Configure (register services)
    for module in self._modules:
        module.configure(self.container)

    # Register any @provider decorated functions
    register_providers_in_container(self.container)

    # Phase 4: Extend (optional - modify registrations)
    for module in self._modules:
        if hasattr(module, "extend"):
            module.extend(self.container)

    # Phase 5: Compile the container (build injection plans, detect cycles)
    self.container.compile()

    # Phase 6: Finalize (optional - configure singletons)
    for module in self._modules:
        if hasattr(module, "finalize"):
            module.finalize(self.container)

    self._initialized = True

start async

start()

Start the application.

Initializes (if not already done) and starts all modules.

Source code in packages/myfy-core/myfy/core/kernel/app.py

async def start(self) -> None:
    """
    Start the application.

    Initializes (if not already done) and starts all modules.
    """
    if not self._initialized:
        self.initialize()

    await self.lifecycle.start_all()

stop async

stop()

Stop the application gracefully.

Source code in packages/myfy-core/myfy/core/kernel/app.py

async def stop(self) -> None:
    """Stop the application gracefully."""
    await self.lifecycle.stop_all()

run async

run()

Run the application until shutdown signal.

This is the main entry point for running the application. Sets up signal handlers and manages the full lifecycle.

Usage

app = Application() await app.run()

Source code in packages/myfy-core/myfy/core/kernel/app.py

async def run(self) -> None:
    """
    Run the application until shutdown signal.

    This is the main entry point for running the application.
    Sets up signal handlers and manages the full lifecycle.

    Usage:
        app = Application()
        await app.run()
    """
    if not self._initialized:
        self.initialize()

    # Set up signal handlers for graceful shutdown
    self.lifecycle.setup_signal_handlers()

    async with self.lifecycle.lifespan():
        app_name = getattr(self.settings, "app_name", "myfy-app")
        print(f"🚀 {app_name} started")
        print(
            f"📦 Loaded {len(self._modules)} module(s): {', '.join(m.name for m in self._modules)}"
        )

        # Wait for shutdown signal
        await self.lifecycle.wait_for_shutdown()

    print(f"👋 {app_name} stopped")

Dependency Injection

provider

provider(
    scope=SINGLETON,
    qualifier=None,
    name=None,
    reloadable=(),
)

Decorator to register a function as a dependency provider.

The decorated function will be registered in the container during application startup. The return type annotation is used as the provided type.

Usage

@provider(scope=SINGLETON) def database(settings: Settings) -> Database: return Database(settings.db_url)

@provider(scope=REQUEST) def unit_of_work(db: Database) -> UnitOfWork: return UnitOfWork(db)

Parameters:

Name	Type	Description	Default
scope	Scope	Lifecycle scope (default: SINGLETON)	SINGLETON
qualifier	str | None	Optional qualifier for multiple providers of same type	None
name	str | None	Optional name for resolution	None
reloadable	tuple[str, ...]	Tuple of settings that can be hot-reloaded	()

Returns:

Type	Description
Callable[[Callable[..., T]], Callable[..., T]]	Decorator that registers the provider

Source code in packages/myfy-core/myfy/core/di/provider.py

def provider(
    scope: Scope = SINGLETON,
    qualifier: str | None = None,
    name: str | None = None,
    reloadable: tuple[str, ...] = (),
) -> Callable[[Callable[..., T]], Callable[..., T]]:
    """
    Decorator to register a function as a dependency provider.

    The decorated function will be registered in the container during
    application startup. The return type annotation is used as the
    provided type.

    Usage:
        @provider(scope=SINGLETON)
        def database(settings: Settings) -> Database:
            return Database(settings.db_url)

        @provider(scope=REQUEST)
        def unit_of_work(db: Database) -> UnitOfWork:
            return UnitOfWork(db)

    Args:
        scope: Lifecycle scope (default: SINGLETON)
        qualifier: Optional qualifier for multiple providers of same type
        name: Optional name for resolution
        reloadable: Tuple of settings that can be hot-reloaded

    Returns:
        Decorator that registers the provider
    """

    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        # Store provider metadata for later registration
        metadata = {
            "factory": func,
            "scope": scope,
            "qualifier": qualifier,
            "name": name,
            "reloadable": reloadable,
        }
        _pending_providers.append((func, metadata))

        # Mark the function with metadata (useful for introspection)
        func.__myfy_provider__ = metadata  # type: ignore

        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> T:
            # During normal calls, just execute the function
            # (The container will handle injection)
            return func(*args, **kwargs)

        return wrapper

    return decorator

Container

Container()

Dependency injection container with compile-time resolution.

Features: - Type-based resolution with optional qualifiers - Three scopes: singleton, request, task - Cycle detection at compile time - Zero reflection on hot path (all plans compiled at startup) - Test-friendly overrides

Source code in packages/myfy-core/myfy/core/di/container.py

def __init__(self):
    self._providers: dict[ProviderKey, ProviderRegistration] = {}
    self._singletons: dict[ProviderKey, Any] = {}
    self._singleton_locks: dict[ProviderKey, threading.Lock] = {}
    self._frozen = False
    self._override_stack: list[dict[type, ProviderFactory]] = []
    self._has_active_overrides = False

register

register(
    type_,
    factory,
    *,
    scope=SINGLETON,
    qualifier=None,
    name=None,
    reloadable=(),
)

Register a provider in the container.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type this provider produces	required
factory	ProviderFactory[T]	Callable that creates instances	required
scope	Scope	Lifecycle scope (singleton, request, task)	SINGLETON
qualifier	str | None	Optional qualifier for multiple providers of same type	None
name	str | None	Optional name for resolution	None
reloadable	tuple[str, ...]	Tuple of field names that can be hot-reloaded	()

Raises:

Type	Description
ContainerFrozenError	If container has been compiled
DuplicateProviderError	If provider already registered

Source code in packages/myfy-core/myfy/core/di/container.py

def register(
    self,
    type_: type[T],
    factory: ProviderFactory[T],
    *,
    scope: Scope = SINGLETON,
    qualifier: str | None = None,
    name: str | None = None,
    reloadable: tuple[str, ...] = (),
) -> None:
    """
    Register a provider in the container.

    Args:
        type_: The type this provider produces
        factory: Callable that creates instances
        scope: Lifecycle scope (singleton, request, task)
        qualifier: Optional qualifier for multiple providers of same type
        name: Optional name for resolution
        reloadable: Tuple of field names that can be hot-reloaded

    Raises:
        ContainerFrozenError: If container has been compiled
        DuplicateProviderError: If provider already registered
    """
    if self._frozen:
        raise ContainerFrozenError

    key = ProviderKey(type_, qualifier, name)

    if key in self._providers:
        raise DuplicateProviderError(str(key))

    registration = ProviderRegistration(key, factory, scope, reloadable)
    self._providers[key] = registration

compile

compile()

Compile all providers - analyze dependencies and build injection plans.

This must be called before resolving dependencies. It: 1. Analyzes all provider factories for dependencies 2. Detects circular dependencies 3. Validates scope compatibility 4. Builds injection plans for fast resolution

Raises:

Type	Description
CircularDependencyError	If circular dependencies detected
ScopeMismatchError	If scope rules violated
ProviderNotFoundError	If required dependency missing

Source code in packages/myfy-core/myfy/core/di/container.py

def compile(self) -> None:
    """
    Compile all providers - analyze dependencies and build injection plans.

    This must be called before resolving dependencies. It:
    1. Analyzes all provider factories for dependencies
    2. Detects circular dependencies
    3. Validates scope compatibility
    4. Builds injection plans for fast resolution

    Raises:
        CircularDependencyError: If circular dependencies detected
        ScopeMismatchError: If scope rules violated
        ProviderNotFoundError: If required dependency missing
    """
    if self._frozen:
        return

    # Analyze dependencies for each provider
    for registration in self._providers.values():
        self._analyze_dependencies(registration)

    # Detect cycles
    for key in self._providers:
        self._check_cycles(key, [])

    # Validate scope rules
    self._validate_scopes()

    # Build injection plans
    for registration in self._providers.values():
        registration.injection_plan = self._build_injection_plan(registration)

    self._frozen = True

get

get(type_, qualifier=None)

Resolve and return an instance of the requested type.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type to resolve	required
qualifier	str | None	Optional qualifier to select specific provider	None

Returns:

Type	Description
T	Instance of the requested type

Raises:

Type	Description
ProviderNotFoundError	If no provider found

Source code in packages/myfy-core/myfy/core/di/container.py

def get(self, type_: type[T], qualifier: str | None = None) -> T:
    """
    Resolve and return an instance of the requested type.

    Args:
        type_: The type to resolve
        qualifier: Optional qualifier to select specific provider

    Returns:
        Instance of the requested type

    Raises:
        ProviderNotFoundError: If no provider found
    """
    if not self._frozen:
        raise DIError("Container must be compiled before resolving dependencies")

    key = ProviderKey(type_, qualifier)
    registration = self._providers.get(key)

    if registration is None:
        raise ProviderNotFoundError(str(key))

    return self._resolve(registration)

override

override(overrides)

Temporarily override providers for testing.

Usage

with container.override({Database: lambda: FakeDB()}): # Tests run with fake database ...

Source code in packages/myfy-core/myfy/core/di/container.py

@contextmanager
def override(self, overrides: dict[type, ProviderFactory]):
    """
    Temporarily override providers for testing.

    Usage:
        with container.override({Database: lambda: FakeDB()}):
            # Tests run with fake database
            ...
    """
    self._override_stack.append(overrides)
    self._has_active_overrides = True
    try:
        yield
    finally:
        self._override_stack.pop()
        self._has_active_overrides = len(self._override_stack) > 0

Modules

BaseModule

BaseModule(name)

Bases: ABC

Base implementation of the Module protocol.

Provides default implementations and helper methods.

Source code in packages/myfy-core/myfy/core/kernel/module.py

def __init__(self, name: str):
    self._name = name

requires property

requires

Module dependencies (default: none).

Override to declare dependencies

@property def requires(self) -> list[type]: return [WebModule, DataModule]

provides property

provides

Extension protocols this module implements (default: none).

Override to declare protocols

@property def provides(self) -> list[type]: return [IWebExtension]

configure abstractmethod

configure(container)

Configure the module by registering providers.

Must be implemented by subclasses.

Source code in packages/myfy-core/myfy/core/kernel/module.py

@abstractmethod
def configure(self, container: Container) -> None:
    """
    Configure the module by registering providers.

    Must be implemented by subclasses.
    """

extend

extend(container)

Extend other modules' service registrations (optional).

Default no-op implementation. Override if needed.

Source code in packages/myfy-core/myfy/core/kernel/module.py

def extend(self, container: Container) -> None:  # noqa: B027
    """
    Extend other modules' service registrations (optional).

    Default no-op implementation. Override if needed.
    """

finalize

finalize(container)

Finalize module configuration after container compilation (optional).

Default no-op implementation. Override if needed.

Source code in packages/myfy-core/myfy/core/kernel/module.py

def finalize(self, container: Container) -> None:  # noqa: B027
    """
    Finalize module configuration after container compilation (optional).

    Default no-op implementation. Override if needed.
    """

start async

start()

Default start implementation (no-op).

Override if your module needs startup logic.

Source code in packages/myfy-core/myfy/core/kernel/module.py

async def start(self) -> None:
    """
    Default start implementation (no-op).

    Override if your module needs startup logic.
    """
    # Default no-op implementation
    return

stop async

stop()

Default stop implementation (no-op).

Override if your module needs cleanup logic.

Source code in packages/myfy-core/myfy/core/kernel/module.py

async def stop(self) -> None:
    """
    Default stop implementation (no-op).

    Override if your module needs cleanup logic.
    """
    # Default no-op implementation
    return

Configuration

BaseSettings

Bases: BaseSettings

Base class for application settings.

Inherit from this to create typed, validated configuration. Supports environment variables and .env files.

Usage

class AppSettings(BaseSettings): app_name: str = "myapp" db_url: str debug: bool = False

model_config = SettingsConfigDict(
    env_prefix="MYAPP_",
    env_file=".env"
)

model_dump_safe

model_dump_safe(**kwargs)

Dump settings with secrets redacted.

Useful for logging and health check endpoints. Redacts fields containing: password, secret, token, key, api_key, private

Source code in packages/myfy-core/myfy/core/config/settings.py

def model_dump_safe(self, **kwargs: Any) -> dict[str, Any]:
    """
    Dump settings with secrets redacted.

    Useful for logging and health check endpoints.
    Redacts fields containing: password, secret, token, key, api_key, private
    """
    dump = self.model_dump(**kwargs)

    # Redact fields that commonly contain secrets
    secret_patterns = ["password", "secret", "token", "key", "api_key", "private"]

    for field_name in list(dump.keys()):
        field_lower = field_name.lower()
        if any(pattern in field_lower for pattern in secret_patterns):
            dump[field_name] = "***REDACTED***"

    return dump

Scopes

scopes

Scope management for dependency injection.

Scopes control the lifetime of injected dependencies: - SINGLETON: One instance per application - REQUEST: One instance per request (via contextvar) - TASK: One instance per background task (via contextvar)

Scope

Bases: str, Enum

Dependency scope definitions.

ScopeContext

Manages scoped dependency instances via contextvars.

Request and task scopes store instances in contextvar dictionaries, ensuring thread-safe and async-safe isolation.

get_request_bag staticmethod

get_request_bag()

Get the request scope bag for the current context.

Raises RuntimeError if bag not initialized. The ASGI adapter should call init_request_scope() before handler execution.

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def get_request_bag() -> dict[str, Any]:
    """
    Get the request scope bag for the current context.

    Raises RuntimeError if bag not initialized.
    The ASGI adapter should call init_request_scope() before handler execution.
    """
    bag = _request_scope_bag.get()
    if bag is None:
        raise RuntimeError(
            "Request scope not initialized. "
            "Ensure the ASGI adapter sets up the request bag before handler execution."
        )
    return bag

get_task_bag staticmethod

get_task_bag()

Get the task scope bag for the current context.

Raises RuntimeError if bag not initialized.

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def get_task_bag() -> dict[str, Any]:
    """
    Get the task scope bag for the current context.

    Raises RuntimeError if bag not initialized.
    """
    bag = _task_scope_bag.get()
    if bag is None:
        raise RuntimeError(
            "Task scope not initialized. "
            "Ensure the task runner sets up the task bag before execution."
        )
    return bag

init_request_scope staticmethod

init_request_scope()

Initialize a new request scope (called by ASGI adapter).

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def init_request_scope() -> dict[str, Any]:
    """Initialize a new request scope (called by ASGI adapter)."""
    bag = {}
    _request_scope_bag.set(bag)
    return bag

init_task_scope staticmethod

init_task_scope()

Initialize a new task scope (called by task runner).

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def init_task_scope() -> dict[str, Any]:
    """Initialize a new task scope (called by task runner)."""
    bag = {}
    _task_scope_bag.set(bag)
    return bag

clear_request_bag staticmethod

clear_request_bag()

Clear the request scope bag (call after request finishes).

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def clear_request_bag() -> None:
    """Clear the request scope bag (call after request finishes)."""
    _request_scope_bag.set(None)

clear_task_bag staticmethod

clear_task_bag()

Clear the task scope bag (call after task finishes).

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def clear_task_bag() -> None:
    """Clear the task scope bag (call after task finishes)."""
    _task_scope_bag.set(None)

get_bag_for_scope staticmethod

get_bag_for_scope(scope)

Get the appropriate bag for the given scope.

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def get_bag_for_scope(scope: Scope) -> dict[str, Any] | None:
    """Get the appropriate bag for the given scope."""
    if scope == Scope.REQUEST:
        return ScopeContext.get_request_bag()
    if scope == Scope.TASK:
        return ScopeContext.get_task_bag()
    return None  # Singleton doesn't use a bag

clear_scope staticmethod

clear_scope(scope)

Clear the given scope's bag.

Source code in packages/myfy-core/myfy/core/di/scopes.py

@staticmethod
def clear_scope(scope: Scope) -> None:
    """Clear the given scope's bag."""
    if scope == Scope.REQUEST:
        ScopeContext.clear_request_bag()
    elif scope == Scope.TASK:
        ScopeContext.clear_task_bag()

---

Quick Links

• Dependency Injection Guide
• Modules Guide
• Configuration Guide