spakky

DI Container, 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

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.

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
    if path is None:  # pragma: no cover
        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 is_package(path):
        modules = list_modules(path, exclude)
    else:  # pragma: no cover
        modules = {resolve_module(path)}

    for item in modules:
        for obj in list_objects(item, lambda x: Pod.exists(x) or Tag.exists(x)):
            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

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.
    """
    for entry_point in entry_points(group=PLUGIN_PATH):  # pragma: no cover
        if include is not None:  # pragma: no cover
            if Plugin(name=entry_point.name) not in include:  # pragma: no cover
                continue  # pragma: no cover
        entry_point_function: Callable[[SpakkyApplication], None] = (
            entry_point.load()
        )  # pragma: no cover
        entry_point_function(self)  # pragma: no cover
    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()
    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: Exception

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.__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.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.

__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.

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.

__resolve_candidate(type_, name, qualifiers)

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],
) -> 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.
    """

    def qualify_pod(pod: Pod) -> bool:
        if any(qualifiers):
            return all(qualifier.selector(pod) for qualifier in qualifiers)
        if name is not None:
            return pod.name == name
        return pod.is_primary

    # Use type index for O(1) lookup instead of O(n) iteration
    pods = self.__type_cache.get(type_, set()).copy()
    if not pods:
        return None

    # Fast path: single candidate - no need to filter
    if len(pods) == 1:
        return next(iter(pods))

    # Multiple candidates: filter by qualifier/name/primary
    qualified = {pod for pod in pods if qualify_pod(pod)}
    if len(qualified) == 1:
        return qualified.pop()
    if not qualified:
        return None
    raise NoUniquePodError(type_, [p.name for p in qualified])

__instantiate_pod(pod, dependency_hierarchy)

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, ...]
) -> 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:
        raise CircularDependencyGraphDetectedError(
            list(dependency_hierarchy) + [pod.type_]
        )
    new_hierarchy = dependency_hierarchy + (pod.type_,)
    dependencies = {
        name: self.__get_internal(
            type_=remove_none(dependency.type_)
            if is_optional(dependency.type_)
            else dependency.type_,
            name=name,
            dependency_hierarchy=new_hierarchy,
            qualifiers=dependency.qualifiers,
        )
        for name, dependency in pod.dependencies.items()
    }
    instance: object = pod.instantiate(dependencies=dependencies)
    post_processed: object = self.__post_process_pod(instance)
    return post_processed

__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:
        pod = post_processor.post_process(pod)
    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.__get_internal(type_=pod.type_, name=pod.name) is None
        ):  # pragma: no cover
            raise NoSuchPodError(pod.type_, pod.name)

__get_internal(type_, name, dependency_hierarchy=None, qualifiers=None)

Internal method to get or create a Pod instance.

Parameters:

Name	Type	Description	Default
type_	type[ObjectT]	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
ObjectT | 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(
    self,
    type_: type[ObjectT],
    name: str | None,
    dependency_hierarchy: tuple[type, ...] | None = None,
    qualifiers: list[Qualifier] | None = None,
) -> ObjectT | 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 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
        if type_ not in self.__forward_type_map:  # pragma: no cover
            return None
        type_ = self.__forward_type_map[type_]  # pragma: no cover

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

    # Try to hit the cache by scope type of pod
    match pod.scope:
        case Pod.Scope.SINGLETON:
            if (cached := self.__get_singleton_cache(pod)) is not None:
                return cast(ObjectT, cached)
            # Double-checked locking for thread-safe lazy singleton creation
            with self.__singleton_lock:
                # Re-check cache after acquiring lock
                if (cached := self.__singleton_cache.get(pod.name)) is not None:
                    return cast(ObjectT, cached)
                instance = self.__instantiate_pod(pod, dependency_hierarchy)
                self.__set_singleton_cache(pod, instance)
                return cast(ObjectT, instance)
        case Pod.Scope.CONTEXT:
            if (cached := self.__get_context_cache(pod)) is not None:
                return cast(ObjectT, cached)
        case Pod.Scope.PROTOTYPE:
            pass

    instance = self.__instantiate_pod(
        pod,
        dependency_hierarchy,
    )

    # Cache the instance based on pod scope
    match pod.scope:
        case Pod.Scope.CONTEXT:
            self.__set_context_cache(pod, instance)
        case Pod.Scope.PROTOTYPE:
            pass

    return cast(ObjectT, instance)

