spakky

spakky는 DI 컨테이너, AOP, 애플리케이션 부트스트랩, Pod 스캔을 제공하는 핵심 패키지입니다.

DI 컨테이너, AOP, 부트스트랩 — Spakky Framework의 핵심 패키지

Application

spakky.core.application.application

Main application class for bootstrapping the Spakky framework.

This module provides the SpakkyApplication class which serves as the entry point for configuring and starting a Spakky application with DI/IoC and AOP support.

CannotDetermineScanPathError

Bases: AbstractSpakkyApplicationError

Raised when the scan path cannot be automatically determined.

SpakkyApplication(application_context)

Main application class for bootstrapping Spakky framework.

Provides a fluent API for configuring dependency injection, aspect-oriented programming, plugin loading, and component scanning.

Initialize the Spakky application.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context to manage Pods.	required

Source code in core/spakky/src/spakky/core/application/application.py

def __init__(self, application_context: IApplicationContext) -> None:
    """Initialize the Spakky application.

    Args:
        application_context: The application context to manage Pods.
    """
    self._application_context = application_context
    self._startup_phase_recorder = NoOpStartupPhaseRecorder()
    self._discovery_manifest_path = None

container property

Get the IoC container.

Returns:

Type	Description
IContainer	The application's dependency injection container.

application_context property

Get the application context.

Returns:

Type	Description
IApplicationContext	The application's context managing Pods and lifecycle.

startup_phase_recorder property

Get the startup phase recorder.

Returns:

Type	Description
IStartupPhaseRecorder	Active or no-op startup phase recorder.

startup_report property

Get the startup diagnostics report.

Returns:

Type	Description
StartupReport	Startup report backing the current phase recorder.

discovery_manifest_path property

Get the configured DiscoveryManifest path.

Returns:

Type	Description
Path | None	Manifest path when reuse is enabled, otherwise None.

enable_startup_diagnostics()

Enable startup diagnostics with an active phase recorder.

Returns:

Type	Description
Self	Self for method chaining.

Source code in core/spakky/src/spakky/core/application/application.py

def enable_startup_diagnostics(self) -> Self:
    """Enable startup diagnostics with an active phase recorder.

    Returns:
        Self for method chaining.
    """
    self._startup_phase_recorder = ActiveStartupPhaseRecorder()
    return self

enable_discovery_manifest(path=None)

Enable reusable scan discovery manifests.

Parameters:

Name	Type	Description	Default
path	Path | str | None	Explicit manifest path. None uses a deterministic project cache path.	None

Returns:

Type	Description
Self	Self for method chaining.

Source code in core/spakky/src/spakky/core/application/application.py

def enable_discovery_manifest(self, path: Path | str | None = None) -> Self:
    """Enable reusable scan discovery manifests.

    Args:
        path: Explicit manifest path. None uses a deterministic project cache path.

    Returns:
        Self for method chaining.
    """
    self._discovery_manifest_path = (
        default_discovery_manifest_path() if path is None else Path(path)
    )
    return self

add(obj)

Register a class or function in the application.

• If the object has a @Pod annotation, it is registered in the container.
• If the object has a @Tag annotation, the tag is registered in the tag registry.

Parameters:

Name	Type	Description	Default
obj	PodType	The class or function to register.	required

Returns:

Type	Description
Self	Self for method chaining.

Source code in core/spakky/src/spakky/core/application/application.py

def add(self, obj: PodType) -> Self:
    """Register a class or function in the application.

    - If the object has a @Pod annotation, it is registered in the container.
    - If the object has a @Tag annotation, the tag is registered in the tag registry.

    Args:
        obj: The class or function to register.

    Returns:
        Self for method chaining.
    """
    if Pod.exists(obj):
        self._application_context.add(obj)
    if Tag.exists(obj):
        tag = Tag.get(obj)
        self._application_context.register_tag(tag)
    return self

scan(path=None, exclude=None)

Scan a module for Pod-annotated classes and functions.

When path is None, automatically detects the caller's package and scans it. If the caller's package is not importable (e.g., in Docker environments where the application root is not in sys.path), the parent directory is automatically added to sys.path to enable package discovery.

Parameters:

Name	Type	Description	Default
path	Module | None	Module or package to scan. If None, scans the caller's package.	None
exclude	set[Module] | None	Set of modules to exclude from scanning.	None

Returns:

Type	Description
Self	Self for method chaining.

Raises:

Type	Description
CannotDetermineScanPathError	If path is None and cannot determine caller's package.

Source code in core/spakky/src/spakky/core/application/application.py

def scan(
    self,
    path: Module | None = None,
    exclude: set[Module] | None = None,
) -> Self:
    """Scan a module for Pod-annotated classes and functions.

    When path is None, automatically detects the caller's package and scans it.
    If the caller's package is not importable (e.g., in Docker environments where
    the application root is not in sys.path), the parent directory is automatically
    added to sys.path to enable package discovery.

    Args:
        path: Module or package to scan. If None, scans the caller's package.
        exclude: Set of modules to exclude from scanning.

    Returns:
        Self for method chaining.

    Raises:
        CannotDetermineScanPathError: If path is None and cannot determine caller's package.
    """
    modules: set[ModuleType]
    caller_module: ModuleType | None = None
    manifest_decision: DiscoveryManifestDecision | None = None
    manifest_fingerprint: DiscoveryManifestFingerprint | None = None
    discovery_candidates: list[DiscoveryManifestCandidate] = []
    discovered_objects: tuple[PodType, ...] = ()
    with self._startup_phase_recorder.record_phase(
        phase_name=STARTUP_PHASE_SCAN
    ) as scan_phase:
        if path is None:  # pragma: no cover - coverage boundary
            caller_frame = inspect.stack()[1]
            caller_file = caller_frame.filename
            caller_dir = Path(caller_file).parent

            # Check if caller is inside a package (has __init__.py)
            if not (caller_dir / "__init__.py").exists():
                raise CannotDetermineScanPathError

            # Ensure the package is importable (adds to sys.path if needed)
            ensure_importable(caller_dir)

            try:
                path = resolve_module(caller_dir.name)
            except ImportError as e:
                raise CannotDetermineScanPathError from e

            caller_module = inspect.getmodule(caller_frame[0])

        if exclude is None:
            exclude = {caller_module} if caller_module else set()
        if self._discovery_manifest_path is not None:
            manifest_fingerprint = DiscoveryManifestFingerprint.from_scan_input(
                path=path,
                exclude=exclude,
            )
            manifest_load_result = DiscoveryManifestStore(
                self._discovery_manifest_path
            ).load(manifest_fingerprint)
            manifest_decision = manifest_load_result.decision
            if manifest_load_result.manifest is not None:
                manifest_hit = self._resolve_manifest_hit(
                    manifest_load_result.manifest.candidates
                )
                manifest_decision = manifest_hit.decision
                discovered_objects = manifest_hit.objects
                if manifest_decision is DiscoveryManifestDecision.HIT:
                    modules = {
                        resolve_module(module_name)
                        for module_name in manifest_load_result.manifest.module_names
                    }
                else:
                    modules = self._discover_modules(path, exclude)
            else:
                modules = self._discover_modules(path, exclude)
            scan_phase.set_diagnostic_details(
                (
                    StartupDiagnosticDetail(
                        key=DISCOVERY_MANIFEST_DETAIL_KEY,
                        value=manifest_decision.value,
                    ),
                )
            )
        else:
            modules = self._discover_modules(path, exclude)
        scan_phase.set_processed_count(len(modules))

    registration_count = 0
    with self._startup_phase_recorder.record_phase(
        phase_name=STARTUP_PHASE_REGISTRATION
    ) as registration_phase:
        if manifest_decision is DiscoveryManifestDecision.HIT:
            for obj in discovered_objects:
                registration_count += self._register_discovered_object(obj)
        else:
            for item in modules:
                objects = list_objects(
                    item,
                    lambda x: Pod.exists(x) or Tag.exists(x),
                )
                for obj in objects:
                    discovery_candidates.append(
                        DiscoveryManifestCandidate.from_object(obj)
                    )
                    registration_count += self._register_discovered_object(obj)
        registration_phase.set_processed_count(registration_count)

    if (
        self._discovery_manifest_path is not None
        and manifest_fingerprint is not None
        and manifest_decision is not DiscoveryManifestDecision.HIT
    ):
        manifest = DiscoveryManifest(
            fingerprint=manifest_fingerprint,
            module_names=tuple(sorted(item.__name__ for item in modules)),
            candidates=tuple(
                sorted(
                    discovery_candidates,
                    key=lambda candidate: (
                        candidate.module_name,
                        candidate.qualname,
                    ),
                )
            ),
        )
        DiscoveryManifestStore(self._discovery_manifest_path).save(manifest)

    return self

load_plugins(include=None)

Load plugins from entry points.

Parameters:

Name	Type	Description	Default
include	set[Plugin] | None	Optional set of plugins to load. If None, loads all available plugins.	None

Returns:

Type	Description
Self	Self for method chaining.

Source code in core/spakky/src/spakky/core/application/application.py

def load_plugins(
    self,
    include: set[Plugin] | None = None,
) -> Self:
    """Load plugins from entry points.

    Args:
        include: Optional set of plugins to load. If None, loads all available plugins.

    Returns:
        Self for method chaining.
    """
    loaded_count = 0
    loaded_base_plugins: set[Plugin] = set()
    available_feature_plugins: set[Plugin] = set()
    contribution_diagnostics = _ContributionDiagnostics()
    with self._startup_phase_recorder.record_phase(
        phase_name=STARTUP_PHASE_LOAD_PLUGINS
    ) as plugin_phase:
        base_entry_points = sorted(
            entry_points(group=PLUGIN_PATH),
            key=lambda entry_point: entry_point.name,
        )
        available_feature_plugins = {
            Plugin(name=entry_point.name)
            for entry_point in base_entry_points
            if _is_core_feature_entry_point(entry_point)
        }
        for entry_point in base_entry_points:
            plugin = Plugin(name=entry_point.name)
            if include is not None and plugin not in include:
                continue
            entry_point_function: Callable[[SpakkyApplication], None] = (
                entry_point.load()
            )
            loaded_count += 1
            plugin_phase.set_processed_count(loaded_count)
            entry_point_function(self)
            loaded_base_plugins.add(plugin)
        for feature_plugin in sorted(
            available_feature_plugins,
            key=lambda plugin: plugin.name,
        ):
            group = _contribution_entry_point_group(feature_plugin)
            contribution_entry_points = sorted(
                entry_points(group=group),
                key=lambda entry_point: entry_point.name,
            )
            for contribution_entry_point in contribution_entry_points:
                provider_plugins = _provider_plugins_for_contribution(
                    contribution_entry_point
                )
                skip_reason = _contribution_skip_reason(
                    feature_plugin=feature_plugin,
                    contribution_entry_point=contribution_entry_point,
                    loaded_base_plugins=loaded_base_plugins,
                    include=include,
                )
                if skip_reason is not None:
                    contribution_diagnostics.record_skipped(
                        reason=skip_reason,
                        group=group,
                        entry_point=contribution_entry_point,
                        provider_plugins=provider_plugins,
                        feature_plugin=feature_plugin,
                    )
                    continue
                try:
                    entry_point_function = contribution_entry_point.load()
                    entry_point_function(self)
                except Exception:
                    contribution_diagnostics.record_failed(
                        group=group,
                        entry_point=contribution_entry_point,
                        provider_plugins=provider_plugins,
                        feature_plugin=feature_plugin,
                    )
                    plugin_phase.set_diagnostic_details(
                        contribution_diagnostics.details()
                    )
                    raise
                loaded_count += 1
                plugin_phase.set_processed_count(loaded_count)
                contribution_diagnostics.record_loaded(
                    group=group,
                    entry_point=contribution_entry_point,
                    provider_plugins=provider_plugins,
                    feature_plugin=feature_plugin,
                )
        plugin_phase.set_diagnostic_details(contribution_diagnostics.details())
    return self

start()

Start the application by initializing all Pods and running post-processors.

Returns:

Type	Description
Self	Self for method chaining.

Source code in core/spakky/src/spakky/core/application/application.py

def start(self) -> Self:
    """Start the application by initializing all Pods and running post-processors.

    Returns:
        Self for method chaining.
    """
    self._application_context.start(
        startup_phase_recorder=self._startup_phase_recorder
    )
    return self

stop()

Stop the application and clean up resources.

Returns:

Type	Description
Self	Self for method chaining.

Source code in core/spakky/src/spakky/core/application/application.py

def stop(self) -> Self:
    """Stop the application and clean up resources.

    Returns:
        Self for method chaining.
    """
    self._application_context.stop()
    return self

options: show_root_heading: false

spakky.core.application.application_context

CannotAssignSystemContextIDError

Bases: AbstractSpakkyApplicationError

Raised when attempting to override the CONTEXT_ID value.

ApplicationContext()

Bases: IApplicationContext

Container managing Pod instances, dependencies, and application lifecycle.

ApplicationContext is responsible for: - Registering and instantiating Pods with dependency injection - Managing Pod scopes (SINGLETON, PROTOTYPE, CONTEXT) - Running post-processors on Pod instances - Coordinating service lifecycle (start/stop) - Managing async event loop for async services

Initialize application context.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __init__(self) -> None:
    """Initialize application context."""
    self.__forward_type_map = {}
    self.__pods = {}
    self.__tags = set()
    self.__type_cache = {}
    self.__bindings = {}
    self.__singleton_cache = {}
    self.__singleton_lock = RLock()
    self.__shutdown_lock = RLock()
    self.__context_cache = ContextVar(CONTEXT_SCOPE_CACHE)
    self.__post_processors = []
    self.__services = []
    self.__async_services = []
    self.__event_loop = None
    self.__event_thread = None
    self.__is_started = False
    self.__startup_metrics = None
    self.task_stop_event = locks.Event()
    self.thread_stop_event = threading.Event()

__forward_type_map = {} instance-attribute

Map for resolving forward reference types.

__pods = {} instance-attribute

Registry of all Pods by name.

__tags = set() instance-attribute

Registry of all Tags.

__type_cache = {} instance-attribute

Cache mapping types to Pods for O(1) lookup.

__bindings = {} instance-attribute

Explicit interface-to-implementation binding policies.

__singleton_cache = {} instance-attribute

Cache of singleton-scoped Pod instances.

__context_cache = ContextVar(CONTEXT_SCOPE_CACHE) instance-attribute

Context-local cache for context-scoped Pods.

__post_processors = [] instance-attribute

List of post-processors applied to Pod instances.

__services = [] instance-attribute

List of synchronous services.

__async_services = [] instance-attribute

List of asynchronous services.

__event_loop = None instance-attribute

Event loop for running async services.

__event_thread = None instance-attribute

Thread running the event loop.

__is_started = False instance-attribute

Whether the context has been started.

__startup_metrics = None instance-attribute

Startup lifecycle metrics for the active start attempt.

pods property

Get read-only view of all registered Pods.

Returns:

Type	Description
dict[str, Pod]	Read-only mapping proxy of Pod registry (O(1) operation).

tags property

Get read-only view of all registered Tags.

Returns:

Type	Description
frozenset[Tag]	Read-only frozenset of Tag registry (O(1) operation).

is_started property

Check if context has been started.

Returns:

Type	Description
bool	True if started.

__type_name(type_)

Return a stable diagnostic name for a dependency type.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __type_name(self, type_: object) -> str:
    """Return a stable diagnostic name for a dependency type."""
    if isinstance(type_, type):
        return type_.__name__
    return str(type_)

__dependency_path_node(pod, dependency_parameter_name=None, requested_type=None)

Create one dependency path node from registered Pod metadata.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __dependency_path_node(
    self,
    pod: Pod,
    dependency_parameter_name: str | None = None,
    requested_type: object | None = None,
) -> PodDependencyPathNode:
    """Create one dependency path node from registered Pod metadata."""
    requested_type_name = (
        None if requested_type is None else self.__type_name(requested_type)
    )
    return PodDependencyPathNode(
        pod_name=pod.name,
        pod_type_name=self.__type_name(pod.type_),
        dependency_parameter_name=dependency_parameter_name,
        requested_type_name=requested_type_name,
    )

__dependency_diagnostic(pod, dependency_parameter_name, requested_type, dependency_path, candidates=(), resolution_hints=())

Build structured diagnostics from the active Pod dependency path.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __dependency_diagnostic(
    self,
    pod: Pod,
    dependency_parameter_name: str | None,
    requested_type: object | None,
    dependency_path: tuple[PodDependencyPathNode, ...],
    candidates: tuple[PodCandidateDiagnostic, ...] = (),
    resolution_hints: tuple[str, ...] = (),
) -> PodDependencyResolutionDiagnostic:
    """Build structured diagnostics from the active Pod dependency path."""
    requested_type_name = (
        None if requested_type is None else self.__type_name(requested_type)
    )
    return PodDependencyResolutionDiagnostic(
        failed_pod_name=pod.name,
        failed_pod_type_name=self.__type_name(pod.type_),
        dependency_parameter_name=dependency_parameter_name,
        requested_type_name=requested_type_name,
        path=dependency_path,
        candidates=candidates,
        resolution_hints=resolution_hints,
    )

__startup_diagnostic_details(exception)

Convert dependency resolution diagnostics into startup report details.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __startup_diagnostic_details(
    self,
    exception: BaseException,
) -> StartupDiagnosticDetails:
    """Convert dependency resolution diagnostics into startup report details."""
    dependency_diagnostic: PodDependencyResolutionDiagnostic | None = None
    if isinstance(exception, UnexpectedDependencyTypeInjectedError):
        dependency_diagnostic = exception.dependency_diagnostic
    if isinstance(exception, NoUniquePodError):
        dependency_diagnostic = exception.dependency_diagnostic
    if isinstance(exception, CircularDependencyGraphDetectedError):
        dependency_diagnostic = exception.dependency_diagnostic
    if dependency_diagnostic is None:
        return ()
    return tuple(
        StartupDiagnosticDetail(key=key, value=value)
        for key, value in dependency_diagnostic.as_detail_pairs()
    )

__candidate_diagnostics(pods)

Return stable ambiguity diagnostics for candidate Pods.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __candidate_diagnostics(
    self,
    pods: set[Pod],
) -> tuple[PodCandidateDiagnostic, ...]:
    """Return stable ambiguity diagnostics for candidate Pods."""
    return tuple(
        PodCandidateDiagnostic(
            pod_name=pod.name,
            pod_type_name=self.__type_name(pod.type_),
            is_primary=pod.is_primary,
        )
        for pod in sorted(pods, key=lambda candidate: candidate.name)
    )

__ambiguity_hints(dependency_parameter_name)

Return resolution hints for a single dependency ambiguity.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __ambiguity_hints(
    self, dependency_parameter_name: str | None
) -> tuple[str, ...]:
    """Return resolution hints for a single dependency ambiguity."""
    hints = (
        "add Annotated[T, Qualifier(...)] to select one candidate",
        "register ApplicationContext.bind(PodBinding(...)) from application config",
        "mark exactly one candidate with @Primary",
    )
    if dependency_parameter_name is None:
        return hints + ("call get(type_, name=...) for an explicit Pod name",)
    return hints + (
        "rename the dependency parameter to one candidate Pod name as legacy fallback",
    )

__ambiguity_diagnostic(requester_pod, dependency_parameter_name, requested_type, dependency_path, candidates, resolution_hints)

Build dependency diagnostics when ambiguity occurs during injection.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __ambiguity_diagnostic(
    self,
    requester_pod: Pod | None,
    dependency_parameter_name: str | None,
    requested_type: type,
    dependency_path: tuple[PodDependencyPathNode, ...],
    candidates: tuple[PodCandidateDiagnostic, ...],
    resolution_hints: tuple[str, ...],
) -> PodDependencyResolutionDiagnostic | None:
    """Build dependency diagnostics when ambiguity occurs during injection."""
    if requester_pod is None:
        return None
    return self.__dependency_diagnostic(
        pod=requester_pod,
        dependency_parameter_name=dependency_parameter_name,
        requested_type=requested_type,
        dependency_path=dependency_path,
        candidates=candidates,
        resolution_hints=resolution_hints,
    )

__binding_target_count(binding)

Return how many target selectors a binding specifies.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __binding_target_count(self, binding: PodBinding) -> int:
    """Return how many target selectors a binding specifies."""
    count = 0
    if binding.implementation_type is not None:
        count += 1
    if binding.implementation_name is not None:
        count += 1
    return count

__validate_binding(binding)

Validate binding shape without requiring Pods to be registered yet.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __validate_binding(self, binding: PodBinding) -> None:
    """Validate binding shape without requiring Pods to be registered yet."""
    if self.__binding_target_count(binding) != 1:
        raise InvalidPodBindingError

__resolve_binding_candidate(type_, pods)

Resolve an explicit binding for this requested type, when configured.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __resolve_binding_candidate(
    self,
    type_: type,
    pods: set[Pod],
) -> Pod | None:
    """Resolve an explicit binding for this requested type, when configured."""
    binding = self.__bindings.get(type_)
    if binding is None:
        return None
    if binding.implementation_name is not None:
        named = {pod for pod in pods if pod.name == binding.implementation_name}
        if len(named) == 1:
            return named.pop()
        raise NoSuchPodBindingTargetError(binding)
    typed = {pod for pod in pods if pod.type_ == binding.implementation_type}
    if len(typed) == 1:
        return typed.pop()
    raise NoSuchPodBindingTargetError(binding)

__candidate_pods_for_type(type_)

Return exact candidates, falling back to generic origin candidates.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __candidate_pods_for_type(self, type_: type) -> set[Pod]:
    """Return exact candidates, falling back to generic origin candidates."""
    pods = self.__type_cache.get(type_, set()).copy()
    if pods:
        return pods
    origin = get_origin(type_)
    if origin is None:
        return pods
    return self.__type_cache.get(origin, set()).copy()

__resolve_binding_candidate_for_type(type_, pods)

