spakky 6.4.0

Core module for Spakky framework to support DI/IoC, AOP, Plugin system, and more.

Spakky

Core module for Spakky Framework - a Spring-inspired dependency injection framework for Python.

Installation

pip install spakky

Or install with plugins:

pip install spakky[fastapi]
pip install spakky[fastapi,kafka,security]

Features

• Dependency Injection: Powerful IoC container with constructor injection
• Aspect-Oriented Programming: Cross-cutting concerns with @Aspect
• Plugin System: Extensible architecture via entry points
• Stereotypes: Semantic annotations (@Controller, @UseCase, etc.)
• Scopes: Singleton, Prototype, and Context-scoped beans
• Type-Safe: Built with Python type hints
• Async First: Native async/await support

Quick Start

Define Pods

from spakky.core.pod.annotations.pod import Pod

@Pod()
class UserRepository:
    def find_by_id(self, user_id: int) -> User | None:
        # Database query logic
        pass

@Pod()
class UserService:
    def __init__(self, repository: UserRepository) -> None:
        self.repository = repository

    def get_user(self, user_id: int) -> User | None:
        return self.repository.find_by_id(user_id)

Bootstrap Application

from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext
import my_app

app = (
    SpakkyApplication(ApplicationContext())
    .load_plugins()
    .scan(my_app)  # or .scan() to auto-detect caller's package
    .start()
)

# Get a service from the container
user_service = app.container.get(UserService)

📘 Auto-scan: When scan() is called without arguments, it automatically detects the caller's package and scans it. This also works in Docker environments where the application root may not be in sys.path - the framework automatically adds the necessary path.

Discovery Manifest

Scan discovery manifest reuse is opt-in and does not replace container caches. Enable it before scan() to persist discovered Pod/Tag candidates and reuse them when the scan target, exclude patterns, Python version, schema version, and source file mtimes/sizes are unchanged:

from pathlib import Path

from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext

app = (
    SpakkyApplication(ApplicationContext())
    .enable_startup_diagnostics()
    .enable_discovery_manifest(Path(".spakky/cache/discovery-manifest.json"))
    .scan(my_app)
)

scan_record = app.startup_report.records[0]
decision = scan_record.diagnostic_details[0].value  # miss, hit, stale_schema, stale_input

If no path is provided, Spakky uses the deterministic project-local cache path .spakky/cache/discovery-manifest.json. Missing, stale, or malformed manifests fall back to fresh discovery and record the decision in startup diagnostics. The decision values are miss, hit, stale_schema, and stale_input; hit replays stored candidates through the normal registration path, while every other decision performs fresh discovery.

Startup Diagnostics

Startup diagnostics are opt-in. The default recorder is no-op, so existing startup behavior is unchanged until diagnostics are explicitly enabled:

from spakky.core.application.application import SpakkyApplication
from spakky.core.application.application_context import ApplicationContext

app = SpakkyApplication(ApplicationContext()).enable_startup_diagnostics()

app.startup_phase_recorder.record_success(
    phase_name="scan",
    elapsed_seconds=0.12,
    processed_count=4,
)

with app.startup_phase_recorder.record_phase(phase_name="start") as phase:
    phase.set_processed_count(1)
    app.start()

report = app.startup_report
first_phase = report.records[0]

StartupReport stores each startup phase name, elapsed seconds, processed count, success/failure status, optional diagnostic details, and an optional structured failure summary. Failure summaries keep the exception type name, message, and diagnostic details without retaining the raw exception object. The application startup pipeline records phases in execution order: load_plugins, scan, registration, post_processor_registration, instantiation, post_processing, and service_start.

DI dependency failures preserve their existing exception types while attaching structured dependency diagnostics from Pod.dependencies, including the failed Pod, dependency parameter, requested type, and dependency path.

Pod Scopes

from spakky.core.pod.annotations.pod import Pod

# Singleton (default) - one instance per container
@Pod(scope=Pod.Scope.SINGLETON)
class SingletonService:
    pass

# Prototype - new instance on each request
@Pod(scope=Pod.Scope.PROTOTYPE)
class PrototypeService:
    pass

# Context - scoped to request/context lifecycle
@Pod(scope=Pod.Scope.CONTEXT)
class ContextScopedService:
    pass

Qualifiers

from spakky.core.pod.annotations.pod import Pod
from spakky.core.pod.annotations.primary import Primary

# Named qualifier
@Pod(name="mysql")
class MySQLRepository(IRepository):
    pass

@Pod(name="postgres")
class PostgresRepository(IRepository):
    pass

# Primary - preferred when multiple implementations exist
@Primary()
@Pod()
class DefaultRepository(IRepository):
    pass

Stereotypes

from spakky.core.stereotype.controller import Controller
from spakky.core.stereotype.usecase import UseCase

@Controller()
class UserController:
    """Groups related handlers together."""
    pass

@UseCase()
class CreateUserUseCase:
    """Encapsulates business logic."""
    pass

Aspect-Oriented Programming

from dataclasses import dataclass
from spakky.core.aop.aspect import Aspect
from spakky.core.aop.interfaces.aspect import IAspect
from spakky.core.aop.pointcut import Before, After
from spakky.core.common.annotation import FunctionAnnotation
from spakky.core.pod.annotations.order import Order