__start_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) -> None:
    """Start all registered sync and async services.

    Raises:
        EventLoopThreadAlreadyStartedInApplicationContextError: If already started.
    """
    if self.__event_loop is not None:  # pragma: no cover
        raise EventLoopThreadAlreadyStartedInApplicationContextError
    if self.__event_thread is not None:  # pragma: no cover
        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()

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

    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
        raise EventLoopThreadNotStartedInApplicationContextError
    if self.__event_thread is None:  # pragma: no cover
        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

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

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
        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
            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
    # pod.type_은 클래스 자체이므로 같은 타입 두 Pod은 이름 충돌로 위에서 차단됨
    if pod.type_ not in self.__type_cache:  # pragma: no branch
        self.__type_cache[pod.type_] = set()
    self.__type_cache[pod.type_].add(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)

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

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()

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

def start(self) -> 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
        raise ApplicationContextAlreadyStartedError()
    self.__is_started = True
    self.__register_post_processors()
    self.__initialize_pods()
    self.__start_services()

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

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
            raise ApplicationContextAlreadyStoppedError()
        self.__stop_services()
        self.__clear_all()
        self.__is_started = False

get(type_, name=None)

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

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

Get a Pod instance by type and optional name.

Parameters:

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

Returns:

Type	Description
ObjectT | 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

def get(
    self,
    type_: type[ObjectT],
    name: str | None = None,
) -> ObjectT | 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
        raise NoSuchPodError(type_, name)
    return instance

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

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

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

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

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

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
        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

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

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
        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

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.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)

__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)

__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.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.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[ObjectT]
) -> ObjectT

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

Manually inject a Pod from the container.

Parameters:

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

Returns:

Type	Description
object | ObjectT	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(
    context: IContainer,
    type_: type[ObjectT],
    name: str | None = None,
) -> object | ObjectT:
    """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

Protocol and errors for Pod container interface.

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

CircularDependencyGraphDetectedError(dependency_chain)

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

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

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

    Args:
        dependency_chain: List of types in dependency order, ending with the duplicate type.
    """
    super().__init__()
    self.dependency_chain = dependency_chain

__str__()

Format error message with visual dependency path.

Returns:

Type	Description
str	Formatted string showing the circular dependency path with tree visualization.

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

def __str__(self) -> str:
    """Format error message with visual dependency path.

    Returns:
        Formatted string showing the circular dependency path with tree visualization.
    """
    if not self.dependency_chain:
        return self.message

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

        # Mark the last element as CIRCULAR
        if i == len(self.dependency_chain) - 1:
            lines.append(f"{indent}{arrow}{type_name} (CIRCULAR!)")
        else:
            lines.append(f"{indent}{arrow}{type_name}")

    return "\n".join(lines)

NoSuchPodError

Bases: AbstractSpakkyPodError

Raised when requested Pod cannot be found in container.

NoUniquePodError

Bases: AbstractSpakkyPodError

Raised when multiple Pods match criteria without clear qualification.

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

Protocol 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.
    """
    ...

get(type_, name=None) abstractmethod

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

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

Get a Pod instance by type and optional name.

Parameters:

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

Returns:

Type	Description
ObjectT | object	The Pod instance.

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

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

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

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

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

Protocol 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

Protocol 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() 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) -> 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

Protocol for Pod post-processors.

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

IPostProcessor

Bases: ABC

Protocol 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.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.

options: show_root_heading: false

Service

spakky.core.service.interfaces

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

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

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

def stop(self) -> None:
    """Stop service and wait for thread to finish."""
    self._stop_event.set()
    if self._thread:  # pragma: no cover
        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

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

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

async def stop_async(self) -> None:
    """Stop service and wait for task to finish."""
    self._stop_event.set()
    if self._task:  # pragma: no cover
        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

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

Stereotype

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

Common

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[ObjectT]	The class to annotate.	required

Returns:

Type	Description
type[ObjectT]	type[ObjectT]: The annotated class.

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

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

    Args:
        obj (type[ObjectT]): The class to annotate.

    Returns:
        type[ObjectT]: 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[..., AnyT]	The function to annotate.	required

Returns:

Type	Description
Callable[..., AnyT]	Callable[..., AnyT]: The annotated function.

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

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

    Args:
        obj (Callable[..., AnyT]): The function to annotate.

    Returns:
        Callable[..., AnyT]: 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.interfaces

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.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

Protocol 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)

Bases: Generic[ObjectT]

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	ObjectT	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: ObjectT,
    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

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:

Name	Type	Description
ObjectT	ObjectT	A proxy instance that wraps the target object.

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

def create(self) -> ObjectT:
    """Create a proxy instance for the target object.

    Returns:
        ObjectT: 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
            return type(None)
        if len(non_none_args) == 1:
            return non_none_args[0]
        return Union[non_none_args]  # type: ignore[valid-type]  # Dynamic Union construction
    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)
        except Exception:  # noqa: BLE001
            continue

        if callable(actual):
            result.append((name, actual))

    return result

options: show_root_heading: false

Utilities

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)
    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