Resolve exact binding first, then generic-origin binding.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __resolve_binding_candidate_for_type(
    self,
    type_: type,
    pods: set[Pod],
) -> Pod | None:
    """Resolve exact binding first, then generic-origin binding."""
    if binding_candidate := self.__resolve_binding_candidate(type_, pods):
        return binding_candidate
    origin = get_origin(type_)
    if origin is None:
        return None
    return self.__resolve_binding_candidate(origin, pods)

__resolve_collection_candidates(type_, qualifiers)

Resolve all collection dependency candidates in stable Pod name order.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __resolve_collection_candidates(
    self,
    type_: type,
    qualifiers: list[Qualifier],
) -> tuple[Pod, ...]:
    """Resolve all collection dependency candidates in stable Pod name order."""
    pods = self.__candidate_pods_for_type(type_)
    if qualifiers:
        pods = {
            pod
            for pod in pods
            if all(qualifier.selector(pod) for qualifier in qualifiers)
        }
    return tuple(sorted(pods, key=lambda pod: pod.name))

__index_type_cache(type_, pod)

Index one Pod under a runtime-resolvable type key.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __index_type_cache(self, type_: type, pod: Pod) -> None:
    """Index one Pod under a runtime-resolvable type key."""
    if type_ not in self.__type_cache:
        self.__type_cache[type_] = set()
    self.__type_cache[type_].add(pod)

__resolve_candidate(type_, name, qualifiers, name_is_dependency_parameter, requester_pod, dependency_path)

Resolve a Pod candidate matching type, name, and qualifiers.

Parameters:

Name	Type	Description	Default
type_	type	The type to search for.	required
name	str | None	Optional name qualifier.	required
qualifiers	list[Qualifier]	List of qualifier annotations.	required

Returns:

Type	Description
Pod | None	Matching Pod or None if not found.

Raises:

Type	Description
NoUniquePodError	If multiple Pods match without clear qualification.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __resolve_candidate(
    self,
    type_: type,
    name: str | None,
    qualifiers: list[Qualifier],
    name_is_dependency_parameter: bool,
    requester_pod: Pod | None,
    dependency_path: tuple[PodDependencyPathNode, ...],
) -> Pod | None:
    """Resolve a Pod candidate matching type, name, and qualifiers.

    Args:
        type_: The type to search for.
        name: Optional name qualifier.
        qualifiers: List of qualifier annotations.

    Returns:
        Matching Pod or None if not found.

    Raises:
        NoUniquePodError: If multiple Pods match without clear qualification.
    """

    # Use type index for O(1) lookup instead of O(n) iteration.
    # Parameterized generics are matched exactly first, then by runtime origin.
    pods = self.__candidate_pods_for_type(type_)
    if not pods:
        return None

    if qualifiers:
        qualified = {
            pod
            for pod in pods
            if all(qualifier.selector(pod) for qualifier in qualifiers)
        }
        if len(qualified) == 1:
            return qualified.pop()
        if not qualified:
            return None
        candidates = self.__candidate_diagnostics(qualified)
        resolution_hints = self.__ambiguity_hints(name)
        raise NoUniquePodError(
            type_,
            candidates,
            self.__ambiguity_diagnostic(
                requester_pod=requester_pod,
                dependency_parameter_name=name,
                requested_type=type_,
                dependency_path=dependency_path,
                candidates=candidates,
                resolution_hints=resolution_hints,
            ),
            resolution_hints,
        )

    if name is not None and not name_is_dependency_parameter:
        named = {pod for pod in pods if pod.name == name}
        if len(named) == 1:
            return named.pop()
        if not named:
            return None

    if binding_candidate := self.__resolve_binding_candidate_for_type(type_, pods):
        return binding_candidate

    # Fast path after explicit selectors and binding policy are honored.
    if len(pods) == 1:
        return next(iter(pods))

    primary = {pod for pod in pods if pod.is_primary}
    if len(primary) == 1:
        return primary.pop()
    if len(primary) > 1:
        candidates = self.__candidate_diagnostics(primary)
        resolution_hints = self.__ambiguity_hints(name)
        raise NoUniquePodError(
            type_,
            candidates,
            self.__ambiguity_diagnostic(
                requester_pod=requester_pod,
                dependency_parameter_name=name,
                requested_type=type_,
                dependency_path=dependency_path,
                candidates=candidates,
                resolution_hints=resolution_hints,
            ),
            resolution_hints,
        )

    if name is not None and name_is_dependency_parameter:
        legacy_named = {pod for pod in pods if pod.name == name}
        if len(legacy_named) == 1:
            return legacy_named.pop()

    candidates = self.__candidate_diagnostics(pods)
    resolution_hints = self.__ambiguity_hints(name)
    raise NoUniquePodError(
        type_,
        candidates,
        self.__ambiguity_diagnostic(
            requester_pod=requester_pod,
            dependency_parameter_name=name,
            requested_type=type_,
            dependency_path=dependency_path,
            candidates=candidates,
            resolution_hints=resolution_hints,
        ),
        resolution_hints,
    )

__instantiate_pod(pod, dependency_hierarchy, dependency_path)

Instantiate a Pod with its dependencies recursively resolved.

Parameters:

Name	Type	Description	Default
pod	Pod	The Pod to instantiate.	required
dependency_hierarchy	tuple[type, ...]	Immutable tuple tracking dependency chain for cycle detection.	required

Returns:

Type	Description
object	The instantiated and post-processed Pod instance.

Raises:

Type	Description
CircularDependencyGraphDetectedError	If circular dependency detected.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __instantiate_pod(
    self,
    pod: Pod,
    dependency_hierarchy: tuple[type, ...],
    dependency_path: tuple[PodDependencyPathNode, ...],
) -> object:
    """Instantiate a Pod with its dependencies recursively resolved.

    Args:
        pod: The Pod to instantiate.
        dependency_hierarchy: Immutable tuple tracking dependency chain for cycle detection.

    Returns:
        The instantiated and post-processed Pod instance.

    Raises:
        CircularDependencyGraphDetectedError: If circular dependency detected.
    """
    if pod.type_ in dependency_hierarchy:
        circular_path = dependency_path + (self.__dependency_path_node(pod),)
        last_node = dependency_path[-1] if dependency_path else None
        raise CircularDependencyGraphDetectedError(
            list(dependency_hierarchy) + [pod.type_],
            dependency_diagnostic=self.__dependency_diagnostic(
                pod=pod,
                dependency_parameter_name=None
                if last_node is None
                else last_node.dependency_parameter_name,
                requested_type=None
                if last_node is None
                else last_node.requested_type_name,
                dependency_path=circular_path,
            ),
        )
    new_hierarchy = dependency_hierarchy + (pod.type_,)
    dependencies: dict[str, object | None] = {}
    for name, dependency in pod.dependencies.items():
        requested_type = (
            remove_none(dependency.type_)
            if is_optional(dependency.type_)
            else dependency.type_
        )
        current_path = dependency_path + (
            self.__dependency_path_node(
                pod=pod,
                dependency_parameter_name=name,
                requested_type=requested_type,
            ),
        )
        if dependency.collection_kind is None:
            resolved_dependency = self.__get_internal(
                type_=remove_none(dependency.type_)
                if is_optional(dependency.type_)
                else dependency.type_,
                name=name,
                dependency_hierarchy=new_hierarchy,
                dependency_path=current_path,
                qualifiers=dependency.qualifiers,
                name_is_dependency_parameter=True,
                requester_pod=pod,
            )
        else:
            resolved_dependency = self.__resolve_collection_dependency(
                dependency=dependency,
                dependency_hierarchy=new_hierarchy,
                dependency_path=current_path,
            )
        if (
            resolved_dependency is None
            and not dependency.has_default
            and not dependency.is_optional
        ):
            raise UnexpectedDependencyTypeInjectedError(
                pod.type_,
                {
                    "name": name,
                    "expected": dependency.type_,
                    "actual": NoneType,
                },
                dependency_diagnostic=self.__dependency_diagnostic(
                    pod=pod,
                    dependency_parameter_name=name,
                    requested_type=requested_type,
                    dependency_path=current_path,
                ),
            )
        dependencies[name] = resolved_dependency
    started_at = perf_counter()
    try:
        instance: object = pod.instantiate(dependencies=dependencies)
    except BaseException:
        self.__record_instantiation_attempt(perf_counter() - started_at)
        raise
    self.__record_instantiation_attempt(perf_counter() - started_at)
    post_processed: object = self.__post_process_pod(instance)
    return post_processed

__get_pod_instance(pod, dependency_hierarchy, dependency_path)

Get or create an instance for an already resolved Pod candidate.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __get_pod_instance(
    self,
    pod: Pod,
    dependency_hierarchy: tuple[type, ...],
    dependency_path: tuple[PodDependencyPathNode, ...],
) -> object:
    """Get or create an instance for an already resolved Pod candidate."""
    if pod.scope is Pod.Scope.SINGLETON:
        if (cached := self.__get_singleton_cache(pod)) is not None:
            return cached
        with self.__singleton_lock:
            if (cached := self.__singleton_cache.get(pod.name)) is not None:
                return cached
            instance = self.__instantiate_pod(
                pod,
                dependency_hierarchy,
                dependency_path,
            )
            self.__set_singleton_cache(pod, instance)
            return instance
    if pod.scope is Pod.Scope.CONTEXT:
        if (cached := self.__get_context_cache(pod)) is not None:
            return cached
    instance = self.__instantiate_pod(
        pod,
        dependency_hierarchy,
        dependency_path,
    )

    if pod.scope is Pod.Scope.CONTEXT:
        self.__set_context_cache(pod, instance)

    return instance

__resolve_collection_dependency(dependency, dependency_hierarchy, dependency_path)

Resolve list/tuple/dict dependency injection without single ambiguity.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __resolve_collection_dependency(
    self,
    dependency: DependencyInfo,
    dependency_hierarchy: tuple[type, ...],
    dependency_path: tuple[PodDependencyPathNode, ...],
) -> object | None:
    """Resolve list/tuple/dict dependency injection without single ambiguity."""
    pods = self.__resolve_collection_candidates(
        type_=dependency.type_,
        qualifiers=dependency.qualifiers,
    )
    if not pods:
        return None

    instances = [
        self.__get_pod_instance(
            pod=pod,
            dependency_hierarchy=dependency_hierarchy,
            dependency_path=dependency_path,
        )
        for pod in pods
    ]
    if dependency.collection_kind is DependencyCollectionKind.LIST:
        return instances
    if dependency.collection_kind is DependencyCollectionKind.TUPLE:
        return tuple(instances)
    if dependency.collection_kind is DependencyCollectionKind.DICT:
        return {
            pod.name: instance
            for pod, instance in zip(pods, instances, strict=True)
        }
    return None

__post_process_pod(pod)

Apply all registered post-processors to a Pod instance.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The post-processed Pod instance.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __post_process_pod(self, pod: object) -> object:
    """Apply all registered post-processors to a Pod instance.

    Args:
        pod: The Pod instance to process.

    Returns:
        The post-processed Pod instance.
    """
    for post_processor in self.__post_processors:
        started_at = perf_counter()
        try:
            pod = post_processor.post_process(pod)
        except BaseException as e:
            self.__record_post_processing_attempt(perf_counter() - started_at, e)
            raise
        self.__record_post_processing_attempt(perf_counter() - started_at)
    return pod

__register_post_processors()

Register built-in and user-defined post-processors.

Registers post-processors in order: 1. ApplicationContextAwareProcessor 2. AspectPostProcessor 3. ServicePostProcessor 4. User-defined IPostProcessor Pods (sorted by @Order)

Source code in core/spakky/src/spakky/core/application/application_context.py

def __register_post_processors(self) -> None:
    """Register built-in and user-defined post-processors.

    Registers post-processors in order:
    1. ApplicationContextAwareProcessor
    2. AspectPostProcessor
    3. ServicePostProcessor
    4. User-defined IPostProcessor Pods (sorted by @Order)
    """
    self.__add_post_processor(ApplicationContextAwareProcessor(self))
    self.__add_post_processor(AspectPostProcessor(self))
    self.__add_post_processor(ServicePostProcessor(self))

    # Find and sort post-processors efficiently using list comprehension
    post_processors = sorted(
        cast(
            list[IPostProcessor],
            list(self.find(lambda x: IPostProcessor in x.base_types)),
        ),
        key=lambda x: Order.get_or_default(x, Order()).order,
    )
    for post_processor in post_processors:
        self.__add_post_processor(post_processor)

__initialize_pods()

Eagerly initialize all non-lazy Pods.

Raises:

Type	Description
NoSuchPodError	If a Pod cannot be instantiated.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __initialize_pods(self) -> None:
    """Eagerly initialize all non-lazy Pods.

    Raises:
        NoSuchPodError: If a Pod cannot be instantiated.
    """
    # Eagerly initialize non-lazy pods using list comprehension for efficiency
    non_lazy_pods = [
        pod for pod in self.__pods.values() if not Lazy.exists(pod.target)
    ]
    for pod in non_lazy_pods:
        if self.__startup_metrics is not None:
            self.__startup_metrics.instantiation_attempt_count += 1
        if (
            self.__get_internal(type_=pod.type_, name=pod.name) is None
        ):  # pragma: no cover - coverage boundary
            raise NoSuchPodError(pod.type_, pod.name)

__get_internal(type_, name, dependency_hierarchy=None, dependency_path=None, qualifiers=None, name_is_dependency_parameter=False, requester_pod=None)

Internal method to get or create a Pod instance.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type to resolve.	required
name	str | None	Optional name qualifier.	required
dependency_hierarchy	tuple[type, ...] | None	Immutable tuple for circular dependency detection.	None
qualifiers	list[Qualifier] | None	List of qualifier annotations.	None

Returns:

Type	Description
T | None	The resolved Pod instance or None if not found.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __get_internal[T: object](
    self,
    type_: type[T],
    name: str | None,
    dependency_hierarchy: tuple[type, ...] | None = None,
    dependency_path: tuple[PodDependencyPathNode, ...] | None = None,
    qualifiers: list[Qualifier] | None = None,
    name_is_dependency_parameter: bool = False,
    requester_pod: Pod | None = None,
) -> T | None:
    """Internal method to get or create a Pod instance.

    Args:
        type_: The type to resolve.
        name: Optional name qualifier.
        dependency_hierarchy: Immutable tuple for circular dependency detection.
        qualifiers: List of qualifier annotations.

    Returns:
        The resolved Pod instance or None if not found.
    """
    if dependency_hierarchy is None:
        # If dependency_hierarchy is None
        # it means that this is the first call on recursive cycle
        dependency_hierarchy = ()
    if dependency_path is None:
        dependency_path = ()
    if qualifiers is None:
        # If qualifiers is None, it means that no qualifier is specified
        qualifiers = []
    if isinstance(
        type_, str
    ):  # To support forward references  # pragma: no cover - coverage boundary
        if (
            type_ not in self.__forward_type_map
        ):  # pragma: no cover - coverage boundary
            return None
        type_ = self.__forward_type_map[
            type_
        ]  # pragma: no cover - coverage boundary

    pod = self.__resolve_candidate(
        type_=type_,
        name=name,
        qualifiers=qualifiers,
        name_is_dependency_parameter=name_is_dependency_parameter,
        requester_pod=requester_pod,
        dependency_path=dependency_path,
    )
    if pod is None:
        return None

    return cast(
        T,
        self.__get_pod_instance(
            pod=pod,
            dependency_hierarchy=dependency_hierarchy,
            dependency_path=dependency_path,
        ),
    )

__start_services(started_services, started_async_services)

Start all registered sync and async services.

Raises:

Type	Description
EventLoopThreadAlreadyStartedInApplicationContextError	If already started.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __start_services(
    self,
    started_services: list[IService],
    started_async_services: list[IAsyncService],
) -> None:
    """Start all registered sync and async services.

    Raises:
        EventLoopThreadAlreadyStartedInApplicationContextError: If already started.
    """
    if self.__event_loop is not None:  # pragma: no cover - coverage boundary
        raise EventLoopThreadAlreadyStartedInApplicationContextError
    if self.__event_thread is not None:  # pragma: no cover - coverage boundary
        raise EventLoopThreadAlreadyStartedInApplicationContextError

    self.__event_loop = new_event_loop()
    self.__event_thread = Thread(
        target=self.__run_event_loop,
        args=(self.__event_loop,),
        daemon=True,
    )
    self.__event_thread.start()

    for service in self.__services:
        service.start()
        started_services.append(service)

    async def start_async_services() -> None:
        if self.__event_loop is None:  # pragma: no cover - coverage boundary
            raise EventLoopThreadNotStartedInApplicationContextError
        for service in self.__async_services:
            await service.start_async()
            started_async_services.append(service)

    run_coroutine_threadsafe(start_async_services(), self.__event_loop).result()

__stop_services()

Stop all services and shutdown event loop.

Raises:

Type	Description
EventLoopThreadNotStartedInApplicationContextError	If not started.

Source code in core/spakky/src/spakky/core/application/application_context.py

def __stop_services(self) -> None:
    """Stop all services and shutdown event loop.

    Raises:
        EventLoopThreadNotStartedInApplicationContextError: If not started.
    """
    if self.__event_loop is None:  # pragma: no cover - coverage boundary
        raise EventLoopThreadNotStartedInApplicationContextError
    if self.__event_thread is None:  # pragma: no cover - coverage boundary
        raise EventLoopThreadNotStartedInApplicationContextError

    # Store references to avoid race condition with concurrent stop() calls
    event_loop = self.__event_loop
    event_thread = self.__event_thread

    for service in self.__services:
        service.stop()

    async def stop_async_services() -> None:
        for service in self.__async_services:
            await service.stop_async()

    run_coroutine_threadsafe(stop_async_services(), event_loop).result()
    event_loop.call_soon_threadsafe(event_loop.stop)  # type: ignore[arg-type]  # stop() is valid callback
    event_thread.join()

    # Clear references after thread has joined
    self.__event_loop = None
    self.__event_thread = None

find(selector)

Find all Pod instances matching selector predicate.

Parameters:

Name	Type	Description	Default
selector	Callable[[Pod], bool]	Predicate function to filter Pods.	required

Returns:

Type	Description
set[object]	Set of matching Pod instances.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def find(self, selector: Callable[[Pod], bool]) -> set[object]:
    """Find all Pod instances matching selector predicate.

    Args:
        selector: Predicate function to filter Pods.

    Returns:
        Set of matching Pod instances.
    """
    # Use set comprehension for optimal filtering and instantiation
    return {
        self.__get_internal(type_=pod.type_, name=pod.name)
        for pod in self.__pods.values()
        if selector(pod)
    }

add(obj)

Register a Pod-annotated class or function.

Parameters:

Name	Type	Description	Default
obj	PodType	The Pod to register.	required

Raises:

Type	Description
CannotRegisterNonPodObjectError	If obj is not annotated with @Pod.
PodNameAlreadyExistsError	If Pod name already registered with different ID.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def add(self, obj: PodType) -> None:
    """Register a Pod-annotated class or function.

    Args:
        obj: The Pod to register.

    Raises:
        CannotRegisterNonPodObjectError: If obj is not annotated with @Pod.
        PodNameAlreadyExistsError: If Pod name already registered with different ID.
    """
    if not Pod.exists(obj):  # pragma: no cover - coverage boundary
        raise CannotRegisterNonPodObjectError(obj)
    pod: Pod = Pod.get(obj)
    if pod.name in self.__pods:
        # 같은 ID의 Pod 재등록은 add() 호출 패턴상 발생하지 않음
        if (
            self.__pods[pod.name].id == pod.id
        ):  # pragma: no cover - coverage boundary
            return
        raise PodNameAlreadyExistsError(pod.name)
    for base_type in pod.base_types:
        self.__forward_type_map[base_type.__name__] = base_type
    self.__pods[pod.name] = pod

    # Update type index for fast lookup.
    self.__index_type_cache(pod.type_, pod)
    pod_type_origin = get_origin(pod.type_)
    if pod_type_origin is not None:
        self.__index_type_cache(pod_type_origin, pod)

    # Also index by all base types for polymorphic lookups
    for base_type in pod.base_types:
        if base_type not in self.__type_cache:
            self.__type_cache[base_type] = set()
        self.__type_cache[base_type].add(pod)
        base_type_origin = get_origin(base_type)
        if base_type_origin is not None:
            self.__index_type_cache(base_type_origin, pod)

bind(binding)

Register an explicit interface-to-implementation binding policy.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def bind(self, binding: PodBinding) -> None:
    """Register an explicit interface-to-implementation binding policy."""
    self.__validate_binding(binding)
    self.__bindings[binding.interface] = binding

bind_to_type(interface, implementation)

Bind an interface to a concrete implementation type.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def bind_to_type(self, interface: type, implementation: type) -> None:
    """Bind an interface to a concrete implementation type."""
    self.bind(PodBinding(interface=interface, implementation_type=implementation))

bind_to_name(interface, name)

Bind an interface to a registered Pod name.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def bind_to_name(self, interface: type, name: str) -> None:
    """Bind an interface to a registered Pod name."""
    self.bind(PodBinding(interface=interface, implementation_name=name))

add_service(service)

Register a service for lifecycle management.

Parameters:

Name	Type	Description	Default
service	IService | IAsyncService	The service to register (sync or async).	required

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def add_service(self, service: IService | IAsyncService) -> None:
    """Register a service for lifecycle management.

    Args:
        service: The service to register (sync or async).
    """
    if isinstance(service, IService):
        self.__services.append(service)
    if isinstance(service, IAsyncService):
        self.__async_services.append(service)

start(startup_phase_recorder=None)

Start the application context.

Registers post-processors, initializes Pods, and starts services.

Raises:

Type	Description
ApplicationContextAlreadyStartedError	If already started.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def start(
    self,
    startup_phase_recorder: IStartupPhaseRecorder | None = None,
) -> None:
    """Start the application context.

    Registers post-processors, initializes Pods, and starts services.

    Raises:
        ApplicationContextAlreadyStartedError: If already started.
    """
    if self.__is_started:  # pragma: no cover - coverage boundary
        raise ApplicationContextAlreadyStartedError()
    recorder = (
        startup_phase_recorder
        if startup_phase_recorder is not None
        else NoOpStartupPhaseRecorder()
    )
    original_service_count = len(self.__services)
    original_async_service_count = len(self.__async_services)
    started_services: list[IService] = []
    started_async_services: list[IAsyncService] = []
    self.__is_started = True
    try:
        with recorder.record_phase(
            phase_name=STARTUP_PHASE_POST_PROCESSOR_REGISTRATION
        ) as phase:
            self.__register_post_processors()
            phase.set_processed_count(len(self.__post_processors))

        startup_metrics = _ApplicationContextStartupMetrics()
        self.__startup_metrics = startup_metrics
        try:
            self.__initialize_pods()
        except BaseException as e:
            if startup_metrics.post_processing_exception is None:
                recorder.record_failure(
                    phase_name=STARTUP_PHASE_INSTANTIATION,
                    elapsed_seconds=startup_metrics.instantiation_elapsed_seconds,
                    exception=e,
                    processed_count=startup_metrics.instantiation_attempt_count,
                    diagnostic_details=self.__startup_diagnostic_details(e),
                )
            else:
                recorder.record_success(
                    phase_name=STARTUP_PHASE_INSTANTIATION,
                    elapsed_seconds=startup_metrics.instantiation_elapsed_seconds,
                    processed_count=startup_metrics.instantiation_attempt_count,
                )
                recorder.record_failure(
                    phase_name=STARTUP_PHASE_POST_PROCESSING,
                    elapsed_seconds=startup_metrics.post_processing_elapsed_seconds,
                    exception=startup_metrics.post_processing_exception,
                    processed_count=startup_metrics.post_processing_application_count,
                    diagnostic_details=self.__startup_diagnostic_details(
                        startup_metrics.post_processing_exception
                    ),
                )
            raise

        recorder.record_success(
            phase_name=STARTUP_PHASE_INSTANTIATION,
            elapsed_seconds=startup_metrics.instantiation_elapsed_seconds,
            processed_count=startup_metrics.instantiation_attempt_count,
        )
        recorder.record_success(
            phase_name=STARTUP_PHASE_POST_PROCESSING,
            elapsed_seconds=startup_metrics.post_processing_elapsed_seconds,
            processed_count=startup_metrics.post_processing_application_count,
        )

        service_start_count = len(self.__services) + len(self.__async_services)
        with recorder.record_phase(
            phase_name=STARTUP_PHASE_SERVICE_START,
            processed_count=service_start_count,
        ):
            self.__start_services(started_services, started_async_services)
    except BaseException:
        self.__rollback_failed_start(
            original_service_count=original_service_count,
            original_async_service_count=original_async_service_count,
            started_services=started_services,
            started_async_services=started_async_services,
        )
        raise
    finally:
        self.__startup_metrics = None

stop()

Stop the application context and clean up resources.

Thread-safe: Multiple concurrent calls to stop() are serialized.

Raises:

Type	Description
ApplicationContextAlreadyStoppedError	If already stopped.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def stop(self) -> None:
    """Stop the application context and clean up resources.

    Thread-safe: Multiple concurrent calls to stop() are serialized.

    Raises:
        ApplicationContextAlreadyStoppedError: If already stopped.
    """
    with self.__shutdown_lock:
        if not self.__is_started:  # pragma: no cover - coverage boundary
            raise ApplicationContextAlreadyStoppedError()
        self.__stop_services()
        self.__clear_all()
        self.__is_started = False

get(type_, name=None)

get(type_: type[T]) -> T

get(type_: type[T], name: str) -> T

Get a Pod instance by type and optional name.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type to retrieve.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
T | object	The Pod instance.

Raises:

Type	Description
NoSuchPodError	If no matching Pod found.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def get[T: object](
    self,
    type_: type[T],
    name: str | None = None,
) -> T | object:
    """Get a Pod instance by type and optional name.

    Args:
        type_: The type to retrieve.
        name: Optional name qualifier.

    Returns:
        The Pod instance.

    Raises:
        NoSuchPodError: If no matching Pod found.
    """
    instance = self.__get_internal(type_=type_, name=name)
    if instance is None:  # pragma: no cover - coverage boundary
        raise NoSuchPodError(type_, name)
    return instance

get_or_none(type_, name=None)

get_or_none(type_: type[T]) -> T | None

get_or_none(type_: type[T], name: str) -> T | None

Get a Pod instance by type and optional name, or None if not found.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type to retrieve.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
T | None	The Pod instance, or None if no matching Pod found.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def get_or_none[T: object](
    self,
    type_: type[T],
    name: str | None = None,
) -> T | None:
    """Get a Pod instance by type and optional name, or None if not found.

    Args:
        type_: The type to retrieve.
        name: Optional name qualifier.

    Returns:
        The Pod instance, or None if no matching Pod found.
    """
    return self.__get_internal(type_=type_, name=name)

contains(type_, name=None)

contains(type_: type) -> bool

contains(type_: type, name: str) -> bool

Check if a Pod is registered.

Parameters:

Name	Type	Description	Default
type_	type	The type to check.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
bool	True if matching Pod exists.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def contains(self, type_: type, name: str | None = None) -> bool:
    """Check if a Pod is registered.

    Args:
        type_: The type to check.
        name: Optional name qualifier.

    Returns:
        True if matching Pod exists.
    """
    if name is not None:
        return name in self.__pods
    # Use type index for O(1) lookup
    return type_ in self.__type_cache and len(self.__type_cache[type_]) > 0

register_tag(tag)

Register a Tag instance.

Parameters:

Name	Type	Description	Default
tag	Tag	The Tag to register.	required

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def register_tag(self, tag: Tag) -> None:
    """Register a Tag instance.

    Args:
        tag: The Tag to register.
    """
    self.__tags.add(tag)

contains_tag(tag)

Check if a Tag is registered.

Parameters:

Name	Type	Description	Default
tag	Tag	The Tag to check.	required

Returns:

Type	Description
bool	True if Tag is registered.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def contains_tag(self, tag: Tag) -> bool:
    """Check if a Tag is registered.

    Args:
        tag: The Tag to check.

    Returns:
        True if Tag is registered.
    """
    return tag in self.__tags

list_tags(selector=None)

List registered Tags, optionally filtered by selector.

Parameters:

Name	Type	Description	Default
selector	Callable[[Tag], bool] | None	Optional predicate to filter Tags.	None

Returns: Set of matching Tags.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def list_tags(
    self, selector: Callable[[Tag], bool] | None = None
) -> frozenset[Tag]:
    """List registered Tags, optionally filtered by selector.

    Args:
        selector: Optional predicate to filter Tags.
    Returns:
        Set of matching Tags.
    """
    if selector is None:
        return frozenset(self.__tags)
    return frozenset(tag for tag in self.__tags if selector(tag))

get_context_id()

Get or create unique ID for current context.

Returns:

Type	Description
UUID	UUID for this context.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def get_context_id(self) -> UUID:
    """Get or create unique ID for current context.

    Returns:
        UUID for this context.
    """
    context = self.__context_cache.get({})
    if CONTEXT_ID not in context:  # pragma: no cover - coverage boundary
        context[CONTEXT_ID] = uuid4()
        self.__context_cache.set(context)
    return cast(UUID, context[CONTEXT_ID])

get_context_value(key)

Get a value from the context-scoped cache.

Parameters:

Name	Type	Description	Default
key	str	The key to retrieve.	required

Returns:

Type	Description
object | None	The cached value, or None if not found.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def get_context_value(self, key: str) -> object | None:
    """Get a value from the context-scoped cache.

    Args:
        key: The key to retrieve.

    Returns:
        The cached value, or None if not found.
    """
    if key == CONTEXT_ID:
        return self.get_context_id()
    context = self.__context_cache.get({})
    return context.get(key)

set_context_value(key, value)

Set a value in the context-scoped cache.

Parameters:

Name	Type	Description	Default
key	str	The key to set.	required
value	object	The value to store.	required

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def set_context_value(self, key: str, value: object) -> None:
    """Set a value in the context-scoped cache.

    Args:
        key: The key to set.
        value: The value to store.
    """
    if key == CONTEXT_ID:  # pragma: no cover - coverage boundary
        raise CannotAssignSystemContextIDError
    context = self.__context_cache.get({})
    context[key] = value
    self.__context_cache.set(context)

clear_context()

Clear context-scoped cache for current context.

Source code in core/spakky/src/spakky/core/application/application_context.py

@override
def clear_context(self) -> None:
    """Clear context-scoped cache for current context."""
    self.__context_cache.set({})

options: show_root_heading: false

spakky.core.application.plugin

Plugin metadata representation.

This module defines the Plugin class for identifying and managing framework plugins.

Plugin

Bases: IEquatable

Immutable plugin identifier.

Plugins are identified by name and used to selectively load framework extensions.

name instance-attribute

Unique name of the plugin.

__hash__()

Compute hash based on plugin name.

Returns:

Type	Description
int	Hash value for this plugin.

Source code in core/spakky/src/spakky/core/application/plugin.py

def __hash__(self) -> int:
    """Compute hash based on plugin name.

    Returns:
        Hash value for this plugin.
    """
    return hash(self.name)

__eq__(__value)

Check equality based on plugin name.

Parameters:

Name	Type	Description	Default
__value	object	The object to compare with.	required

Returns:

Type	Description
bool	True if both plugins have the same name.

Source code in core/spakky/src/spakky/core/application/plugin.py

def __eq__(self, __value: object) -> bool:
    """Check equality based on plugin name.

    Args:
        __value: The object to compare with.

    Returns:
        True if both plugins have the same name.
    """
    if not isinstance(__value, Plugin):
        return False
    return self.name == __value.name

options: show_root_heading: false

spakky.core.application.discovery_manifest

Discovery manifest storage for reusable scan discovery results.

DiscoveryManifestDecision

Bases: StrEnum

Decision made while loading a discovery manifest.

DiscoveryManifestSourceFingerprint

Fingerprint for one Python source file involved in scan discovery.

path instance-attribute

Absolute source file path.

mtime_ns instance-attribute

Source file modification time in nanoseconds.

size instance-attribute

Source file size in bytes.

to_json()

Serialize the source fingerprint.

Returns:

Type	Description
JsonObject	JSON-compatible object.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def to_json(self) -> JsonObject:
    """Serialize the source fingerprint.

    Returns:
        JSON-compatible object.
    """
    return {
        "path": self.path,
        "mtime_ns": self.mtime_ns,
        "size": self.size,
    }

DiscoveryManifestFingerprint

Input fingerprint used to decide whether a manifest is fresh.

schema_version instance-attribute

Manifest schema version used by the current runtime.

python_version instance-attribute

Major/minor Python version.

module_name instance-attribute

Scan target module name.

is_package instance-attribute

Whether the target is a package scan.

exclude instance-attribute

Normalized exclude module patterns.

sources instance-attribute

Source files that affect scan discovery.

from_scan_input(path, exclude) classmethod

Build a fingerprint from scan inputs.

Parameters:

Name	Type	Description	Default
path	Module	Scan target module or module name.	required
exclude	set[Module]	Exclude module patterns.	required

Returns:

Type	Description
Self	Fingerprint for the current scan input.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

@classmethod
def from_scan_input(
    cls,
    path: Module,
    exclude: set[Module],
) -> Self:
    """Build a fingerprint from scan inputs.

    Args:
        path: Scan target module or module name.
        exclude: Exclude module patterns.

    Returns:
        Fingerprint for the current scan input.
    """
    module = resolve_module(path)
    source_files = _source_files_for(module)
    return cls(
        schema_version=DISCOVERY_MANIFEST_SCHEMA_VERSION,
        python_version=f"{sys.version_info.major}.{sys.version_info.minor}",
        module_name=module.__name__,
        is_package=is_package(module),
        exclude=tuple(sorted(_normalize_module_name(item) for item in exclude)),
        sources=tuple(_source_fingerprint(source) for source in source_files),
    )

to_json()

Serialize the fingerprint.

Returns:

Type	Description
JsonObject	JSON-compatible object.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def to_json(self) -> JsonObject:
    """Serialize the fingerprint.

    Returns:
        JSON-compatible object.
    """
    return {
        "schema_version": self.schema_version,
        "python_version": self.python_version,
        "module_name": self.module_name,
        "is_package": self.is_package,
        "exclude": list(self.exclude),
        "sources": [source.to_json() for source in self.sources],
    }

DiscoveryManifestCandidate

Pod or Tag candidate discovered during scan.

module_name instance-attribute

Module containing the candidate object.

qualname instance-attribute

Qualified object name inside the module.

from_object(obj) classmethod

Build a candidate identity from a discovered object.

Parameters:

Name	Type	Description	Default
obj	type | FunctionType	Discovered class or function.	required

Returns:

Type	Description
Self	Candidate identity.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

@classmethod
def from_object(cls, obj: type | FunctionType) -> Self:
    """Build a candidate identity from a discovered object.

    Args:
        obj: Discovered class or function.

    Returns:
        Candidate identity.
    """
    return cls(module_name=obj.__module__, qualname=obj.__qualname__)

matches(obj)

Check whether an object matches this candidate identity.

Parameters:

Name	Type	Description	Default
obj	object	Candidate object loaded from a module.	required

Returns:

Type	Description
bool	True when the object has the same module and qualified name.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def matches(self, obj: object) -> bool:
    """Check whether an object matches this candidate identity.

    Args:
        obj: Candidate object loaded from a module.

    Returns:
        True when the object has the same module and qualified name.
    """
    if isinstance(obj, FunctionType):
        return (
            obj.__module__ == self.module_name and obj.__qualname__ == self.qualname
        )
    if isinstance(obj, type):
        return (
            obj.__module__ == self.module_name and obj.__qualname__ == self.qualname
        )
    return False

to_json()

Serialize the candidate.

Returns:

Type	Description
JsonObject	JSON-compatible object.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def to_json(self) -> JsonObject:
    """Serialize the candidate.

    Returns:
        JSON-compatible object.
    """
    return {
        "module_name": self.module_name,
        "qualname": self.qualname,
    }

DiscoveryManifest

Reusable scan discovery manifest.

fingerprint instance-attribute

Input fingerprint that produced this manifest.

module_names instance-attribute

Modules scanned during fresh discovery.

candidates instance-attribute

Pod and Tag candidates discovered during fresh discovery.

to_json()

Serialize the manifest.

Returns:

Type	Description
JsonObject	JSON-compatible object.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def to_json(self) -> JsonObject:
    """Serialize the manifest.

    Returns:
        JSON-compatible object.
    """
    return {
        "schema_version": DISCOVERY_MANIFEST_SCHEMA_VERSION,
        "fingerprint": self.fingerprint.to_json(),
        "module_names": list(self.module_names),
        "candidates": [candidate.to_json() for candidate in self.candidates],
    }

DiscoveryManifestLoadResult

Result of attempting to load a discovery manifest.

decision instance-attribute

Manifest reuse decision.

manifest = None class-attribute instance-attribute

Loaded manifest when the decision is hit.

DiscoveryManifestStore(path)

File-backed DiscoveryManifest store.

Initialize the store.

Parameters:

Name	Type	Description	Default
path	Path	Manifest JSON file path.	required

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def __init__(self, path: Path) -> None:
    """Initialize the store.

    Args:
        path: Manifest JSON file path.
    """
    self._path = path

load(fingerprint)

Load a manifest and compare it to the current fingerprint.

Parameters:

Name	Type	Description	Default
fingerprint	DiscoveryManifestFingerprint	Current scan input fingerprint.	required

Returns:

Type	Description
DiscoveryManifestLoadResult	Manifest load decision and loaded manifest on hit.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def load(
    self,
    fingerprint: DiscoveryManifestFingerprint,
) -> DiscoveryManifestLoadResult:
    """Load a manifest and compare it to the current fingerprint.

    Args:
        fingerprint: Current scan input fingerprint.

    Returns:
        Manifest load decision and loaded manifest on hit.
    """
    if not self._path.exists():
        return DiscoveryManifestLoadResult(
            decision=DiscoveryManifestDecision.MISS,
        )

    try:
        payload: object = json.loads(self._path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return DiscoveryManifestLoadResult(
            decision=DiscoveryManifestDecision.MISS,
        )

    manifest = _manifest_from_json(payload)
    if manifest is None:
        return DiscoveryManifestLoadResult(
            decision=DiscoveryManifestDecision.MISS,
        )
    if manifest.fingerprint.schema_version != DISCOVERY_MANIFEST_SCHEMA_VERSION:
        return DiscoveryManifestLoadResult(
            decision=DiscoveryManifestDecision.STALE_SCHEMA,
        )
    if manifest.fingerprint.to_json() != fingerprint.to_json():
        return DiscoveryManifestLoadResult(
            decision=DiscoveryManifestDecision.STALE_INPUT,
        )
    return DiscoveryManifestLoadResult(
        decision=DiscoveryManifestDecision.HIT,
        manifest=manifest,
    )

save(manifest)

Persist a manifest.

Parameters:

Name	Type	Description	Default
manifest	DiscoveryManifest	Manifest generated from fresh discovery.	required

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def save(self, manifest: DiscoveryManifest) -> None:
    """Persist a manifest.

    Args:
        manifest: Manifest generated from fresh discovery.
    """
    self._path.parent.mkdir(parents=True, exist_ok=True)
    self._path.write_text(
        json.dumps(manifest.to_json(), indent=2, sort_keys=True),
        encoding="utf-8",
    )

default_discovery_manifest_path()

Return the deterministic project-local manifest path.

Returns:

Type	Description
Path	Default DiscoveryManifest path under the project cache directory.

Source code in core/spakky/src/spakky/core/application/discovery_manifest.py

def default_discovery_manifest_path() -> Path:
    """Return the deterministic project-local manifest path.

    Returns:
        Default DiscoveryManifest path under the project cache directory.
    """
    return Path.cwd() / DEFAULT_DISCOVERY_MANIFEST_PATH

options: show_root_heading: false

spakky.core.application.startup_diagnostics

Startup diagnostics report and phase recorder contracts.

StartupElapsedTimeCannotBeNegativeError

Bases: AbstractSpakkyApplicationError

Raised when a startup phase elapsed time is negative.

StartupProcessedCountCannotBeNegativeError

Bases: AbstractSpakkyApplicationError

Raised when a startup phase processed count is negative.

StartupFailureSummaryRequiredError

Bases: AbstractSpakkyApplicationError

Raised when a failure record has no failure summary.

StartupFailureSummaryNotAllowedError

Bases: AbstractSpakkyApplicationError

Raised when a success record has a failure summary.

StartupPhaseStatus

Bases: StrEnum

Result status for a startup phase record.

StartupDiagnosticDetail

Structured diagnostic detail attached to a startup failure.

key instance-attribute

Machine-readable diagnostic detail key.

value instance-attribute

Diagnostic detail value.

StartupFailureSummary

Structured startup failure summary without the raw exception object.

exception_type_name instance-attribute

Concrete exception type name.

message instance-attribute

Exception message captured at failure time.

diagnostic_details = () class-attribute instance-attribute

Optional structured diagnostic details.

from_exception(exception, diagnostic_details=()) classmethod

Create a failure summary from an exception.

Parameters:

Name	Type	Description	Default
exception	BaseException	Exception raised by a startup phase.	required
diagnostic_details	StartupDiagnosticDetails	Optional structured diagnostic details.	()

Returns:

Type	Description
Self	Failure summary containing exception type and message only.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@classmethod
def from_exception(
    cls,
    exception: BaseException,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> Self:
    """Create a failure summary from an exception.

    Args:
        exception: Exception raised by a startup phase.
        diagnostic_details: Optional structured diagnostic details.

    Returns:
        Failure summary containing exception type and message only.
    """
    resolved_diagnostic_details = diagnostic_details
    if isinstance(exception, IStartupDiagnosticDetailProvider):
        resolved_diagnostic_details = (
            *diagnostic_details,
            *exception.startup_diagnostic_details,
        )
    return cls(
        exception_type_name=type(exception).__name__,
        message=str(exception),
        diagnostic_details=resolved_diagnostic_details,
    )

StartupPhaseRecord

Single startup phase timing, count, status, and failure record.

phase_name instance-attribute

Name of the startup phase.

elapsed_seconds instance-attribute

Elapsed wall-clock seconds for the phase.

processed_count instance-attribute

Number of domain objects processed by the phase.

status instance-attribute

Success or failure status for the phase.

failure_summary = None class-attribute instance-attribute

Failure summary when status is failure; absent for success records.

diagnostic_details = () class-attribute instance-attribute

Structured diagnostic details attached to the phase record.

__post_init__()

Validate phase record numeric invariants.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __post_init__(self) -> None:
    """Validate phase record numeric invariants."""
    if self.elapsed_seconds < 0:
        raise StartupElapsedTimeCannotBeNegativeError()
    if self.processed_count < 0:
        raise StartupProcessedCountCannotBeNegativeError()
    if (
        self.status is StartupPhaseStatus.SUCCESS
        and self.failure_summary is not None
    ):
        raise StartupFailureSummaryNotAllowedError()
    if self.status is StartupPhaseStatus.FAILURE and self.failure_summary is None:
        raise StartupFailureSummaryRequiredError()

StartupReport()

Record collection for one application startup attempt.

Initialize an empty startup report.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __init__(self) -> None:
    """Initialize an empty startup report."""
    self._records = []

records property

Get startup phase records in insertion order.

Returns:

Type	Description
tuple[StartupPhaseRecord, ...]	Immutable snapshot of recorded startup phases.

add_record(record)

Append a phase record.

Parameters:

Name	Type	Description	Default
record	StartupPhaseRecord	Phase record to append.	required

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def add_record(self, record: StartupPhaseRecord) -> None:
    """Append a phase record.

    Args:
        record: Phase record to append.
    """
    self._records.append(record)

IStartupPhaseRecorder

Bases: ABC

Interface for startup phase recorders.

report abstractmethod property

Get the startup report backing this recorder.

record_success(phase_name, elapsed_seconds, processed_count=0, diagnostic_details=()) abstractmethod

Record a successful startup phase.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@abstractmethod
def record_success(
    self,
    phase_name: StartupPhaseName,
    elapsed_seconds: StartupElapsedSeconds,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecord:
    """Record a successful startup phase."""
    ...

record_failure(phase_name, elapsed_seconds, exception, processed_count=0, diagnostic_details=()) abstractmethod

Record a failed startup phase.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@abstractmethod
def record_failure(
    self,
    phase_name: StartupPhaseName,
    elapsed_seconds: StartupElapsedSeconds,
    exception: BaseException,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecord:
    """Record a failed startup phase."""
    ...

record_phase(phase_name, processed_count=0, diagnostic_details=())

Measure a startup phase and record success or failure.

Parameters:

Name	Type	Description	Default
phase_name	StartupPhaseName	Name of the startup phase.	required
processed_count	StartupProcessedCount	Number of domain objects processed by the phase.	0
diagnostic_details	StartupDiagnosticDetails	Optional structured diagnostic details.	()

Returns:

Type	Description
StartupPhaseRecording	Context manager that records the phase and re-raises failures.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def record_phase(
    self,
    phase_name: StartupPhaseName,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecording:
    """Measure a startup phase and record success or failure.

    Args:
        phase_name: Name of the startup phase.
        processed_count: Number of domain objects processed by the phase.
        diagnostic_details: Optional structured diagnostic details.

    Returns:
        Context manager that records the phase and re-raises failures.
    """
    return StartupPhaseRecording(
        recorder=self,
        phase_name=phase_name,
        processed_count=processed_count,
        diagnostic_details=diagnostic_details,
    )

IStartupDiagnosticDetailProvider

Bases: ABC

Interface for startup exceptions that expose structured diagnostics.

startup_diagnostic_details abstractmethod property

Structured diagnostic details to attach to startup failure summaries.

StartupPhaseRecording(recorder, phase_name, processed_count, diagnostic_details)

Context manager for measuring and recording one startup phase.

Initialize a startup phase recording context.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __init__(
    self,
    recorder: IStartupPhaseRecorder,
    phase_name: StartupPhaseName,
    processed_count: StartupProcessedCount,
    diagnostic_details: StartupDiagnosticDetails,
) -> None:
    """Initialize a startup phase recording context."""
    if processed_count < 0:
        raise StartupProcessedCountCannotBeNegativeError()
    self._recorder = recorder
    self._phase_name = phase_name
    self._processed_count = processed_count
    self._diagnostic_details = diagnostic_details
    self._started_at = perf_counter()

__enter__()

Enter the startup phase recording context.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __enter__(self) -> Self:
    """Enter the startup phase recording context."""
    return self

__exit__(_exception_type, exception, _traceback)

Record the phase outcome and preserve exception propagation.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __exit__(
    self,
    _exception_type: type[BaseException] | None,
    exception: BaseException | None,
    _traceback: TracebackType | None,
) -> bool:
    """Record the phase outcome and preserve exception propagation."""
    elapsed_seconds = perf_counter() - self._started_at
    if exception is None:
        self._recorder.record_success(
            phase_name=self._phase_name,
            elapsed_seconds=elapsed_seconds,
            processed_count=self._processed_count,
            diagnostic_details=self._diagnostic_details,
        )
    else:
        self._recorder.record_failure(
            phase_name=self._phase_name,
            elapsed_seconds=elapsed_seconds,
            exception=exception,
            processed_count=self._processed_count,
            diagnostic_details=self._diagnostic_details,
        )
    return False

set_processed_count(processed_count)

Set the processed count before the phase exits.

Parameters:

Name	Type	Description	Default
processed_count	StartupProcessedCount	Final processed count for the measured phase.	required

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def set_processed_count(self, processed_count: StartupProcessedCount) -> None:
    """Set the processed count before the phase exits.

    Args:
        processed_count: Final processed count for the measured phase.
    """
    if processed_count < 0:
        raise StartupProcessedCountCannotBeNegativeError()
    self._processed_count = processed_count

set_diagnostic_details(diagnostic_details)

Set structured diagnostic details before the phase exits.

Parameters:

Name	Type	Description	Default
diagnostic_details	StartupDiagnosticDetails	Structured diagnostic details to attach.	required

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def set_diagnostic_details(
    self,
    diagnostic_details: StartupDiagnosticDetails,
) -> None:
    """Set structured diagnostic details before the phase exits.

    Args:
        diagnostic_details: Structured diagnostic details to attach.
    """
    self._diagnostic_details = diagnostic_details

ActiveStartupPhaseRecorder(report=None)

Bases: IStartupPhaseRecorder

Recorder that appends phase records to a startup report.

Initialize an active phase recorder.

Parameters:

Name	Type	Description	Default
report	StartupReport | None	Existing report to append to. None means create a new report.	None

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __init__(self, report: StartupReport | None = None) -> None:
    """Initialize an active phase recorder.

    Args:
        report: Existing report to append to. None means create a new report.
    """
    # None means the caller did not provide an existing startup attempt report.
    self._report = report if report is not None else StartupReport()

report property

Get the startup report backing this recorder.

record_success(phase_name, elapsed_seconds, processed_count=0, diagnostic_details=())

Record a successful startup phase.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@override
def record_success(
    self,
    phase_name: StartupPhaseName,
    elapsed_seconds: StartupElapsedSeconds,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecord:
    """Record a successful startup phase."""
    record = StartupPhaseRecord(
        phase_name=phase_name,
        elapsed_seconds=elapsed_seconds,
        processed_count=processed_count,
        status=StartupPhaseStatus.SUCCESS,
        diagnostic_details=diagnostic_details,
    )
    self._report.add_record(record)
    return record

record_failure(phase_name, elapsed_seconds, exception, processed_count=0, diagnostic_details=())

Record a failed startup phase.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@override
def record_failure(
    self,
    phase_name: StartupPhaseName,
    elapsed_seconds: StartupElapsedSeconds,
    exception: BaseException,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecord:
    """Record a failed startup phase."""
    record = StartupPhaseRecord(
        phase_name=phase_name,
        elapsed_seconds=elapsed_seconds,
        processed_count=processed_count,
        status=StartupPhaseStatus.FAILURE,
        failure_summary=StartupFailureSummary.from_exception(
            exception,
            diagnostic_details,
        ),
        diagnostic_details=diagnostic_details,
    )
    self._report.add_record(record)
    return record

NoOpStartupPhaseRecorder()

Bases: IStartupPhaseRecorder

Recorder that preserves startup behavior without report side effects.

Initialize a no-op recorder with an empty report.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

def __init__(self) -> None:
    """Initialize a no-op recorder with an empty report."""
    self._report = StartupReport()

report property

Get the empty report backing this recorder.

record_success(phase_name, elapsed_seconds, processed_count=0, diagnostic_details=())

Create a success record without mutating the report.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@override
def record_success(
    self,
    phase_name: StartupPhaseName,
    elapsed_seconds: StartupElapsedSeconds,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecord:
    """Create a success record without mutating the report."""
    return StartupPhaseRecord(
        phase_name=phase_name,
        elapsed_seconds=elapsed_seconds,
        processed_count=processed_count,
        status=StartupPhaseStatus.SUCCESS,
        diagnostic_details=diagnostic_details,
    )

record_failure(phase_name, elapsed_seconds, exception, processed_count=0, diagnostic_details=())

Create a failure record without mutating the report.

Source code in core/spakky/src/spakky/core/application/startup_diagnostics.py

@override
def record_failure(
    self,
    phase_name: StartupPhaseName,
    elapsed_seconds: StartupElapsedSeconds,
    exception: BaseException,
    processed_count: StartupProcessedCount = 0,
    diagnostic_details: StartupDiagnosticDetails = (),
) -> StartupPhaseRecord:
    """Create a failure record without mutating the report."""
    return StartupPhaseRecord(
        phase_name=phase_name,
        elapsed_seconds=elapsed_seconds,
        processed_count=processed_count,
        status=StartupPhaseStatus.FAILURE,
        failure_summary=StartupFailureSummary.from_exception(
            exception,
            diagnostic_details,
        ),
        diagnostic_details=diagnostic_details,
    )

options: show_root_heading: false

spakky.core.application.error

Error types for application-level exceptions.

This module defines the base error class for all application-related errors.

AbstractSpakkyApplicationError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for all application-level errors.

options: show_root_heading: false

AOP

spakky.core.aop.aspect

Aspect annotations for declaring AOP aspects as Pods.

This module provides decorators for marking classes as aspects that can intercept method calls across the application.

Aspect(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Pod decorator for synchronous aspects.

Marks a class as an aspect that can intercept synchronous method calls. The class must implement IAspect interface.

matches(pod)

Check if this aspect should be applied to a pod.

Parameters:

Name	Type	Description	Default
pod	object	The pod object to check for matching pointcuts.	required

Returns:

Name	Type	Description
bool	bool	True if any pointcut matches the pod, False otherwise.

Raises:

Type	Description
AspectInheritanceError	If target is not a class Pod or doesn't implement IAspect.

Source code in core/spakky/src/spakky/core/aop/aspect.py

def matches(self, pod: object) -> bool:
    """Check if this aspect should be applied to a pod.

    Args:
        pod: The pod object to check for matching pointcuts.

    Returns:
        bool: True if any pointcut matches the pod, False otherwise.

    Raises:
        AspectInheritanceError: If target is not a class Pod or doesn't implement IAspect.
    """
    # Check if pod itself is callable and matches any pointcut
    if callable(pod):
        for annotation, target_method in self.pointcuts.items():
            if (advice := annotation.get_or_none(target_method)) is not None:
                if advice.matches(pod):
                    return True
    # Use get_callable_methods to avoid invoking property getters
    pod_methods = get_callable_methods(pod)
    for annotation, target_method in self.pointcuts.items():
        if (advice := annotation.get_or_none(target_method)) is not None:
            for _, method in pod_methods:
                if advice.matches(method):
                    return True
    return False

AsyncAspect(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Pod decorator for asynchronous aspects.

Marks a class as an aspect that can intercept asynchronous method calls. The class must implement IAsyncAspect interface.

matches(pod)

Check if this async aspect should be applied to a pod.

Parameters:

Name	Type	Description	Default
pod	object	The pod object to check for matching pointcuts.	required

Returns:

Name	Type	Description
bool	bool	True if any pointcut matches the pod, False otherwise.

Raises:

Type	Description
AspectInheritanceError	If target is not a class Pod or doesn't implement IAsyncAspect.

Source code in core/spakky/src/spakky/core/aop/aspect.py

def matches(self, pod: object) -> bool:
    """Check if this async aspect should be applied to a pod.

    Args:
        pod: The pod object to check for matching pointcuts.

    Returns:
        bool: True if any pointcut matches the pod, False otherwise.

    Raises:
        AspectInheritanceError: If target is not a class Pod or doesn't implement IAsyncAspect.
    """
    # Check if pod itself is callable and matches any pointcut
    if callable(pod):
        for annotation, target_method in self.pointcuts.items():
            if (advice := annotation.get_or_none(target_method)) is not None:
                if advice.matches(pod):
                    return True
    # Use get_callable_methods to avoid invoking property getters
    pod_methods = get_callable_methods(pod)
    for annotation, target_method in self.pointcuts.items():
        if (advice := annotation.get_or_none(target_method)) is not None:
            for _, method in pod_methods:
                if advice.matches(method):
                    return True
    return False

options: show_root_heading: false

spakky.core.aop.advisor

Advisor classes that coordinate aspect execution around method calls.

Advisors wrap methods with aspect logic, executing before/around/after advice in the correct order and handling exceptions appropriately.

Advisor(instance, next)

Advisor for synchronous methods with aspect interception.

Initialize the advisor.

Parameters:

Name	Type	Description	Default
instance	IAspect	The aspect instance providing advice.	required
next	Func	The next function to call in the chain.	required

Source code in core/spakky/src/spakky/core/aop/advisor.py

def __init__(self, instance: IAspect, next: Func) -> None:
    """Initialize the advisor.

    Args:
        instance: The aspect instance providing advice.
        next: The next function to call in the chain.
    """
    self.instance = instance
    self.next = next

instance = instance instance-attribute

The aspect instance that provides advice.

next = next instance-attribute

The next function in the advice chain.

__getattr__(name)

Delegate attribute access to the next function in the chain.

Source code in core/spakky/src/spakky/core/aop/advisor.py

def __getattr__(self, name: str) -> Any:
    """Delegate attribute access to the next function in the chain."""
    return getattr(
        self.next, name
    )  # framework proxy delegates wrapped callable attrs

__call__(*args, **kwargs)

Execute the advised method with aspect logic.

Coordinates the execution of before, around, after_returning, after_raising, and after advice in the correct order.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the method.	()
**kwargs	Any	Keyword arguments for the method.	{}

Returns:

Name	Type	Description
Any	Any	The result of the method call.

Raises:

Type	Description
Exception	Any exception raised by the advised method or aspect.

Source code in core/spakky/src/spakky/core/aop/advisor.py

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Execute the advised method with aspect logic.

    Coordinates the execution of before, around, after_returning, after_raising,
    and after advice in the correct order.

    Args:
        *args: Positional arguments for the method.
        **kwargs: Keyword arguments for the method.

    Returns:
        Any: The result of the method call.

    Raises:
        Exception: Any exception raised by the advised method or aspect.
    """
    self.instance.before(*args, **kwargs)
    try:
        result = self.instance.around(self.next, *args, **kwargs)
        self.instance.after_returning(result)
        return result
    except Exception as e:
        self.instance.after_raising(e)
        raise
    finally:
        self.instance.after()

AsyncAdvisor(instance, next)

Advisor for asynchronous methods with aspect interception.

Initialize the async advisor.

Parameters:

Name	Type	Description	Default
instance	IAsyncAspect	The async aspect instance providing advice.	required
next	AsyncFunc	The next async function to call in the chain.	required

Source code in core/spakky/src/spakky/core/aop/advisor.py

def __init__(self, instance: IAsyncAspect, next: AsyncFunc) -> None:
    """Initialize the async advisor.

    Args:
        instance: The async aspect instance providing advice.
        next: The next async function to call in the chain.
    """
    self.instance = instance
    self.next = next

instance = instance instance-attribute

The async aspect instance that provides advice.

next = next instance-attribute

The next async function in the advice chain.

__getattr__(name)

Delegate attribute access to the next async function in the chain.

Source code in core/spakky/src/spakky/core/aop/advisor.py

def __getattr__(self, name: str) -> Any:
    """Delegate attribute access to the next async function in the chain."""
    return getattr(
        self.next, name
    )  # framework proxy delegates wrapped callable attrs

__call__(*args, **kwargs) async

Execute the advised async method with aspect logic.

Coordinates the execution of before, around, after_returning, after_raising, and after advice in the correct order for async methods.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the method.	()
**kwargs	Any	Keyword arguments for the method.	{}

Returns:

Name	Type	Description
Any	Any	The result of the async method call.

Raises:

Type	Description
Exception	Any exception raised by the advised method or aspect.

Source code in core/spakky/src/spakky/core/aop/advisor.py

async def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Execute the advised async method with aspect logic.

    Coordinates the execution of before, around, after_returning, after_raising,
    and after advice in the correct order for async methods.

    Args:
        *args: Positional arguments for the method.
        **kwargs: Keyword arguments for the method.

    Returns:
        Any: The result of the async method call.

    Raises:
        Exception: Any exception raised by the advised method or aspect.
    """
    await self.instance.before_async(*args, **kwargs)
    try:
        result = await self.instance.around_async(self.next, *args, **kwargs)
        await self.instance.after_returning_async(result)
        return result
    except Exception as e:
        await self.instance.after_raising_async(e)
        raise
    finally:
        await self.instance.after_async()

options: show_root_heading: false

spakky.core.aop.pointcut

Pointcut annotations for defining when aspects should be applied.

This module provides decorators for specifying when aspect advice should intercept method calls (before, after, around, etc.).

AbstractPointCut(pointcut) dataclass

Bases: FunctionAnnotation, ABC

Base class for pointcut annotations.

pointcut instance-attribute

Predicate function that determines if a method matches this pointcut.

matches(method)

Check if a method matches this pointcut.

Parameters:

Name	Type	Description	Default
method	Func	The method to check.	required

Returns:

Name	Type	Description
bool	bool	True if the method matches the pointcut predicate, False otherwise.

Source code in core/spakky/src/spakky/core/aop/pointcut.py

def matches(self, method: Func) -> bool:
    """Check if a method matches this pointcut.

    Args:
        method: The method to check.

    Returns:
        bool: True if the method matches the pointcut predicate, False otherwise.
    """
    return self.pointcut(method)

Before(pointcut) dataclass

Bases: AbstractPointCut

Pointcut for advice that executes before a method call.

AfterReturning(pointcut) dataclass

Bases: AbstractPointCut

Pointcut for advice that executes after a method returns successfully.

AfterRaising(pointcut) dataclass

Bases: AbstractPointCut

Pointcut for advice that executes after a method raises an exception.

After(pointcut) dataclass

Bases: AbstractPointCut

Pointcut for advice that executes after a method call (regardless of outcome).

Around(pointcut) dataclass

Bases: AbstractPointCut

Pointcut for advice that wraps around a method call, controlling its execution.

options: show_root_heading: false

spakky.core.aop.interfaces

options: show_root_heading: false

spakky.core.aop.interfaces.aspect

Interface definitions for aspect implementations.

This module defines the protocols that aspect classes must implement to intercept method calls in the AOP system.

IAspect

Bases: ABC

Interface for synchronous aspect implementations.

before(*args, **kwargs)

Execute before the target method is called.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the target method.	()
**kwargs	Any	Keyword arguments for the target method.	{}

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

def before(self, *args: Any, **kwargs: Any) -> None:
    """Execute before the target method is called.

    Args:
        *args: Positional arguments for the target method.
        **kwargs: Keyword arguments for the target method.
    """
    return

after_raising(error)

Execute after the target method raises an exception.

Parameters:

Name	Type	Description	Default
error	Exception	The exception that was raised.	required

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

def after_raising(self, error: Exception) -> None:
    """Execute after the target method raises an exception.

    Args:
        error: The exception that was raised.
    """
    return

after_returning(result)

Execute after the target method returns successfully.

Parameters:

Name	Type	Description	Default
result	Any	The return value from the target method.	required

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

def after_returning(self, result: Any) -> None:
    """Execute after the target method returns successfully.

    Args:
        result: The return value from the target method.
    """
    return

after()

Execute after the target method completes (regardless of outcome).

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

def after(self) -> None:
    """Execute after the target method completes (regardless of outcome)."""
    return

around(joinpoint, *args, **kwargs)

Wrap around the target method execution.

Parameters:

Name	Type	Description	Default
joinpoint	Func	The target method to be executed.	required
*args	Any	Positional arguments for the target method.	()
**kwargs	Any	Keyword arguments for the target method.	{}

Returns:

Type	Description
Any	The result of the joinpoint execution.

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

def around(
    self,
    joinpoint: Func,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Wrap around the target method execution.

    Args:
        joinpoint: The target method to be executed.
        *args: Positional arguments for the target method.
        **kwargs: Keyword arguments for the target method.

    Returns:
        The result of the joinpoint execution.
    """
    return joinpoint(*args, **kwargs)

IAsyncAspect

Bases: ABC

Interface for asynchronous aspect implementations.

before_async(*args, **kwargs) async

Execute before the target async method is called.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the target method.	()
**kwargs	Any	Keyword arguments for the target method.	{}

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

async def before_async(self, *args: Any, **kwargs: Any) -> None:
    """Execute before the target async method is called.

    Args:
        *args: Positional arguments for the target method.
        **kwargs: Keyword arguments for the target method.
    """
    return

after_raising_async(error) async

Execute after the target async method raises an exception.

Parameters:

Name	Type	Description	Default
error	Exception	The exception that was raised.	required

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

async def after_raising_async(self, error: Exception) -> None:
    """Execute after the target async method raises an exception.

    Args:
        error: The exception that was raised.
    """
    return

after_returning_async(result) async

Execute after the target async method returns successfully.

Parameters:

Name	Type	Description	Default
result	Any	The return value from the target method.	required

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

async def after_returning_async(self, result: Any) -> None:
    """Execute after the target async method returns successfully.

    Args:
        result: The return value from the target method.
    """
    return

after_async() async

Execute after the target async method completes (regardless of outcome).

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

async def after_async(self) -> None:
    """Execute after the target async method completes (regardless of outcome)."""
    return

around_async(joinpoint, *args, **kwargs) async

Wrap around the target async method execution.

Parameters:

Name	Type	Description	Default
joinpoint	AsyncFunc	The target async method to be executed.	required
*args	Any	Positional arguments for the target method.	()
**kwargs	Any	Keyword arguments for the target method.	{}

Returns:

Type	Description
Any	The result of the joinpoint execution.

Source code in core/spakky/src/spakky/core/aop/interfaces/aspect.py

async def around_async(
    self,
    joinpoint: AsyncFunc,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Wrap around the target async method execution.

    Args:
        joinpoint: The target async method to be executed.
        *args: Positional arguments for the target method.
        **kwargs: Keyword arguments for the target method.

    Returns:
        The result of the joinpoint execution.
    """
    return await joinpoint(*args, **kwargs)

options: show_root_heading: false

spakky.core.aop.error

Error classes for AOP-related exceptions.

AbstractSpakkyAOPError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for all AOP-related errors.

AspectInheritanceError

Bases: AbstractSpakkyAOPError

Raised when an aspect class doesn't implement required interfaces.

Aspect classes must inherit from either IAspect (for sync) or IAsyncAspect (for async).

options: show_root_heading: false

Pod

spakky.core.pod.annotations

options: show_root_heading: false

spakky.core.pod.annotations.pod

Pod annotation for dependency injection container registration.

This module provides the core @Pod decorator that marks classes and functions as managed beans in the IoC container, along with dependency resolution logic.

DependencyInfo(name, type_, has_default=False, is_optional=False, qualifiers=list[Qualifier](), collection_kind=None, collection_key_type=None) dataclass

Information about a Pod's dependency for injection.

Attributes:

Name	Type	Description
name	str	The parameter name of the dependency.
type_	Class	The type of the dependency.
has_default	bool	Whether the dependency has a default value.
is_optional	bool	Whether the dependency is optional (can be None).
qualifiers	list[Qualifier]	List of qualifiers for disambiguation.

DependencyCollectionKind

Bases: Enum

Supported collection dependency injection shapes.

LIST = auto() class-attribute instance-attribute

Inject all matching Pods as a stable list.

TUPLE = auto() class-attribute instance-attribute

Inject all matching Pods as a stable tuple.

DICT = auto() class-attribute instance-attribute

Inject matching Pods keyed by registered Pod name.

CannotDeterminePodTypeError

Bases: PodAnnotationFailedError

Raised when Pod type cannot be inferred from annotations.

CannotUseVarArgsInPodError

Bases: PodAnnotationFailedError

Raised when args or *kwargs are used in Pod dependencies.

CannotUsePositionalOnlyArgsInPodError

Bases: PodAnnotationFailedError

Raised when positional-only arguments are used in Pod.

CannotUseOptionalReturnTypeInPodError

Bases: PodAnnotationFailedError

Raised when function Pod has Optional return type.

UnsupportedCollectionDependencyTypeError

Bases: PodAnnotationFailedError

Raised when a collection dependency annotation is unsupported.

UnexpectedDependencyNameInjectedError

Bases: PodInstantiationFailedError

Raised when an unexpected dependency name is injected.

UnexpectedDependencyTypeInjectedError(*args, dependency_diagnostic=None)

Bases: PodInstantiationFailedError

Raised when an injected dependency has wrong type.

Initialize with optional dependency graph diagnostics.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def __init__(
    self,
    *args: object,
    dependency_diagnostic: PodDependencyResolutionDiagnostic | None = None,
) -> None:
    """Initialize with optional dependency graph diagnostics."""
    self.dependency_diagnostic = dependency_diagnostic
    super().__init__(*args)

Pod(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Annotation, IEquatable

Annotation for marking classes and functions as managed Pods in the IoC container.

Pods are automatically instantiated by the container with their dependencies injected.

id = field(init=False, default_factory=uuid4) class-attribute instance-attribute

Unique identifier for this Pod instance.

name = field(kw_only=True, default='') class-attribute instance-attribute

Optional name for qualifying this Pod.

scope = field(kw_only=True, default=(Scope.SINGLETON)) class-attribute instance-attribute

The lifecycle scope of this Pod.

type_ = field(init=False) class-attribute instance-attribute

The resolved type of this Pod.

base_types = field(init=False, default_factory=(set[type])) class-attribute instance-attribute

Set of base types and interfaces this Pod implements.

target = field(init=False) class-attribute instance-attribute

The target class or function being registered as a Pod.

dependencies = field(init=False, default_factory=dict) class-attribute instance-attribute

Map of dependency names to their injection information.

is_primary property

Check if this Pod is marked as primary.

Returns:

Type	Description
bool	True if the target has @Primary annotation.

dependency_qualifiers property

Get qualifiers for all dependencies.

Returns:

Type	Description
dict[str, list[Qualifier]]	Map of dependency names to their qualifier annotations.

Scope

Bases: Enum

Lifecycle scope for Pod instances.

SINGLETON = auto() class-attribute instance-attribute

One instance shared across the entire application.

PROTOTYPE = auto() class-attribute instance-attribute

New instance created on each request.

CONTEXT = auto() class-attribute instance-attribute

Instance scoped to request/context lifecycle.

__runtime_base_types(type_)

Return base types indexed by the runtime container.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def __runtime_base_types(self, type_: type) -> set[type]:
    """Return base types indexed by the runtime container."""
    return set(generic_mro(type_))

__get_dependencies(obj)

Extract dependency information from constructor or function parameters.

Parameters:

Name	Type	Description	Default
obj	PodType	The class or function to analyze for dependencies.	required

Returns:

Type	Description
DependencyMap	Map of parameter names to their dependency information.

Raises:

Type	Description
CannotUsePositionalOnlyArgsInPodError	If positional-only parameters are found.
CannotUseVarArgsInPodError	If args or *kwargs are found.
CannotDeterminePodTypeError	If parameter has no type annotation.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def __get_dependencies(self, obj: PodType) -> DependencyMap:
    """Extract dependency information from constructor or function parameters.

    Args:
        obj: The class or function to analyze for dependencies.

    Returns:
        Map of parameter names to their dependency information.

    Raises:
        CannotUsePositionalOnlyArgsInPodError: If positional-only parameters are found.
        CannotUseVarArgsInPodError: If *args or **kwargs are found.
        CannotDeterminePodTypeError: If parameter has no type annotation.
    """
    if isclass(obj):
        if has_default_constructor(obj):
            # If obj is a class with a default constructor,
            # then return an empty dictionary
            return {}
        obj = obj.__init__  # Get constructor if obj is a class
    parameters: list[Parameter] = list(inspect.signature(obj).parameters.values())
    if is_instance_method(obj):
        # Remove self parameter if obj is an instance method
        parameters = parameters[1:]

    annotation_globalns = self.__annotation_globalns(obj)
    dependencies: DependencyMap = {}
    for parameter in parameters:
        if parameter.kind == Parameter.POSITIONAL_ONLY:
            raise CannotUsePositionalOnlyArgsInPodError(obj, parameter.name)
        if parameter.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
            raise CannotUseVarArgsInPodError(obj, parameter.name)
        if parameter.annotation == Parameter.empty:
            raise CannotDeterminePodTypeError(obj, parameter.name)
        annotation = self.__resolve_parameter_annotation(
            annotation=parameter.annotation,
            globalns=annotation_globalns,
        )
        type_, qualifiers, is_optional_dependency = (
            self.__normalize_dependency_annotation(annotation)
        )
        has_default = parameter.default != Parameter.empty
        collection_dependency = self.__collection_dependency_info(
            name=parameter.name,
            annotation=type_,
            has_default=has_default,
            is_optional_dependency=is_optional_dependency,
            qualifiers=qualifiers,
        )
        if collection_dependency is not None:
            dependencies[parameter.name] = collection_dependency
            continue
        if not self.__is_dependency_type(type_):
            raise CannotDeterminePodTypeError(obj, parameter.name)
        dependencies[parameter.name] = DependencyInfo(
            name=parameter.name,
            type_=type_,
            is_optional=is_optional_dependency,
            has_default=has_default,
            qualifiers=qualifiers,
        )

    return dependencies

__call__(obj)

Apply Pod annotation to target class or function.

Parameters:

Name	Type	Description	Default
obj	T	The class or function to decorate.	required

Returns:

Type	Description
T	The original object unchanged.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def __call__[T: PodType](self, obj: T) -> T:
    """Apply Pod annotation to target class or function.

    Args:
        obj: The class or function to decorate.

    Returns:
        The original object unchanged.
    """
    self._initialize(obj)
    return super().__call__(obj)

__hash__()

Compute hash based on Pod name.

Returns:

Type	Description
int	Hash value for this Pod.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def __hash__(self) -> int:
    """Compute hash based on Pod name.

    Returns:
        Hash value for this Pod.
    """
    return hash(self.name)

__eq__(value)

Check equality based on Pod name.

Parameters:

Name	Type	Description	Default
value	object	The object to compare with.	required

Returns:

Type	Description
bool	True if both Pods have the same name.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def __eq__(self, value: object) -> bool:
    """Check equality based on Pod name.

    Args:
        value: The object to compare with.

    Returns:
        True if both Pods have the same name.
    """
    if self is value:  # pragma: no cover - coverage boundary
        return True
    if not isinstance(value, Pod):
        return False  # pragma: no cover - coverage boundary (Pod 메타데이터 비교에서 다른 타입과 비교되지 않음)
    return self.name == value.name

is_family_with(type_)

Check if this Pod is compatible with a given type.

Parameters:

Name	Type	Description	Default
type_	type	The type to check compatibility with.	required

Returns:

Type	Description
bool	True if type matches Pod type or is in its base types.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def is_family_with(self, type_: type) -> bool:
    """Check if this Pod is compatible with a given type.

    Args:
        type_: The type to check compatibility with.

    Returns:
        True if type matches Pod type or is in its base types.
    """
    return type_ == self.type_ or type_ in self.base_types

instantiate(dependencies)

Create an instance of this Pod with injected dependencies.

Parameters:

Name	Type	Description	Default
dependencies	dict[str, object | None]	Map of dependency names to their resolved instances.	required

Returns:

Type	Description
object	The instantiated Pod object.

Raises:

Type	Description
UnexpectedDependencyNameInjectedError	If unknown dependency name provided.
UnexpectedDependencyTypeInjectedError	If required non-optional dependency is None.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def instantiate(self, dependencies: dict[str, object | None]) -> object:
    """Create an instance of this Pod with injected dependencies.

    Args:
        dependencies: Map of dependency names to their resolved instances.

    Returns:
        The instantiated Pod object.

    Raises:
        UnexpectedDependencyNameInjectedError: If unknown dependency name provided.
        UnexpectedDependencyTypeInjectedError: If required non-optional dependency is None.
    """
    final_dependencies: dict[str, object] = {}
    for name, dependency in dependencies.items():
        if name not in self.dependencies:  # pragma: no cover - coverage boundary
            raise UnexpectedDependencyNameInjectedError(self.type_, name)
        dependency_info: DependencyInfo = self.dependencies[name]
        if dependency is None:
            if dependency_info.has_default:
                # If dependency is None and has a default value,
                # do not include it in the final dependencies
                # so, the default value will be used
                continue
            if (
                not dependency_info.is_optional
            ):  # pragma: no cover - coverage boundary
                raise UnexpectedDependencyTypeInjectedError(
                    self.type_,
                    {
                        "name": name,
                        "expected": dependency_info.type_,
                        "actual": NoneType,
                    },
                )
        final_dependencies[name] = dependency
    return self.target(**final_dependencies)

is_class_pod(pod)

Check if a Pod target is a class.

Parameters:

Name	Type	Description	Default
pod	PodType	The Pod target to check.	required

Returns:

Type	Description
TypeGuard[Class]	True if pod is a class type.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def is_class_pod(pod: PodType) -> TypeGuard[Class]:
    """Check if a Pod target is a class.

    Args:
        pod: The Pod target to check.

    Returns:
        True if pod is a class type.
    """
    return isclass(pod)

is_function_pod(pod)

Check if a Pod target is a function.

Parameters:

Name	Type	Description	Default
pod	PodType	The Pod target to check.	required

Returns:

Type	Description
TypeGuard[Func]	True if pod is a function.

Source code in core/spakky/src/spakky/core/pod/annotations/pod.py

def is_function_pod(pod: PodType) -> TypeGuard[Func]:
    """Check if a Pod target is a function.

    Args:
        pod: The Pod target to check.

    Returns:
        True if pod is a function.
    """
    return isfunction(pod)

options: show_root_heading: false

spakky.core.pod.annotations.lazy

Annotation for lazy Pod initialization.

This module provides the @Lazy annotation to defer Pod initialization until first use.

Lazy() dataclass

Bases: ClassAnnotation

Mark a Pod for lazy initialization.

Pods marked with @Lazy are not instantiated during application startup but only when first requested from the container.

Warning

When using @Lazy with singleton-scoped Pods in multi-threaded environments, the first concurrent access may result in multiple instantiations. To ensure thread-safety for lazy singletons, consider using a factory pattern or initializing the Pod during application startup by removing @Lazy.

Example

@Lazy() @Pod() class ExpensiveService: def init(self): # Heavy initialization deferred until first use pass

options: show_root_heading: false

spakky.core.pod.annotations.order

Annotation for controlling execution order.

This module provides the @Order annotation to control the order of post-processor execution and aspect application.

Order(order=sys.maxsize) dataclass

Bases: ClassAnnotation

Control execution order of Pods.

Lower order values execute first. Default is sys.maxsize (last). Commonly used for post-processors and aspects.

order = field(default=(sys.maxsize)) class-attribute instance-attribute

Execution order priority (lower executes first).

__post_init__()

Validate order value.

Raises:

Type	Description
ValueError	If order is negative.

Source code in core/spakky/src/spakky/core/pod/annotations/order.py

def __post_init__(self) -> None:
    """Validate order value.

    Raises:
        ValueError: If order is negative.
    """
    if self.order < 0:
        raise NegativeOrderValueError

options: show_root_heading: false

spakky.core.pod.annotations.primary

Annotation for marking preferred Pod implementation.

This module provides the @Primary annotation to indicate the preferred implementation when multiple Pods of the same type exist.

Primary() dataclass

Bases: ClassAnnotation

Mark a Pod as the primary implementation.

When multiple Pods match a dependency type, the one marked with @Primary is selected by default without requiring explicit qualification.

options: show_root_heading: false

spakky.core.pod.annotations.qualifier

Metadata for qualifying dependency injection.

This module provides the Qualifier class for creating custom dependency qualifiers in Annotated type hints.

Qualifier(selector) dataclass

Bases: AbstractMetadata

Metadata for qualifying which Pod to inject.

Used in Annotated type hints to select specific Pods when multiple candidates exist.

Example

@Pod() class Service: def init( self, repo: Annotated[IRepo, Qualifier(lambda p: p.name == "primary")], ) -> None: self.repo = repo

selector instance-attribute

Predicate function to filter Pod candidates.

__post_init__()

Validate that selector is callable.

Raises:

Type	Description
TypeError	If selector is not callable.

Source code in core/spakky/src/spakky/core/pod/annotations/qualifier.py

def __post_init__(self) -> None:
    """Validate that selector is callable.

    Raises:
        TypeError: If selector is not callable.
    """
    if not callable(self.selector):
        raise QualifierSelectorNotCallableError

options: show_root_heading: false

spakky.core.pod.annotations.tag

Tag() dataclass

Bases: Annotation, IEquatable

Base class for custom metadata tags attached to Pods.

__eq__(value)

Compare tags by their dataclass field values.

Source code in core/spakky/src/spakky/core/pod/annotations/tag.py

def __eq__(self, value: object) -> bool:
    """Compare tags by their dataclass field values."""
    if self is value:
        return True
    if not isinstance(value, Tag):
        return False
    return astuple(self) == astuple(value)

__hash__()

Hash based on dataclass field values.

Source code in core/spakky/src/spakky/core/pod/annotations/tag.py

def __hash__(self) -> int:
    """Hash based on dataclass field values."""
    return hash(astuple(self))

options: show_root_heading: false

spakky.core.pod.binding

Binding policy values for dependency injection candidate selection.

PodBinding

Explicit interface-to-implementation binding policy.

Application or feature configuration can register this value with ApplicationContext to select one implementation when multiple Pods provide the same interface.

interface instance-attribute

Requested interface or port type.

implementation_type = None class-attribute instance-attribute

Concrete implementation type to select.

implementation_name = None class-attribute instance-attribute

Registered Pod name to select.

options: show_root_heading: false

spakky.core.pod.diagnostics

Structured diagnostics for Pod dependency resolution.

PodCandidateDiagnostic

One candidate Pod observed during dependency resolution.

pod_name instance-attribute

Registered Pod name for the candidate.

pod_type_name instance-attribute

Registered Pod type name for the candidate.

is_primary instance-attribute

Whether the candidate is marked with @Primary.

PodDependencyPathNode

One Pod dependency graph edge observed during resolution.

pod_name instance-attribute

Registered Pod name for the node being instantiated.

pod_type_name instance-attribute

Registered Pod type name for the node being instantiated.

dependency_parameter_name = None class-attribute instance-attribute

Constructor or factory parameter that requested the next dependency.

requested_type_name = None class-attribute instance-attribute

Requested dependency type name for this graph edge.

PodDependencyResolutionDiagnostic

Structured dependency resolution failure details.

failed_pod_name instance-attribute

Registered Pod name where resolution failed.

failed_pod_type_name instance-attribute

Registered Pod type name where resolution failed.

dependency_parameter_name instance-attribute

Dependency parameter that could not be resolved, when applicable.

requested_type_name instance-attribute

Requested dependency type name that failed, when applicable.

path instance-attribute

Dependency path from the root Pod to the failure point.

candidates = () class-attribute instance-attribute

Candidate Pods involved in an ambiguity, when applicable.

resolution_hints = () class-attribute instance-attribute

Stable human-readable actions that can resolve the failure.

as_detail_pairs()

Return stable key/value details for report adapters.

Source code in core/spakky/src/spakky/core/pod/diagnostics.py

def as_detail_pairs(self) -> tuple[tuple[str, str], ...]:
    """Return stable key/value details for report adapters."""
    path_value = " -> ".join(
        node.pod_type_name
        if node.dependency_parameter_name is None
        else (
            f"{node.pod_type_name}.{node.dependency_parameter_name}"
            f":{node.requested_type_name}"
        )
        for node in self.path
    )
    details = (
        ("failed_pod", self.failed_pod_name),
        ("failed_pod_type", self.failed_pod_type_name),
        ("dependency_path", path_value),
    )
    if self.dependency_parameter_name is None:
        dependency_details = details
    else:
        dependency_details = details + (
            ("dependency_parameter", self.dependency_parameter_name),
            ("requested_type", self.requested_type_name or ""),
        )
    if not self.candidates:
        return dependency_details
    candidate_value = ", ".join(
        f"{candidate.pod_name}:{candidate.pod_type_name}"
        f":primary={candidate.is_primary}"
        for candidate in self.candidates
    )
    hint_value = "; ".join(self.resolution_hints)
    return dependency_details + (
        ("candidates", candidate_value),
        ("resolution_hints", hint_value),
    )

options: show_root_heading: false

spakky.core.pod.inject

Utility function for manual dependency injection.

This module provides the inject() function for programmatic Pod retrieval from the container when constructor injection is not available.

inject(context, type_, name=None)

inject(context: IContainer, type_: type[T]) -> T

inject(context: IContainer, type_: type[T], name: str) -> T

Manually inject a Pod from the container.

Parameters:

Name	Type	Description	Default
context	IContainer	The container to retrieve from.	required
type_	type[T]	The type of Pod to retrieve.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
object | T	The requested Pod instance.

Example

service = inject(app.container, UserService) named_service = inject(app.container, IRepo, "postgres")

Source code in core/spakky/src/spakky/core/pod/inject.py

def inject[T: object](
    context: IContainer,
    type_: type[T],
    name: str | None = None,
) -> object | T:
    """Manually inject a Pod from the container.

    Args:
        context: The container to retrieve from.
        type_: The type of Pod to retrieve.
        name: Optional name qualifier.

    Returns:
        The requested Pod instance.

    Example:
        service = inject(app.container, UserService)
        named_service = inject(app.container, IRepo, "postgres")
    """
    if name is not None:
        return context.get(type_=type_, name=name)
    return context.get(type_=type_)

options: show_root_heading: false

spakky.core.pod.interfaces.container

Interface and errors for Pod container interface.

This module defines the IContainer protocol for managing Pod lifecycle and dependency injection.

CircularDependencyGraphDetectedError(dependency_chain, dependency_diagnostic=None)

Bases: AbstractSpakkyPodError, IRepresentable

Raised when circular dependency is detected during Pod instantiation.

Attributes:

Name	Type	Description
dependency_chain	list[type]	List of types showing the circular dependency path.

Initialize with dependency chain information.

Parameters:

Name	Type	Description	Default
dependency_chain	list[type]	List of types in dependency order, ending with the duplicate type.	required
dependency_diagnostic	PodDependencyResolutionDiagnostic | None	Optional structured dependency path details.	None

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

def __init__(
    self,
    dependency_chain: list[type],
    dependency_diagnostic: PodDependencyResolutionDiagnostic | None = None,
) -> None:
    """Initialize with dependency chain information.

    Args:
        dependency_chain: List of types in dependency order, ending with the duplicate type.
        dependency_diagnostic: Optional structured dependency path details.
    """
    self.dependency_chain = dependency_chain
    self.dependency_diagnostic = dependency_diagnostic
    if not dependency_chain:
        super().__init__(self.message)
        return

    lines = [self.message, "Dependency path:"]
    for i, type_ in enumerate(dependency_chain):
        type_name = (
            type_.__name__
            if hasattr(type_, "__name__")  # diagnostic fallback for typing aliases
            else str(type_)
        )
        indent = "  " * i
        arrow = "└─> " if i > 0 else ""

        if i == len(dependency_chain) - 1:
            lines.append(f"{indent}{arrow}{type_name} (CIRCULAR!)")
        else:
            lines.append(f"{indent}{arrow}{type_name}")

    super().__init__("\n".join(lines))

NoSuchPodError

Bases: AbstractSpakkyPodError

Raised when requested Pod cannot be found in container.

NoUniquePodError(requested_type, candidates, dependency_diagnostic=None, resolution_hints=())

Bases: AbstractSpakkyPodError

Raised when multiple Pods match criteria without clear qualification.

Initialize ambiguity details.

Parameters:

Name	Type	Description	Default
requested_type	type	Dependency type requested by the caller.	required
candidates	tuple[PodCandidateDiagnostic, ...]	Candidate Pods that matched the requested type.	required
dependency_diagnostic	PodDependencyResolutionDiagnostic | None	Optional structured dependency path.	None
resolution_hints	tuple[str, ...]	Actions that can make the selection explicit.	()

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

def __init__(
    self,
    requested_type: type,
    candidates: tuple[PodCandidateDiagnostic, ...],
    dependency_diagnostic: PodDependencyResolutionDiagnostic | None = None,
    resolution_hints: tuple[str, ...] = (),
) -> None:
    """Initialize ambiguity details.

    Args:
        requested_type: Dependency type requested by the caller.
        candidates: Candidate Pods that matched the requested type.
        dependency_diagnostic: Optional structured dependency path.
        resolution_hints: Actions that can make the selection explicit.
    """
    self.requested_type = requested_type
    self.candidates = candidates
    self.dependency_diagnostic = dependency_diagnostic
    self.resolution_hints = resolution_hints
    candidate_names = ", ".join(candidate.pod_name for candidate in candidates)
    super().__init__(
        "\n".join(
            (
                self.message,
                "Requested type: " + requested_type.__name__,
                "Candidates: " + candidate_names,
                "Resolution: " + "; ".join(resolution_hints),
            )
        )
    )

InvalidPodBindingError

Bases: AbstractSpakkyPodError

Raised when a binding policy does not identify exactly one target kind.

NoSuchPodBindingTargetError(binding)

Bases: AbstractSpakkyPodError

Raised when an explicit binding does not match any candidate Pod.

Initialize binding target details.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

def __init__(self, binding: PodBinding) -> None:
    """Initialize binding target details."""
    self.binding = binding
    super().__init__(self.message)

PodBindingNotSupportedError

Bases: AbstractSpakkyPodError

Raised when a container does not implement explicit binding policies.

CannotRegisterNonPodObjectError

Bases: AbstractSpakkyPodError

Raised when attempting to register object without @Pod annotation.

PodNameAlreadyExistsError

Bases: AbstractSpakkyPodError

Raised when Pod name conflicts with existing registration.

IContainer

Bases: ABC

Interface for IoC container managing Pod instances.

pods abstractmethod property

Get all registered Pods.

Returns:

Type	Description
dict[str, Pod]	Dictionary mapping Pod names to Pod metadata.

add(obj) abstractmethod

Register a Pod in the container.

Parameters:

Name	Type	Description	Default
obj	PodType	The Pod-annotated class or function to register.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

@abstractmethod
def add(self, obj: PodType) -> None:
    """Register a Pod in the container.

    Args:
        obj: The Pod-annotated class or function to register.
    """
    ...

bind(binding)

Register an explicit interface-to-implementation binding policy.

Parameters:

Name	Type	Description	Default
binding	PodBinding	Binding value supplied by application or feature config.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

def bind(self, binding: PodBinding) -> None:
    """Register an explicit interface-to-implementation binding policy.

    Args:
        binding: Binding value supplied by application or feature config.
    """
    raise PodBindingNotSupportedError

bind_to_type(interface, implementation)

Bind an interface to a concrete implementation type.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

def bind_to_type(self, interface: type, implementation: type) -> None:
    """Bind an interface to a concrete implementation type."""
    raise PodBindingNotSupportedError

bind_to_name(interface, name)

Bind an interface to a registered Pod name.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

def bind_to_name(self, interface: type, name: str) -> None:
    """Bind an interface to a registered Pod name."""
    raise PodBindingNotSupportedError

get(type_, name=None) abstractmethod

get(type_: type[T]) -> T

get(type_: type[T], name: str) -> T

Get a Pod instance by type and optional name.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type to retrieve.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
T | object	The Pod instance.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

@abstractmethod
def get[T: object](
    self,
    type_: type[T],
    name: str | None = None,
) -> T | object:
    """Get a Pod instance by type and optional name.

    Args:
        type_: The type to retrieve.
        name: Optional name qualifier.

    Returns:
        The Pod instance.
    """
    ...

get_or_none(type_, name=None) abstractmethod

get_or_none(type_: type[T]) -> T | None

get_or_none(type_: type[T], name: str) -> T | None

Get a Pod instance by type and optional name, or None if not found.

Parameters:

Name	Type	Description	Default
type_	type[T]	The type to retrieve.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
T | None	The Pod instance, or None if no matching Pod found.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

@abstractmethod
def get_or_none[T: object](
    self,
    type_: type[T],
    name: str | None = None,
) -> T | None:
    """Get a Pod instance by type and optional name, or None if not found.

    Args:
        type_: The type to retrieve.
        name: Optional name qualifier.

    Returns:
        The Pod instance, or None if no matching Pod found.
    """
    ...

contains(type_, name=None) abstractmethod

contains(type_: type) -> bool

contains(type_: type, name: str) -> bool

Check if a Pod is registered.

Parameters:

Name	Type	Description	Default
type_	type	The type to check.	required
name	str | None	Optional name qualifier.	None

Returns:

Type	Description
bool	True if matching Pod exists.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

@abstractmethod
def contains(
    self,
    type_: type,
    name: str | None = None,
) -> bool:
    """Check if a Pod is registered.

    Args:
        type_: The type to check.
        name: Optional name qualifier.

    Returns:
        True if matching Pod exists.
    """
    ...

find(selector) abstractmethod

Find all Pod instances matching selector predicate.

Parameters:

Name	Type	Description	Default
selector	Callable[[Pod], bool]	Predicate function to filter Pods.	required

Returns:

Type	Description
set[object]	Set of matching Pod instances.

Source code in core/spakky/src/spakky/core/pod/interfaces/container.py

@abstractmethod
def find(self, selector: Callable[[Pod], bool]) -> set[object]:
    """Find all Pod instances matching selector predicate.

    Args:
        selector: Predicate function to filter Pods.

    Returns:
        Set of matching Pod instances.
    """
    ...

options: show_root_heading: false

spakky.core.pod.interfaces.application_context

Interface and errors for application context interface.

This module defines the IApplicationContext protocol for managing application lifecycle and service coordination.

ApplicationContextAlreadyStartedError

Bases: AbstractSpakkyApplicationError

Raised when attempting to start an already started context.

ApplicationContextAlreadyStoppedError

Bases: AbstractSpakkyApplicationError

Raised when attempting to stop an already stopped context.

EventLoopThreadNotStartedInApplicationContextError

Bases: AbstractSpakkyApplicationError

Raised when event loop thread is not running but required.

EventLoopThreadAlreadyStartedInApplicationContextError

Bases: AbstractSpakkyApplicationError

Raised when attempting to start already running event loop thread.

IApplicationContext

Bases: IContainer, ITagRegistry, ABC

Interface for application context managing Pod lifecycle and services.

Extends IContainer with service management and lifecycle control.

thread_stop_event instance-attribute

Threading event for stopping background threads.

task_stop_event instance-attribute

Async event for stopping background tasks.

is_started abstractmethod property

Check if context is started.

Returns:

Type	Description
bool	True if context has been started.

add_service(service) abstractmethod

Register a service for lifecycle management.

Parameters:

Name	Type	Description	Default
service	IService | IAsyncService	The service to register.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def add_service(self, service: IService | IAsyncService) -> None:
    """Register a service for lifecycle management.

    Args:
        service: The service to register.
    """
    ...

start(startup_phase_recorder=None) abstractmethod

Start the application context and all services.

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def start(
    self,
    startup_phase_recorder: IStartupPhaseRecorder | None = None,
) -> None:
    """Start the application context and all services."""
    ...

stop() abstractmethod

Stop the application context and clean up resources.

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def stop(self) -> None:
    """Stop the application context and clean up resources."""
    ...

get_context_id() abstractmethod

Get unique ID for current context.

Returns:

Type	Description
UUID	UUID for this context.

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def get_context_id(self) -> UUID:
    """Get unique ID for current context.

    Returns:
        UUID for this context.
    """
    ...

get_context_value(key) abstractmethod

Get a value from the context-scoped cache.

Parameters:

Name	Type	Description	Default
key	str	The key to retrieve.	required

Returns:

Type	Description
object | None	The cached value, or None if not found.

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def get_context_value(self, key: str) -> object | None:
    """Get a value from the context-scoped cache.

    Args:
        key: The key to retrieve.

    Returns:
        The cached value, or None if not found.
    """
    ...

set_context_value(key, value) abstractmethod

Set a value in the context-scoped cache.

Parameters:

Name	Type	Description	Default
key	str	The key to set.	required
value	object	The value to store.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def set_context_value(self, key: str, value: object) -> None:
    """Set a value in the context-scoped cache.

    Args:
        key: The key to set.
        value: The value to store.
    """
    ...

clear_context() abstractmethod

Clear context-scoped cache for current context.

Source code in core/spakky/src/spakky/core/pod/interfaces/application_context.py

@abstractmethod
def clear_context(self) -> None:
    """Clear context-scoped cache for current context."""
    ...

options: show_root_heading: false

spakky.core.pod.interfaces.post_processor

Interface for Pod post-processors.

This module defines the IPostProcessor protocol for transforming Pod instances after creation but before use.

IPostProcessor

Bases: ABC

Interface for processing Pods after instantiation.

Post-processors can wrap, modify, or enhance Pod instances. Common uses include AOP proxy creation, dependency injection, and lifecycle management.

post_process(pod) abstractmethod

Process a Pod instance.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The processed Pod instance (may be wrapped or modified).

Source code in core/spakky/src/spakky/core/pod/interfaces/post_processor.py

@abstractmethod
def post_process(self, pod: object) -> object:
    """Process a Pod instance.

    Args:
        pod: The Pod instance to process.

    Returns:
        The processed Pod instance (may be wrapped or modified).
    """
    ...

options: show_root_heading: false

spakky.core.pod.interfaces.tag_registry

ITagRegistry

Bases: ABC

Registry for managing custom metadata tags attached to Pods.

tags abstractmethod property

All registered tags.

register_tag(tag) abstractmethod

Register a tag in the registry.

Source code in core/spakky/src/spakky/core/pod/interfaces/tag_registry.py

@abstractmethod
def register_tag(self, tag: Tag) -> None:
    """Register a tag in the registry."""
    ...

contains_tag(tag) abstractmethod

Check if a tag is registered.

Source code in core/spakky/src/spakky/core/pod/interfaces/tag_registry.py

@abstractmethod
def contains_tag(self, tag: Tag) -> bool:
    """Check if a tag is registered."""
    ...

list_tags(selector=None) abstractmethod

List tags, optionally filtered by a selector predicate.

Source code in core/spakky/src/spakky/core/pod/interfaces/tag_registry.py

@abstractmethod
def list_tags(
    self, selector: Callable[[Tag], bool] | None = None
) -> frozenset[Tag]:
    """List tags, optionally filtered by a selector predicate."""
    ...

options: show_root_heading: false

spakky.core.pod.interfaces.aware

options: show_root_heading: false

spakky.core.pod.interfaces.aware.aware

Base protocol for aware interfaces.

This module defines the base IAware marker protocol for dependency injection of framework services into Pods.

IAware

Bases: ABC

Marker protocol for Pods that require framework service injection.

Implementing this protocol allows Pods to receive framework services like logger, container, or application context through setter injection.

options: show_root_heading: false

spakky.core.pod.interfaces.aware.application_context_aware

Interface for application context injection.

This module defines IApplicationContextAware for Pods that need access to the application context.

IApplicationContextAware

Bases: IAware, ABC

Interface for Pods requiring application context injection.

Pods implementing this protocol will have set_application_context() called during post-processing with the current application context.

set_application_context(application_context) abstractmethod

Inject application context.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context instance.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/aware/application_context_aware.py

@abstractmethod
def set_application_context(self, application_context: IApplicationContext) -> None:
    """Inject application context.

    Args:
        application_context: The application context instance.
    """
    ...

options: show_root_heading: false

spakky.core.pod.interfaces.aware.container_aware

Interface for container injection.

This module defines IContainerAware for Pods that need access to the IoC container.

IContainerAware

Bases: IAware, ABC

Interface for Pods requiring container injection.

Pods implementing this protocol will have set_container() called during post-processing with the IoC container instance.

set_container(container) abstractmethod

Inject container.

Parameters:

Name	Type	Description	Default
container	IContainer	The IoC container instance.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/aware/container_aware.py

@abstractmethod
def set_container(self, container: IContainer) -> None:
    """Inject container.

    Args:
        container: The IoC container instance.
    """
    ...

options: show_root_heading: false

spakky.core.pod.interfaces.aware.tag_registry_aware

Interface for container injection.

This module defines IContainerAware for Pods that need access to the IoC container.

ITagRegistryAware

Bases: IAware, ABC

Interface for Pods requiring tag registry injection.

Pods implementing this protocol will have set_tag_registry() called during post-processing with the tag registry instance.

set_tag_registry(tag_registry) abstractmethod

Inject tag registry.

Parameters:

Name	Type	Description	Default
tag_registry	ITagRegistry	The tag registry instance.	required

Source code in core/spakky/src/spakky/core/pod/interfaces/aware/tag_registry_aware.py

@abstractmethod
def set_tag_registry(self, tag_registry: ITagRegistry) -> None:
    """Inject tag registry.

    Args:
        tag_registry: The tag registry instance.
    """
    ...

options: show_root_heading: false

spakky.core.pod.post_processors.aware_post_processor

Post-processor for injecting framework services into aware Pods.

This module provides ApplicationContextAwareProcessor which injects container and application context into Pods implementing aware interfaces.

ApplicationContextAwareProcessor(application_context)

Bases: IPostProcessor

Post-processor for injecting framework services into aware Pods.

Checks if Pods implement aware interfaces and injects corresponding services: - IContainerAware: Injects IoC container - IApplicationContextAware: Injects application context

Initialize aware post-processor.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context to inject.	required

Source code in core/spakky/src/spakky/core/pod/post_processors/aware_post_processor.py

def __init__(self, application_context: IApplicationContext) -> None:
    """Initialize aware post-processor.

    Args:
        application_context: The application context to inject.
    """
    self.__application_context = application_context

post_process(pod)

Inject framework services into aware Pods.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The Pod instance with injected services.

Source code in core/spakky/src/spakky/core/pod/post_processors/aware_post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Inject framework services into aware Pods.

    Args:
        pod: The Pod instance to process.

    Returns:
        The Pod instance with injected services.
    """
    if isinstance(pod, IContainerAware):
        pod.set_container(self.__application_context)
    if isinstance(pod, ITagRegistryAware):
        pod.set_tag_registry(self.__application_context)
    if isinstance(pod, IApplicationContextAware):
        pod.set_application_context(self.__application_context)
    return pod

options: show_root_heading: false

spakky.core.pod.error

Error types for Pod-related exceptions.

This module defines base error classes for Pod annotation and instantiation failures.

AbstractSpakkyPodError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for all Pod-related errors.

PodAnnotationFailedError

Bases: AbstractSpakkyPodError

Raised when Pod annotation process fails.

PodInstantiationFailedError

Bases: AbstractSpakkyPodError

Raised when Pod instantiation fails.

NegativeOrderValueError

Bases: AbstractSpakkyPodError

Raised when Order annotation receives a negative value.

QualifierSelectorNotCallableError

Bases: AbstractSpakkyPodError

Raised when Qualifier selector is not callable.

options: show_root_heading: false

Service

spakky.core.service.interfaces

options: show_root_heading: false

spakky.core.service.interfaces.service

Interfaces for application services with lifecycle management.

This module defines interfaces for services that run during application lifecycle with start and stop capabilities.

IService

Bases: ABC

Interface for synchronous services with lifecycle management.

set_stop_event(stop_event) abstractmethod

Set threading event for stop signaling.

Parameters:

Name	Type	Description	Default
stop_event	Event	Event to signal service shutdown.	required

Source code in core/spakky/src/spakky/core/service/interfaces/service.py

@abstractmethod
def set_stop_event(self, stop_event: threading.Event) -> None:
    """Set threading event for stop signaling.

    Args:
        stop_event: Event to signal service shutdown.
    """
    ...

start() abstractmethod

Start the service.

Source code in core/spakky/src/spakky/core/service/interfaces/service.py

@abstractmethod
def start(self) -> None:
    """Start the service."""
    ...

stop() abstractmethod

Stop the service and clean up resources.

Source code in core/spakky/src/spakky/core/service/interfaces/service.py

@abstractmethod
def stop(self) -> None:
    """Stop the service and clean up resources."""
    ...

IAsyncService

Bases: ABC

Interface for asynchronous services with lifecycle management.

set_stop_event(stop_event) abstractmethod

Set async event for stop signaling.

Parameters:

Name	Type	Description	Default
stop_event	Event	Async event to signal service shutdown.	required

Source code in core/spakky/src/spakky/core/service/interfaces/service.py

@abstractmethod
def set_stop_event(self, stop_event: locks.Event) -> None:
    """Set async event for stop signaling.

    Args:
        stop_event: Async event to signal service shutdown.
    """
    ...

start_async() abstractmethod async

Start the service asynchronously.

Source code in core/spakky/src/spakky/core/service/interfaces/service.py

@abstractmethod
async def start_async(self) -> None:
    """Start the service asynchronously."""
    ...

stop_async() abstractmethod async

Stop the service and clean up resources asynchronously.

Source code in core/spakky/src/spakky/core/service/interfaces/service.py

@abstractmethod
async def stop_async(self) -> None:
    """Stop the service and clean up resources asynchronously."""
    ...

options: show_root_heading: false

spakky.core.service.background

Abstract base classes for background services.

This module provides base implementations for long-running background services with proper lifecycle management.

AbstractBackgroundService

Bases: IService, ABC

Base class for synchronous background services.

Runs in a dedicated thread and can be started/stopped gracefully.

set_stop_event(stop_event)

Set stop event for shutdown signaling.

Parameters:

Name	Type	Description	Default
stop_event	Event	Event to signal service shutdown.	required

Source code in core/spakky/src/spakky/core/service/background.py

@override
def set_stop_event(self, stop_event: threading.Event) -> None:
    """Set stop event for shutdown signaling.

    Args:
        stop_event: Event to signal service shutdown.
    """
    self._stop_event = stop_event

start()

Start service in background thread.

Source code in core/spakky/src/spakky/core/service/background.py

@override
def start(self) -> None:
    """Start service in background thread."""
    self._stop_event.clear()
    self.initialize()
    self._thread = Thread(target=self.run, daemon=True, name=type(self).__name__)
    self._thread.start()

stop()

Stop service and wait for thread to finish.

Source code in core/spakky/src/spakky/core/service/background.py

@override
def stop(self) -> None:
    """Stop service and wait for thread to finish."""
    self._stop_event.set()
    if self._thread:  # pragma: no cover - coverage boundary
        self._thread.join()
    self.dispose()

initialize() abstractmethod

Initialize service before starting.

Called once before the service thread starts.

Source code in core/spakky/src/spakky/core/service/background.py

@abstractmethod
def initialize(self) -> None:
    """Initialize service before starting.

    Called once before the service thread starts.
    """
    ...

dispose() abstractmethod

Clean up resources after stopping.

Called once after the service thread stops.

Source code in core/spakky/src/spakky/core/service/background.py

@abstractmethod
def dispose(self) -> None:
    """Clean up resources after stopping.

    Called once after the service thread stops.
    """
    ...

run() abstractmethod

Main service loop.

Runs in background thread. Should check _stop_event periodically and exit when it's set.

Source code in core/spakky/src/spakky/core/service/background.py

@abstractmethod
def run(self) -> None:
    """Main service loop.

    Runs in background thread. Should check _stop_event periodically
    and exit when it's set.
    """
    ...

AbstractAsyncBackgroundService

Bases: IAsyncService, ABC

Base class for asynchronous background services.

Runs as an async task and can be started/stopped gracefully.

set_stop_event(stop_event)

Set stop event for shutdown signaling.

Parameters:

Name	Type	Description	Default
stop_event	Event	Async event to signal service shutdown.	required

Source code in core/spakky/src/spakky/core/service/background.py

@override
def set_stop_event(self, stop_event: locks.Event) -> None:
    """Set stop event for shutdown signaling.

    Args:
        stop_event: Async event to signal service shutdown.
    """
    self._stop_event = stop_event

start_async() async

Start service as background task.

Source code in core/spakky/src/spakky/core/service/background.py

@override
async def start_async(self) -> None:
    """Start service as background task."""
    self._stop_event.clear()
    await self.initialize_async()
    self._task = tasks.create_task(coro=self.run_async(), name=type(self).__name__)

stop_async() async

Stop service and wait for task to finish.

Source code in core/spakky/src/spakky/core/service/background.py

@override
async def stop_async(self) -> None:
    """Stop service and wait for task to finish."""
    self._stop_event.set()
    if self._task:  # pragma: no cover - coverage boundary
        await self._task
    await self.dispose_async()

initialize_async() abstractmethod async

Initialize service before starting.

Called once before the service task starts.

Source code in core/spakky/src/spakky/core/service/background.py

@abstractmethod
async def initialize_async(self) -> None:
    """Initialize service before starting.

    Called once before the service task starts.
    """
    ...

dispose_async() abstractmethod async

Clean up resources after stopping.

Called once after the service task stops.

Source code in core/spakky/src/spakky/core/service/background.py

@abstractmethod
async def dispose_async(self) -> None:
    """Clean up resources after stopping.

    Called once after the service task stops.
    """
    ...

run_async() abstractmethod async

Main service loop.

Runs as async task. Should check _stop_event periodically and exit when it's set.

Source code in core/spakky/src/spakky/core/service/background.py

@abstractmethod
async def run_async(self) -> None:
    """Main service loop.

    Runs as async task. Should check _stop_event periodically
    and exit when it's set.
    """
    ...

options: show_root_heading: false

spakky.core.service.post_processor

Post-processor for registering services with application context.

This module provides ServicePostProcessor which automatically registers service Pods with the application context for lifecycle management.

ServicePostProcessor(application_context)

Bases: IPostProcessor

Post-processor for registering service Pods.

Detects Pods implementing IService or IAsyncService and registers them with the application context for automatic lifecycle management.

Initialize service post-processor.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	Application context for service registration.	required

Source code in core/spakky/src/spakky/core/service/post_processor.py

def __init__(self, application_context: IApplicationContext) -> None:
    """Initialize service post-processor.

    Args:
        application_context: Application context for service registration.
    """
    super().__init__()
    self.__application_context = application_context

post_process(pod)

Register service Pods with application context.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The Pod instance unchanged.

Source code in core/spakky/src/spakky/core/service/post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Register service Pods with application context.

    Args:
        pod: The Pod instance to process.

    Returns:
        The Pod instance unchanged.
    """
    if isinstance(pod, IService):
        pod.set_stop_event(self.__application_context.thread_stop_event)
        self.__application_context.add_service(pod)
        logger.debug(
            f"[{type(self).__name__}] {type(pod).__name__!r} added to container"
        )
    if isinstance(pod, IAsyncService):
        pod.set_stop_event(self.__application_context.task_stop_event)
        self.__application_context.add_service(pod)
        logger.debug(
            f"[{type(self).__name__}] {type(pod).__name__!r} added to container"
        )
    return pod

options: show_root_heading: false

스테레오타입

spakky.core.stereotype.configuration

Configuration stereotype for organizing configuration Pods.

This module provides @Configuration stereotype for grouping related configuration and factory method Pods.

Configuration(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Stereotype for configuration classes containing factory methods.

Classes decorated with @Configuration typically contain @Pod-annotated factory methods that produce other Pods.

options: show_root_heading: false

spakky.core.stereotype.controller

Controller stereotype for grouping request handlers.

This module provides @Controller stereotype for organizing classes that handle external requests (HTTP, CLI, etc.).

Controller(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Stereotype for controller classes handling external requests.

Controllers typically contain route handlers, command handlers, or other request processing methods.

options: show_root_heading: false

spakky.core.stereotype.usecase

UseCase stereotype for encapsulating business logic.

This module provides @UseCase stereotype for organizing classes that implement application-specific business rules.

UseCase(*, name='', scope=Scope.SINGLETON) dataclass

Bases: Pod

Stereotype for use case classes encapsulating business logic.

UseCases represent application-specific business operations, orchestrating domain entities and services to fulfill requirements.

options: show_root_heading: false

공통

spakky.core.common.annotation

Annotation() dataclass

Base class for type-safely injecting metadata(annotation as said) into objects.

all(obj) classmethod

Get all list of annotations from the object.

Parameters:

Name	Type	Description	Default
obj	object	The object to get the annotations from.	required

Returns:

Type	Description
list[Self]	list[Self]: List of annotations.

Source code in core/spakky/src/spakky/core/common/annotation.py

@final
@classmethod
def all(cls, obj: object) -> list[Self]:
    """Get all list of annotations from the object.

    Args:
        obj: The object to get the annotations from.

    Returns:
        list[Self]: List of annotations.
    """
    metadata: dict[type, list[Self]] = cls.__get_metadata(obj)
    annotations: list[Self] = metadata.get(cls, [])
    return annotations

get(obj) classmethod

Get a single annotation from the object.

Parameters:

Name	Type	Description	Default
obj	object	The object to get the annotation from.	required

Returns:

Name	Type	Description
Self	Self	The annotation.

Raises:

Type	Description
AnnotationNotFoundError	If no annotation is found.
MultipleAnnotationFoundError	If multiple annotations are found.

Source code in core/spakky/src/spakky/core/common/annotation.py

@final
@classmethod
def get(cls, obj: object) -> Self:
    """Get a single annotation from the object.

    Args:
        obj: The object to get the annotation from.

    Returns:
        Self: The annotation.

    Raises:
        AnnotationNotFoundError: If no annotation is found.
        MultipleAnnotationFoundError: If multiple annotations are found.
    """
    metadata: dict[type, list[Self]] = cls.__get_metadata(obj)
    if cls not in metadata:
        raise AnnotationNotFoundError(cls, obj)
    annotations: list[Self] = metadata.get(cls, [])
    if len(annotations) > 1:
        raise MultipleAnnotationFoundError(cls, obj)
    return annotations[0]

get_or_none(obj) classmethod

Get a single annotation from the object or None if not found.

Parameters:

Name	Type	Description	Default
obj	object	The object to get the annotation from.	required

Raises:

Type	Description
MultipleAnnotationFoundError	If multiple annotations are found.

Returns:

Type	Description
Self | None	Self | None: The annotation or None if not found.

Source code in core/spakky/src/spakky/core/common/annotation.py

@final
@classmethod
def get_or_none(cls, obj: object) -> Self | None:
    """Get a single annotation from the object or None if not found.

    Args:
        obj: The object to get the annotation from.

    Raises:
        MultipleAnnotationFoundError: If multiple annotations are found.

    Returns:
        Self | None: The annotation or None if not found.
    """
    metadata: dict[type, list[Self]] = cls.__get_metadata(obj)
    if cls not in metadata:
        return None
    annotations: list[Self] = metadata.get(cls, [])
    if len(annotations) > 1:
        raise MultipleAnnotationFoundError(cls, obj)
    return annotations[0]

get_or_default(obj, default) classmethod

Get a single annotation from the object or a default value if not found.

Parameters:

Name	Type	Description	Default
obj	object	The object to get the annotation from.	required
default	Self	The default value to return if not found.	required

Raises:

Type	Description
MultipleAnnotationFoundError	If multiple annotations are found.

Returns:

Name	Type	Description
Self	Self	The annotation or the default value if not found.

Source code in core/spakky/src/spakky/core/common/annotation.py

@final
@classmethod
def get_or_default(cls, obj: object, default: Self) -> Self:
    """Get a single annotation from the object or a default value if not found.

    Args:
        obj: The object to get the annotation from.
        default: The default value to return if not found.

    Raises:
        MultipleAnnotationFoundError: If multiple annotations are found.

    Returns:
        Self: The annotation or the default value if not found.
    """
    metadata: dict[type, list[Self]] = cls.__get_metadata(obj)
    if cls not in metadata:
        return default
    annotations: list[Self] = metadata.get(cls, [])
    if len(annotations) > 1:
        raise MultipleAnnotationFoundError(cls, obj)
    return annotations[0]

exists(obj) classmethod

Check if the annotation exists in the object.

Parameters:

Name	Type	Description	Default
obj	object	The object to check.	required

Returns:

Name	Type	Description
bool	bool	True if the annotation exists, False otherwise.

Source code in core/spakky/src/spakky/core/common/annotation.py

@final
@classmethod
def exists(cls, obj: object) -> bool:
    """Check if the annotation exists in the object.

    Args:
        obj: The object to check.

    Returns:
        bool: True if the annotation exists, False otherwise.
    """
    metadata: dict[type, list[Self]] = cls.__get_metadata(obj)
    return cls in metadata

ClassAnnotation() dataclass

Bases: Annotation

Annotation for classes.

__call__(obj)

Call method to annotate a class.

Parameters:

Name	Type	Description	Default
obj	type[T]	The class to annotate.	required

Returns:

Type	Description
type[T]	The annotated class.

Source code in core/spakky/src/spakky/core/common/annotation.py

def __call__[T: object](self, obj: type[T]) -> type[T]:
    """Call method to annotate a class.

    Args:
        obj: The class to annotate.

    Returns:
        The annotated class.
    """
    return super().__call__(obj)

FunctionAnnotation() dataclass

Bases: Annotation

Annotation for functions.

__call__(obj)

Call method to annotate a function.

Parameters:

Name	Type	Description	Default
obj	Callable[..., T]	The function to annotate.	required

Returns:

Type	Description
Callable[..., T]	The annotated function.

Source code in core/spakky/src/spakky/core/common/annotation.py

def __call__[T](self, obj: Callable[..., T]) -> Callable[..., T]:
    """Call method to annotate a function.

    Args:
        obj: The function to annotate.

    Returns:
        The annotated function.
    """
    return super().__call__(obj)

AnnotationNotFoundError

Bases: AbstractSpakkyFrameworkError

Exception raised when no annotation is found in an object.

MultipleAnnotationFoundError

Bases: AbstractSpakkyFrameworkError

Exception raised when multiple annotations are found in an object.

options: show_root_heading: false

spakky.core.common.constants

Core constants for Spakky framework.

options: show_root_heading: false

spakky.core.common.error

AbstractSpakkyFrameworkError

Bases: Exception, ABC

Base class for all Spakky framework errors.

The error message can be defined in two ways: 1. Class-level: Define message as a class attribute for a fixed message. 2. Instance-level: Pass message to the constructor to override the class default.

If neither is provided, the message will be an empty string.

message class-attribute

A human-readable message describing the error.

GenericMROTypeError

Bases: AbstractSpakkyFrameworkError

Raised when a non-type, non-generic-alias is passed to generic_mro.

options: show_root_heading: false

spakky.core.common.interfaces

options: show_root_heading: false

spakky.core.common.interfaces.cloneable

ICloneable

Bases: ABC

Interface for cloneable objects.

clone() abstractmethod

Creates a clone of the current object.

Returns:

Name	Type	Description
Self	Self	A new instance that is a clone of the current object.

Source code in core/spakky/src/spakky/core/common/interfaces/cloneable.py

@abstractmethod
def clone(self) -> Self:
    """Creates a clone of the current object.

    Returns:
        Self: A new instance that is a clone of the current object.
    """

options: show_root_heading: false

spakky.core.common.interfaces.comparable

IComparable

Bases: ABC

Interface for comparable objects.

__lt__(__value) abstractmethod

Less than comparison.

Parameters:

Name	Type	Description	Default
__value	Self	The value to compare against.	required

Returns:

Name	Type	Description
bool	bool	True if self is less than __value, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/comparable.py

@abstractmethod
def __lt__(self, __value: Self) -> bool:
    """Less than comparison.

    Args:
        __value (Self): The value to compare against.

    Returns:
        bool: True if self is less than __value, False otherwise.
    """

__le__(__value) abstractmethod

Less than or equal comparison.

Parameters:

Name	Type	Description	Default
__value	Self	The value to compare against.	required

Returns: bool: True if self is less than or equal to __value, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/comparable.py

@abstractmethod
def __le__(self, __value: Self) -> bool:
    """Less than or equal comparison.

    Args:
        __value (Self): The value to compare against.
    Returns:
        bool: True if self is less than or equal to __value, False otherwise.
    """

__gt__(__value) abstractmethod

Greater than comparison.

Parameters:

Name	Type	Description	Default
__value	Self	The value to compare against.	required

Returns:

Name	Type	Description
bool	bool	True if self is greater than __value, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/comparable.py

@abstractmethod
def __gt__(self, __value: Self) -> bool:
    """Greater than comparison.

    Args:
        __value (Self): The value to compare against.

    Returns:
        bool: True if self is greater than __value, False otherwise.
    """

__ge__(__value) abstractmethod

Greater than or equal comparison.

Parameters:

Name	Type	Description	Default
__value	Self	The value to compare against.	required

Returns: bool: True if self is greater than or equal to __value, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/comparable.py

@abstractmethod
def __ge__(self, __value: Self) -> bool:
    """Greater than or equal comparison.

    Args:
        __value (Self): The value to compare against.
    Returns:
        bool: True if self is greater than or equal to __value, False otherwise.
    """

options: show_root_heading: false

spakky.core.common.interfaces.disposable

IDisposable

Bases: ABC

Interface for disposable objects.

__enter__() abstractmethod

Enters the runtime context related to this object.

Returns:

Name	Type	Description
Self	Self	The object itself.

Source code in core/spakky/src/spakky/core/common/interfaces/disposable.py

@abstractmethod
def __enter__(self) -> Self:
    """Enters the runtime context related to this object.

    Returns:
        Self: The object itself.
    """

__exit__(__exc_type, __exc_value, __traceback) abstractmethod

Exits the runtime context related to this object.

Parameters:

Name	Type	Description	Default
__exc_type	type[BaseException] | None	The exception type.	required
__exc_value	BaseException | None	The exception value.	required
__traceback	TracebackType | None	The traceback object.	required

Returns:

Type	Description
bool | None	bool | None: True if the exception was handled, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/disposable.py

@abstractmethod
def __exit__(
    self,
    __exc_type: type[BaseException] | None,
    __exc_value: BaseException | None,
    __traceback: TracebackType | None,
) -> bool | None:
    """Exits the runtime context related to this object.

    Args:
        __exc_type (type[BaseException] | None): The exception type.
        __exc_value (BaseException | None): The exception value.
        __traceback (TracebackType | None): The traceback object.

    Returns:
        bool | None: True if the exception was handled, False otherwise.
    """

IAsyncDisposable

Bases: ABC

Interface for asynchronously disposable objects.

__aenter__() abstractmethod async

Asynchronously enters the runtime context related to this object.

Returns:

Name	Type	Description
Self	Self	The object itself.

Source code in core/spakky/src/spakky/core/common/interfaces/disposable.py

@abstractmethod
async def __aenter__(self) -> Self:
    """Asynchronously enters the runtime context related to this object.

    Returns:
        Self: The object itself.
    """

__aexit__(__exc_type, __exc_value, __traceback) abstractmethod async

Asynchronously exits the runtime context related to this object.

Parameters:

Name	Type	Description	Default
__exc_type	type[BaseException] | None	The exception type.	required
__exc_value	BaseException | None	The exception value.	required
__traceback	TracebackType | None	The traceback object.	required

Returns:

Type	Description
bool | None	bool | None: True if the exception was handled, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/disposable.py

@abstractmethod
async def __aexit__(
    self,
    __exc_type: type[BaseException] | None,
    __exc_value: BaseException | None,
    __traceback: TracebackType | None,
) -> bool | None:
    """Asynchronously exits the runtime context related to this object.

    Args:
        __exc_type (type[BaseException] | None): The exception type.
        __exc_value (BaseException | None): The exception value.
        __traceback (TracebackType | None): The traceback object.

    Returns:
        bool | None: True if the exception was handled, False otherwise.
    """

options: show_root_heading: false

spakky.core.common.interfaces.equatable

IEquatable

Bases: ABC

Interface for equatable objects.

__eq__(__value) abstractmethod

Checks equality with another object.

Parameters:

Name	Type	Description	Default
__value	object	The object to compare with.	required

Returns:

Name	Type	Description
bool	bool	True if equal, False otherwise.

Source code in core/spakky/src/spakky/core/common/interfaces/equatable.py

@abstractmethod
def __eq__(self, __value: object) -> bool:
    """Checks equality with another object.

    Args:
        __value (object): The object to compare with.

    Returns:
        bool: True if equal, False otherwise.
    """

__hash__() abstractmethod

Returns the hash of the object.

Returns:

Name	Type	Description
int	int	The hash value.

Source code in core/spakky/src/spakky/core/common/interfaces/equatable.py

@abstractmethod
def __hash__(self) -> int:
    """Returns the hash of the object.

    Returns:
        int: The hash value.
    """

options: show_root_heading: false

spakky.core.common.interfaces.representable

IRepresentable

Bases: ABC

Interface for representable objects.

__str__() abstractmethod

Returns the string representation of the object.

Returns:

Name	Type	Description
str	str	The string representation.

Source code in core/spakky/src/spakky/core/common/interfaces/representable.py

@abstractmethod
def __str__(self) -> str:
    """Returns the string representation of the object.

    Returns:
        str: The string representation.
    """

__repr__() abstractmethod

Returns the official string representation of the object.

Returns:

Name	Type	Description
str	str	The official string representation.

Source code in core/spakky/src/spakky/core/common/interfaces/representable.py

@abstractmethod
def __repr__(self) -> str:
    """Returns the official string representation of the object.

    Returns:
        str: The official string representation.
    """

options: show_root_heading: false

spakky.core.common.metadata

Metadata extraction utilities for Annotated types.

This module provides utilities for extracting metadata from Python's Annotated type hints.

MetadataNotFoundError

Bases: AbstractSpakkyFrameworkError

Raised when expected metadata is not found in an Annotated type.

InvalidAnnotatedTypeError

Bases: AbstractSpakkyFrameworkError

Raised when an invalid Annotated type is provided.

AbstractMetadata() dataclass

Bases: ABC

Abstract base class for type metadata.

get_actual_type(annotated) classmethod

Get the actual Python type from the Annotated type.

Parameters:

Name	Type	Description	Default
annotated	AnnotatedType	The Annotated type to extract the actual type from.	required

Returns:

Type	Description
type[Any]	type[Any]: The actual Python type.

Source code in core/spakky/src/spakky/core/common/metadata.py

@classmethod
def get_actual_type(cls, annotated: AnnotatedType) -> type[Any]:
    """Get the actual Python type from the Annotated type.

    Args:
        annotated (AnnotatedType): The Annotated type to extract the actual type from.

    Returns:
        type[Any]: The actual Python type.
    """
    if not cls._validate_annotated(annotated):
        raise InvalidAnnotatedTypeError
    return annotated.__origin__

all(annotated) classmethod

Get all metadata of this type from the Annotated type.

Parameters:

Name	Type	Description	Default
annotated	AnnotatedType	The Annotated type to extract metadata from.	required

Returns:

Type	Description
list[Self]	list[Self]: List of metadata instances of this type.

Source code in core/spakky/src/spakky/core/common/metadata.py

@classmethod
def all(cls, annotated: AnnotatedType) -> list[Self]:
    """Get all metadata of this type from the Annotated type.

    Args:
        annotated (AnnotatedType): The Annotated type to extract metadata from.

    Returns:
        list[Self]: List of metadata instances of this type.
    """
    if not cls._validate_annotated(annotated):
        return []
    metadatas = get_args(annotated)
    return [data for data in metadatas if isinstance(data, cls)]

get(annotated) classmethod

Get a single metadata of this type from the Annotated type.

Parameters:

Name	Type	Description	Default
annotated	AnnotatedType	The Annotated type to extract metadata from.	required

Returns: Self: The metadata instance of this type.

Source code in core/spakky/src/spakky/core/common/metadata.py

@classmethod
def get(cls, annotated: AnnotatedType) -> Self:
    """Get a single metadata of this type from the Annotated type.

    Args:
        annotated (AnnotatedType): The Annotated type to extract metadata from.
    Returns:
        Self: The metadata instance of this type.
    """
    if not cls._validate_annotated(annotated):
        raise InvalidAnnotatedTypeError
    metadatas = get_args(annotated)
    found = next(iter(data for data in metadatas if isinstance(data, cls)), None)
    if found is None:
        raise MetadataNotFoundError
    return found

get_or_none(annotated) classmethod

Get a single metadata of this type from the Annotated type, or None if not found.

Parameters:

Name	Type	Description	Default
annotated	AnnotatedType	The Annotated type to extract metadata from.	required

Returns:

Type	Description
Self | None	Self | None: The metadata instance of this type, or None if not found.

Source code in core/spakky/src/spakky/core/common/metadata.py

@classmethod
def get_or_none(cls, annotated: AnnotatedType) -> Self | None:
    """Get a single metadata of this type from the Annotated type, or None if not found.

    Args:
        annotated (AnnotatedType): The Annotated type to extract metadata from.

    Returns:
        Self | None: The metadata instance of this type, or None if not found.
    """
    if not cls._validate_annotated(annotated):
        return None
    metadatas = get_args(annotated)
    return next(iter(data for data in metadatas if isinstance(data, cls)), None)

get_or_default(annotated, default) classmethod

Get a single metadata of this type from the Annotated type, or a default if not found.

Parameters:

Name	Type	Description	Default
annotated	AnnotatedType	The Annotated type to extract metadata from.	required
default	Self	The default metadata to return if not found.	required

Returns:

Name	Type	Description
Self	Self	The metadata instance of this type, or the default if not found.

Source code in core/spakky/src/spakky/core/common/metadata.py

@classmethod
def get_or_default(cls, annotated: AnnotatedType, default: Self) -> Self:
    """Get a single metadata of this type from the Annotated type, or a default if not found.

    Args:
        annotated (AnnotatedType): The Annotated type to extract metadata from.
        default (Self): The default metadata to return if not found.

    Returns:
        Self: The metadata instance of this type, or the default if not found.
    """
    if not cls._validate_annotated(annotated):
        return default
    metadatas = get_args(annotated)
    return next(
        iter(data for data in metadatas if isinstance(data, cls)),
        default,
    )

exists(annotated) classmethod

Check if metadata of this type exists in the Annotated type.

Parameters:

Name	Type	Description	Default
annotated	AnnotatedType	The Annotated type to check.	required

Returns:

Name	Type	Description
bool	bool	True if metadata of this type exists, False otherwise.

Source code in core/spakky/src/spakky/core/common/metadata.py

@classmethod
def exists(cls, annotated: AnnotatedType) -> bool:
    """Check if metadata of this type exists in the Annotated type.

    Args:
        annotated (AnnotatedType): The Annotated type to check.

    Returns:
        bool: True if metadata of this type exists, False otherwise.
    """
    if not cls._validate_annotated(annotated):
        return False
    metadatas = get_args(annotated)
    return any(isinstance(data, cls) for data in metadatas)

options: show_root_heading: false

spakky.core.common.mro

Method Resolution Order (MRO) utilities for generic types.

This module provides utilities for computing MRO (Method Resolution Order) for generic types, including parameterized generics like list[int] or dict[str, Any].

generic_mro(tp)

Compute the Method Resolution Order for a generic type.

Supports both regular classes and parameterized generic types (e.g., list[int]).

Parameters:

Name	Type	Description	Default
tp	Any	The type or generic alias to compute MRO for.	required

Returns:

Type	Description
list[Any]	list[type]: The method resolution order as a list of types.

Raises:

Type	Description
TypeError	If tp is not a type or generic alias.

Source code in core/spakky/src/spakky/core/common/mro.py

def generic_mro(tp: Any) -> list[Any]:
    """Compute the Method Resolution Order for a generic type.

    Supports both regular classes and parameterized generic types (e.g., list[int]).

    Args:
        tp: The type or generic alias to compute MRO for.

    Returns:
        list[type]: The method resolution order as a list of types.

    Raises:
        TypeError: If tp is not a type or generic alias.
    """
    origin = get_origin(tp)
    if (
        origin is None
        and not hasattr(tp, ORIGIN_BASES)  # typing internals generic alias check
    ):
        if not isinstance(tp, type):
            raise GenericMROTypeError
        return tp.mro()
    # Sentinel value to avoid subscripting Generic while resolving generic bases.
    result = {Generic: Generic}
    _generic_mro(result, tp)  # type: ignore[arg-type]  # tp is validated above
    cls = origin if origin is not None else tp
    return list(result.get(sub_cls, sub_cls) for sub_cls in cls.__mro__)

is_family_with(tp, target)

Check if a type is related to a target type through its MRO.

Parameters:

Name	Type	Description	Default
tp	Any	The type to check.	required
target	type[T]	The target type to look for in the MRO.	required

Returns:

Type	Description
TypeGuard[type[T]]	True if target is in tp's MRO, False otherwise.

Source code in core/spakky/src/spakky/core/common/mro.py

def is_family_with[T: object](tp: Any, target: type[T]) -> TypeGuard[type[T]]:
    """Check if a type is related to a target type through its MRO.

    Args:
        tp: The type to check.
        target: The target type to look for in the MRO.

    Returns:
        True if target is in tp's MRO, False otherwise.
    """
    return target in generic_mro(tp)

options: show_root_heading: false

spakky.core.common.mutability

Utilities for creating mutable and immutable dataclasses.

This module provides decorators that create dataclasses with predefined mutability settings.

mutable(cls)

Decorator to create a mutable dataclass with keyword-only arguments.

Parameters:

Name	Type	Description	Default
cls	type[T]	The class to transform into a mutable dataclass.	required

Returns:

Type	Description
type[T]	A mutable dataclass with frozen=False, kw_only=True, eq=False.

Source code in core/spakky/src/spakky/core/common/mutability.py

@dataclass_transform(
    eq_default=False,
    kw_only_default=True,
    frozen_default=False,
    field_specifiers=(field,),
)
def mutable[T](cls: type[T]) -> type[T]:
    """Decorator to create a mutable dataclass with keyword-only arguments.

    Args:
        cls: The class to transform into a mutable dataclass.

    Returns:
        A mutable dataclass with frozen=False, kw_only=True, eq=False.
    """
    return dataclass(frozen=False, kw_only=True, eq=False)(cls)

immutable(cls)

Decorator to create an immutable dataclass with keyword-only arguments.

Parameters:

Name	Type	Description	Default
cls	type[T]	The class to transform into an immutable dataclass.	required

Returns:

Type	Description
type[T]	An immutable dataclass with frozen=True, kw_only=True, eq=False.

Source code in core/spakky/src/spakky/core/common/mutability.py

@dataclass_transform(
    eq_default=False,
    kw_only_default=True,
    frozen_default=True,
    field_specifiers=(field,),
)
def immutable[T](cls: type[T]) -> type[T]:
    """Decorator to create an immutable dataclass with keyword-only arguments.

    Args:
        cls: The class to transform into an immutable dataclass.

    Returns:
        An immutable dataclass with frozen=True, kw_only=True, eq=False.
    """
    return dataclass(frozen=True, kw_only=True, eq=False)(cls)

options: show_root_heading: false

spakky.core.common.proxy

Dynamic proxy implementation for intercepting object method calls and attribute access.

This module provides a proxy pattern implementation that allows intercepting and modifying method calls and attribute access on target objects.

IProxyHandler

Bases: ABC

Interface for proxy handlers that intercept object operations.

call(target, method, *args, **kwargs) abstractmethod

Intercept a synchronous method call.

Parameters:

Name	Type	Description	Default
target	object	The target object.	required
method	Func	The method being called.	required
*args	Any	Positional arguments for the method.	()
**kwargs	Any	Keyword arguments for the method.	{}

Returns:

Name	Type	Description
Any	Any	The result of the method call.

Source code in core/spakky/src/spakky/core/common/proxy.py

@abstractmethod
def call(
    self,
    target: object,
    method: Func,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Intercept a synchronous method call.

    Args:
        target: The target object.
        method: The method being called.
        *args: Positional arguments for the method.
        **kwargs: Keyword arguments for the method.

    Returns:
        Any: The result of the method call.
    """
    ...

call_async(target, method, *args, **kwargs) abstractmethod async

Intercept an asynchronous method call.

Parameters:

Name	Type	Description	Default
target	object	The target object.	required
method	AsyncFunc	The async method being called.	required
*args	Any	Positional arguments for the method.	()
**kwargs	Any	Keyword arguments for the method.	{}

Returns:

Name	Type	Description
Any	Any	The result of the async method call.

Source code in core/spakky/src/spakky/core/common/proxy.py

@abstractmethod
async def call_async(
    self,
    target: object,
    method: AsyncFunc,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Intercept an asynchronous method call.

    Args:
        target: The target object.
        method: The async method being called.
        *args: Positional arguments for the method.
        **kwargs: Keyword arguments for the method.

    Returns:
        Any: The result of the async method call.
    """
    ...

get(target, name) abstractmethod

Intercept attribute access.

Parameters:

Name	Type	Description	Default
target	object	The target object.	required
name	str	The attribute name being accessed.	required

Returns:

Name	Type	Description
Any	Any	The attribute value.

Source code in core/spakky/src/spakky/core/common/proxy.py

@abstractmethod
def get(self, target: object, name: str) -> Any:
    """Intercept attribute access.

    Args:
        target: The target object.
        name: The attribute name being accessed.

    Returns:
        Any: The attribute value.
    """
    ...

set(target, name, value) abstractmethod

Intercept attribute assignment.

Parameters:

Name	Type	Description	Default
target	object	The target object.	required
name	str	The attribute name being set.	required
value	Any	The value being assigned.	required

Source code in core/spakky/src/spakky/core/common/proxy.py

@abstractmethod
def set(self, target: object, name: str, value: Any) -> None:
    """Intercept attribute assignment.

    Args:
        target: The target object.
        name: The attribute name being set.
        value: The value being assigned.
    """
    ...

delete(target, name) abstractmethod

Intercept attribute deletion.

Parameters:

Name	Type	Description	Default
target	object	The target object.	required
name	str	The attribute name being deleted.	required

Source code in core/spakky/src/spakky/core/common/proxy.py

@abstractmethod
def delete(self, target: object, name: str) -> None:
    """Intercept attribute deletion.

    Args:
        target: The target object.
        name: The attribute name being deleted.
    """
    ...

AbstractProxyHandler

Bases: IProxyHandler, ABC

Abstract base class for proxy handlers with default pass-through implementations.

call(target, method, *args, **kwargs)

Default implementation that directly calls the method.

Parameters:

Name	Type	Description	Default
target	object	The target object.	required
method	Func	The method being called.	required
*args	Any	Positional arguments for the method.	()
**kwargs	Any	Keyword arguments for the method.	{}

Returns:

Name	Type	Description
Any	Any	The result of the method call.

Source code in core/spakky/src/spakky/core/common/proxy.py

def call(
    self,
    target: object,
    method: Func,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Default implementation that directly calls the method.

    Args:
        target: The target object.
        method: The method being called.
        *args: Positional arguments for the method.
        **kwargs: Keyword arguments for the method.

    Returns:
        Any: The result of the method call.
    """
    return method(*args, **kwargs)

ProxyFactory(target, handler)

Factory for creating dynamic proxy objects.

Creates a proxy that intercepts method calls and attribute access on a target object, delegating the interception logic to a handler.

Initialize the proxy factory.

Parameters:

Name	Type	Description	Default
target	T	The target object to proxy.	required
handler	IProxyHandler	The handler that implements interception logic.	required

Source code in core/spakky/src/spakky/core/common/proxy.py

def __init__(
    self,
    target: T,
    handler: IProxyHandler,
) -> None:
    """Initialize the proxy factory.

    Args:
        target: The target object to proxy.
        handler: The handler that implements interception logic.
    """
    self._type = type(target)
    self._target = target
    self._handler = handler
    self._wrapper_cache = {}

ATTRIBUTES_TO_IGNORE = frozenset(['__dict__', '__class__', '__weakref__', '__base__', '__bases__', '__mro__', '__subclasses__', '__name__', '__qualname__', '__module__', '__annotations__', '__doc__']) class-attribute

Class-level attributes that should not be proxied.

create()

Create a proxy instance for the target object.

Returns:

Type	Description
T	A proxy instance that wraps the target object.

Source code in core/spakky/src/spakky/core/common/proxy.py

def create(self) -> T:
    """Create a proxy instance for the target object.

    Returns:
        A proxy instance that wraps the target object.
    """
    return new_class(
        name=self._type.__name__ + DYNAMIC_PROXY_CLASS_NAME_SUFFIX,
        bases=(self._type,),
        exec_body=lambda ns: ns.update(
            __getattribute__=self.__proxy_getattribute__,
            __setattr__=self.__proxy_setattr__,
            __delattr__=self.__proxy_delattr__,
            __dir__=self.__proxy_dir__,
            __init__=self.__proxy_init__,
        ),
    )()

options: show_root_heading: false

spakky.core.common.types

Type aliases and utility functions for the Spakky framework.

This module provides common type aliases and utility functions for working with types, particularly for Optional/Union type handling.

is_optional(type_)

Check if a type is Optional (Union with None).

Parameters:

Name	Type	Description	Default
type_	Any	The type to check.	required

Returns:

Name	Type	Description
bool	bool	True if the type is Optional[T] or Union[T, None], False otherwise.

Source code in core/spakky/src/spakky/core/common/types.py

def is_optional(type_: Any) -> bool:
    """Check if a type is Optional (Union with None).

    Args:
        type_: The type to check.

    Returns:
        bool: True if the type is Optional[T] or Union[T, None], False otherwise.
    """
    is_union_type: bool = get_origin(type_) in (UnionType, Union)
    includes_none: bool = type(None) in get_args(type_)
    is_union_with_none: bool = is_union_type and includes_none
    return is_union_with_none

remove_none(type_)

Remove None from a Union type.

If the type is Union[T, None], returns T. If the type is Union[T1, T2, ..., None], returns Union[T1, T2, ...]. If the type is not a Union, returns it unchanged.

Parameters:

Name	Type	Description	Default
type_	Any	The type to process.	required

Returns:

Name	Type	Description
Any	Any	The type with None removed from Union types.

Source code in core/spakky/src/spakky/core/common/types.py

def remove_none(type_: Any) -> Any:
    """Remove None from a Union type.

    If the type is Union[T, None], returns T.
    If the type is Union[T1, T2, ..., None], returns Union[T1, T2, ...].
    If the type is not a Union, returns it unchanged.

    Args:
        type_: The type to process.

    Returns:
        Any: The type with None removed from Union types.
    """
    origin = get_origin(type_)
    if origin in (UnionType, Union):
        args = get_args(type_)
        non_none_args = tuple(a for a in args if a is not type(None))
        if not non_none_args:  # pragma: no cover - coverage boundary
            return type(None)
        if len(non_none_args) == 1:
            return non_none_args[0]
        return reduce(or_, non_none_args)
    return type_

get_callable_methods(obj)

Get callable members excluding properties.

Uses inspect.getmembers_static() to avoid invoking descriptors during introspection, then retrieves bound methods only for non-property callables.

Parameters:

Name	Type	Description	Default
obj	object	The object to inspect.	required

Returns:

Type	Description
list[tuple[str, Callable[..., Any]]]	List of (name, bound_method) tuples for callable non-property members.

Source code in core/spakky/src/spakky/core/common/types.py

def get_callable_methods(obj: object) -> list[tuple[str, Callable[..., Any]]]:
    """Get callable members excluding properties.

    Uses inspect.getmembers_static() to avoid invoking descriptors during
    introspection, then retrieves bound methods only for non-property callables.

    Args:
        obj: The object to inspect.

    Returns:
        List of (name, bound_method) tuples for callable non-property members.
    """
    result: list[tuple[str, Callable[..., object]]] = []

    for name, value in getmembers_static(obj):
        if isinstance(value, property):
            continue

        try:
            actual = getattr(
                obj, name
            )  # introspection skips properties raising on access
        except Exception:  # noqa: BLE001
            continue

        if callable(actual):
            result.append((name, actual))

    return result

options: show_root_heading: false

로깅 인터페이스

spakky.core.logging.interfaces.log_context_binder

ILogContextBinder — log context binding interface.

ILogContextBinder

Bases: ABC

Interface for binding contextual key-value pairs to log records.

Implementations manage a context store (e.g., contextvars) that structured logging filters can read from. Plugins that need to enrich log records (such as spakky-opentelemetry) depend on this interface instead of importing a concrete LogContext class directly, keeping plugin-to-plugin dependencies out of the graph.

bind(**kwargs) abstractmethod

Add key-value pairs to the current log context.

Parameters:

Name	Type	Description	Default
**kwargs	str	Key-value pairs to bind.	{}

Source code in core/spakky/src/spakky/core/logging/interfaces/log_context_binder.py

@abstractmethod
def bind(self, **kwargs: str) -> None:
    """Add key-value pairs to the current log context.

    Args:
        **kwargs: Key-value pairs to bind.
    """
    ...

unbind(*keys) abstractmethod

Remove keys from the current log context.

Parameters:

Name	Type	Description	Default
*keys	str	Keys to remove.	()

Source code in core/spakky/src/spakky/core/logging/interfaces/log_context_binder.py

@abstractmethod
def unbind(self, *keys: str) -> None:
    """Remove keys from the current log context.

    Args:
        *keys: Keys to remove.
    """
    ...

options: show_root_heading: false

유틸리티

spakky.core.utils.casing

String case conversion utilities.

This module provides functions for converting between PascalCase and snake_case.

pascal_to_snake(pascal_case)

Convert PascalCase string to snake_case.

Parameters:

Name	Type	Description	Default
pascal_case	str	String in PascalCase format.	required

Returns:

Type	Description
str	String converted to snake_case.

Example

pascal_to_snake("UserService") 'user_service'

Source code in core/spakky/src/spakky/core/utils/casing.py

def pascal_to_snake(pascal_case: str) -> str:
    """Convert PascalCase string to snake_case.

    Args:
        pascal_case: String in PascalCase format.

    Returns:
        String converted to snake_case.

    Example:
        >>> pascal_to_snake("UserService")
        'user_service'
    """
    return PATTERN.sub("_", pascal_case).lower()

snake_to_pascal(snake_case)

Convert snake_case string to PascalCase.

Parameters:

Name	Type	Description	Default
snake_case	str	String in snake_case format.	required

Returns:

Type	Description
str	String converted to PascalCase.

Example

snake_to_pascal("user_service") 'UserService'

Source code in core/spakky/src/spakky/core/utils/casing.py

def snake_to_pascal(snake_case: str) -> str:
    """Convert snake_case string to PascalCase.

    Args:
        snake_case: String in snake_case format.

    Returns:
        String converted to PascalCase.

    Example:
        >>> snake_to_pascal("user_service")
        'UserService'
    """
    return "".join(word.title() for word in snake_case.split("_"))

options: show_root_heading: false

spakky.core.utils.inspection

Function and class inspection utilities.

This module provides utilities for introspecting functions and classes to determine their characteristics.

is_instance_method(obj)

Check if a function is an instance method.

Parameters:

Name	Type	Description	Default
obj	Func	The function to check.	required

Returns:

Type	Description
bool	True if the function is an instance method (has 'self' as first parameter).

Source code in core/spakky/src/spakky/core/utils/inspection.py

def is_instance_method(obj: Func) -> bool:
    """Check if a function is an instance method.

    Args:
        obj: The function to check.

    Returns:
        True if the function is an instance method (has 'self' as first parameter).
    """
    if ismethod(obj):
        return True
    spec: FullArgSpec = getfullargspec(obj)
    if len(spec.args) > 0 and spec.args[0] == SELF:
        return True
    return False

has_default_constructor(cls)

Check if a class has a default (no-argument) constructor.

Parameters:

Name	Type	Description	Default
cls	type[object]	The class to check.	required

Returns:

Type	Description
bool	True if the class uses the default object.init or protocol placeholder.

Source code in core/spakky/src/spakky/core/utils/inspection.py

def has_default_constructor(cls: type[object]) -> bool:
    """Check if a class has a default (no-argument) constructor.

    Args:
        cls: The class to check.

    Returns:
        True if the class uses the default object.__init__ or protocol placeholder.
    """
    constructor: Action = getattr(cls, INIT)  # constructor introspection by magic name
    if constructor is object.__init__ or constructor.__name__ == PROTOCOL_INIT:
        # If the constructor is the default constructor
        # or a placeholder for the default constructor
        return True
    return False

get_fully_qualified_name(obj)

Return the fully qualified name (FQN) of an object.

Resolution rules

1. If obj is a class, return obj.__module__ + "." + obj.__qualname__.
2. If obj is a function or method, return its own FQN.
3. Otherwise, treat obj as an instance and return the FQN of type(obj).

Parameters:

Name	Type	Description	Default
obj	object	Class, function, method, or arbitrary instance.	required

Returns:

Type	Description
str	A dotted fully qualified name, for example:
str	module.path.ClassName or module.path.ClassName.method_name.

Example

class Foo: ... def bar(self) -> None: pass get_fully_qualified_name(Foo) 'main.Foo' get_fully_qualified_name(Foo.bar) 'main.Foo.bar' get_fully_qualified_name(Foo()) 'main.Foo'

Source code in core/spakky/src/spakky/core/utils/inspection.py

def get_fully_qualified_name(obj: object) -> str:
    """Return the fully qualified name (FQN) of an object.

    Resolution rules:
        1. If ``obj`` is a class, return ``obj.__module__ + "." + obj.__qualname__``.
        2. If ``obj`` is a function or method, return its own FQN.
        3. Otherwise, treat ``obj`` as an instance and return the FQN of ``type(obj)``.

    Args:
        obj: Class, function, method, or arbitrary instance.

    Returns:
        A dotted fully qualified name, for example:
        ``module.path.ClassName`` or ``module.path.ClassName.method_name``.

    Example:
        >>> class Foo:
        ...     def bar(self) -> None: pass
        >>> get_fully_qualified_name(Foo)
        '__main__.Foo'
        >>> get_fully_qualified_name(Foo.bar)
        '__main__.Foo.bar'
        >>> get_fully_qualified_name(Foo())
        '__main__.Foo'
    """

    if isinstance(obj, type):
        return f"{obj.__module__}.{obj.__qualname__}"

    if isinstance(obj, (FunctionType, MethodType, BuiltinFunctionType)):
        return f"{obj.__module__}.{obj.__qualname__}"

    cls = type(obj)
    return f"{cls.__module__}.{cls.__qualname__}"

options: show_root_heading: false

spakky.core.utils.naming

Python naming convention utilities.

This module provides functions for checking Python naming conventions, such as identifying public/private identifiers.

PRIVATE_PREFIX = '_' module-attribute

Prefix for private identifiers in Python naming convention.

DUNDER_PREFIX = '__' module-attribute

Prefix and suffix for magic/dunder methods.

is_dunder_name(name)

Check if a name is a magic/dunder method name.

Dunder (double underscore) names like init, str are special methods that are considered public in Python.

Parameters:

Name	Type	Description	Default
name	str	The identifier name to check.	required

Returns:

Type	Description
bool	True if the name is a dunder name (starts and ends with __).

Example

is_dunder_name("init") True is_dunder_name("str") True is_dunder_name("__private") False

Source code in core/spakky/src/spakky/core/utils/naming.py

def is_dunder_name(name: str) -> bool:
    """Check if a name is a magic/dunder method name.

    Dunder (double underscore) names like __init__, __str__ are special
    methods that are considered public in Python.

    Args:
        name: The identifier name to check.

    Returns:
        True if the name is a dunder name (starts and ends with __).

    Example:
        >>> is_dunder_name("__init__")
        True
        >>> is_dunder_name("__str__")
        True
        >>> is_dunder_name("__private")
        False
    """
    return (
        name.startswith(DUNDER_PREFIX)
        and name.endswith(DUNDER_PREFIX)
        and len(name) > 4  # noqa: PLR2004
    )

is_public_name(name)

Check if a name follows public naming convention.

In Python: - Names starting with underscore are private (e.g., _internal) - Names with double underscore prefix are name-mangled (e.g., private) - Dunder names (__init, str) are public magic methods

Parameters:

Name	Type	Description	Default
name	str	The identifier name to check.	required

Returns:

Type	Description
bool	True if the name is public.

Example

is_public_name("username") True is_public_name("init") True is_public_name("_internal") False is_public_name("__private") False

Source code in core/spakky/src/spakky/core/utils/naming.py

def is_public_name(name: str) -> bool:
    """Check if a name follows public naming convention.

    In Python:
    - Names starting with underscore are private (e.g., _internal)
    - Names with double underscore prefix are name-mangled (e.g., __private)
    - Dunder names (__init__, __str__) are public magic methods

    Args:
        name: The identifier name to check.

    Returns:
        True if the name is public.

    Example:
        >>> is_public_name("username")
        True
        >>> is_public_name("__init__")
        True
        >>> is_public_name("_internal")
        False
        >>> is_public_name("__private")
        False
    """
    if is_dunder_name(name):
        return True
    return not name.startswith(PRIVATE_PREFIX)

is_private_name(name)

Check if a name follows private naming convention.

In Python, names starting with underscore are considered private, except for dunder names which are public magic methods.

Parameters:

Name	Type	Description	Default
name	str	The identifier name to check.	required

Returns:

Type	Description
bool	True if the name is private.

Example

is_private_name("_internal") True is_private_name("mangled") True is_private_name("__init") False is_private_name("public") False

Source code in core/spakky/src/spakky/core/utils/naming.py

def is_private_name(name: str) -> bool:
    """Check if a name follows private naming convention.

    In Python, names starting with underscore are considered private,
    except for dunder names which are public magic methods.

    Args:
        name: The identifier name to check.

    Returns:
        True if the name is private.

    Example:
        >>> is_private_name("_internal")
        True
        >>> is_private_name("__mangled")
        True
        >>> is_private_name("__init__")
        False
        >>> is_private_name("public")
        False
    """
    return not is_public_name(name)

options: show_root_heading: false

spakky.core.utils.uuid

uuid7()

Generate a UUID v7 (time-ordered) identifier.

Source code in core/spakky/src/spakky/core/utils/uuid.py

def uuid7() -> UUID:
    """Generate a UUID v7 (time-ordered) identifier."""
    return cast(UUID, get_id())

options: show_root_heading: false