@dataclass
class Traced(FunctionAnnotation): ...

# Create custom aspect
@Order(0)
@Aspect()
class TracingAspect(IAspect):
    @Before(lambda m: Traced.exists(m))
    def before(self, *args, **kwargs) -> None:
        print("Before method execution")

    @After(lambda m: Traced.exists(m))
    def after(self, *args, **kwargs) -> None:
        print("After method execution")

# Apply to methods
@Pod()
class MyService:
    @Traced()
    def my_method(self) -> str:
        return "Hello"

Async Aspects

from spakky.core.aop.aspect import AsyncAspect
from spakky.core.aop.interfaces.aspect import IAsyncAspect
from spakky.core.aop.pointcut import Around

@Order(0)
@AsyncAspect()
class TimingAspect(IAsyncAspect):
    @Around(lambda m: hasattr(m, "__timed__"))
    async def around_async(self, joinpoint, *args, **kwargs):
        start = time.time()
        result = await joinpoint(*args, **kwargs)
        elapsed = time.time() - start
        print(f"Execution time: {elapsed:.2f}s")
        return result

Context Management

ApplicationContext provides context-scoped value storage:

from spakky.core.application.application_context import ApplicationContext

context = ApplicationContext()

# Get unique context ID
context_id = context.get_context_id()

# Store and retrieve context values
context.set_context_value("user_id", 123)
user_id = context.get_context_value("user_id")  # Returns 123

# Clear context (except system-managed keys)
context.clear_context()

⚠️ Note: System-managed keys like "__spakky_context_id__" cannot be overridden via set_context_value().

Tag Registry

ApplicationContext implements ITagRegistry for managing custom metadata tags. Tags are dataclass-based annotations that can be registered and queried at runtime.

Defining Custom Tags

from dataclasses import dataclass
from spakky.core.pod.annotations.tag import Tag

@dataclass(eq=False)
class MyCustomTag(Tag):
    """Custom tag for marking specific components."""
    category: str = ""

Registering and Querying Tags

from spakky.core.application.application_context import ApplicationContext

context = ApplicationContext()

# Register tags
tag = MyCustomTag(category="database")
context.register_tag(tag)

# Check if tag exists
exists = context.contains_tag(tag)  # True

# Get all tags
all_tags = context.tags  # frozenset of all registered tags

# Filter tags with selector
db_tags = context.list_tags(lambda t: isinstance(t, MyCustomTag) and t.category == "database")

Tag Registry Aware Pods

Pods can receive the tag registry via ITagRegistryAware:

from spakky.core.pod.annotations.pod import Pod
from spakky.core.pod.interfaces.aware.tag_registry_aware import ITagRegistryAware
from spakky.core.pod.interfaces.tag_registry import ITagRegistry

@Pod()
class SchemaRegistry(ITagRegistryAware):
    def __init__(self) -> None:
        self._tag_registry: ITagRegistry | None = None

    def set_tag_registry(self, tag_registry: ITagRegistry) -> None:
        self._tag_registry = tag_registry
        # Access registered tags
        for tag in tag_registry.list_tags(MyCustomTag.exists):
            # Process tags...
            pass

Plugin System

Plugins extend framework functionality through entry points.

Creating a Plugin

1. Create package with uv init --lib spakky-<name> in plugins/ directory
2. Register in root pyproject.toml's [tool.uv.workspace] members
3. Define entry point in plugin's pyproject.toml:

[project.entry-points."spakky.plugins"]
spakky-<name> = "spakky.plugins.<name>.main:initialize"

4. Implement initialization function:

# In spakky.plugins.<name>/main.py
from spakky.core.application.application import SpakkyApplication

def initialize(app: SpakkyApplication) -> None:
    # Register plugin components
    pass

See Contributing Guide for detailed instructions.

Available Plugins

Plugin	Description
spakky-fastapi	FastAPI integration
spakky-typer	Typer CLI integration
spakky-sqlalchemy	SQLAlchemy ORM integration
spakky-kafka	Apache Kafka event system
spakky-rabbitmq	RabbitMQ event system
spakky-celery	Celery task dispatch
spakky-logging	Structured logging with AOP
spakky-opentelemetry	OpenTelemetry SDK bridge
spakky-security	Security utilities

Core Modules

Module	Description
spakky.core.pod	Dependency injection container and annotations
spakky.core.aop	Aspect-oriented programming framework
spakky.core.application	Application context and lifecycle
spakky.core.stereotype	Semantic stereotype annotations
spakky.core.service	Service lifecycle interfaces
spakky.core.common	Core utilities (annotation, types, metadata)
spakky.core.utils	Utility functions

Related Packages

Package	Description
spakky-domain	DDD building blocks (Entity, AggregateRoot, ValueObject, Event)
spakky-data	Repository and transaction abstractions
spakky-event	Event handling (@EventHandler stereotype)
spakky-task	Task queue abstraction (@TaskHandler, @task, @schedule)
spakky-tracing	Distributed tracing abstraction (TraceContext, Propagator)
spakky-outbox	Transactional Outbox pattern (OutboxEventBus, Relay)

License

MIT