python-components 0.4.0

A lightweight Python framework for managing the lifecycle and dependencies of stateful components

Python Components

A lightweight Python framework for managing the lifecycle and dependencies of stateful components.

Inspired by Stuart Sierra's Component library for Clojure.

Overview

Component is a small framework for managing the lifecycle of software components that have runtime state. It provides:

• Dependency injection using explicit dependency declarations
• Lifecycle management with start and shutdown methods
• Automatic dependency resolution using topological sorting
• Clean separation between component definition and composition

Installation

uv add python-components

Or with pip:

pip install python-components

Quick Start

1. Define Your Components

from python_components import Component

class Database(Component):
    def __init__(self, host: str, port: int):
        super().__init__()
        self.host = host
        self.port = port
        self.connection = None
    
    def start(self):
        """Connect to the database."""
        self.connection = connect_to_db(self.host, self.port)
        print(f"Database connected to {self.host}:{self.port}")
    
    def shutdown(self):
        """Close the database connection."""
        if self.connection:
            self.connection.close()
            print("Database connection closed")

class Cache(Component):
    def __init__(self):
        super().__init__()
        self.database = None  # injected by the System, see below
        self.data = {}

    def start(self):
        """Warm the cache using the injected database dependency."""
        self.data = self.database.load_initial_cache()
        print("Cache warmed from database")

    def shutdown(self):
        """Clear the cache."""
        self.data.clear()
        print("Cache cleared")

class WebServer(Component):
    def __init__(self, port: int):
        super().__init__()
        self.port = port
        self.server = None
    
    def start(self):
        """Start the web server."""
        self.server = start_server(self.port)
        print(f"Web server started on port {self.port}")
    
    def shutdown(self):
        """Stop the web server."""
        if self.server:
            self.server.stop()
            print("Web server stopped")

2. Compose Your System

from python_components import System

# Build the system
system_map = {
    "database": Database(host="localhost", port=5432),
    "cache": Cache().using(["database"]),
    "web_server": WebServer(port=8080).using(["database", "cache"]),
}

system = System(system_map)

3. Start and Stop the System

# Start all components in dependency order
system.start()

# Your application runs...

# Shutdown all components in reverse dependency order
system.shutdown()

Key Concepts

Components

A component is any object that:

• Extends the Component abstract base class
• Implements start() and shutdown() methods
• May depend on other components

Dependencies

Declare dependencies using the using() method:

component = MyComponent().using(["dependency1", "dependency2"])

Before the component's start() runs, the System sets each declared dependency as an instance attribute on the component, named after the key in the system map:

cache = Cache().using(["database"])
# before cache.start() is called, System does the equivalent of:
#   cache.database = system_map["database"]

If you want the injected attribute to have a different name than the system-map key (for example, to avoid a name clash or to keep the component decoupled from a specific system-map layout), pass a mapping of attribute name to system-map key instead:

api = Api().using({"db": "async_database"})
# before api.start() is called, System does the equivalent of:
#   api.db = system_map["async_database"]

component.dependencies is a dict[str, str] (attribute name -> system-map key) populated by using(). Dependencies are automatically resolved and components are started in dependency order; injection happens once, right before start() runs.

Accessing components from outside the system

Components that are part of the system get their dependencies injected as attributes automatically. Code that lives outside the system — such as a web request handler — can look a component up by name with system.get_component(name):

from fastapi import Request

@app.get("/users/{user_id}")
def get_user(user_id: str, request: Request):
    database = request.app.state.system.get_component("database")
    return database.fetch_user(user_id)

System

A system is a collection of components with declared dependencies. The System class:

• Builds a dependency graph from component declarations
• Performs topological sorting to determine initialization order
• Starts components in dependency order
• Shuts down components in reverse dependency order
• Detects circular dependencies and raises an error

Async components

Components may define start()/shutdown() as regular functions or as async def. The System detects which is which and awaits async ones automatically — you can freely mix sync and async components in the same system.

class AsyncDatabase(Component):
    def __init__(self, dsn: str):
        super().__init__()
        self.dsn = dsn
        self.pool = None

    async def start(self):
        self.pool = await create_pool(self.dsn)

    async def shutdown(self):
        await self.pool.close()

The canonical async entry points are await system.astart() and await system.ashutdown(). system.start()/system.shutdown() (no a prefix) still work, with behavior that depends on the system's contents:

System contents	Calling system.start() / system.shutdown()
Only sync components	Runs entirely synchronously; never touches asyncio.
Has async components, no event loop running	Drives the coroutine with asyncio.run(...) for you.
Has async components, event loop already running	Raises RuntimeError telling you to await system.astart() / await system.ashutdown() instead.

In an async application (e.g. a FastAPI app), use astart()/ashutdown() directly, or use the system as an async context manager:

from contextlib import asynccontextmanager

from fastapi import FastAPI

system = System({
    "database": AsyncDatabase(dsn="postgresql://localhost/mydb"),
    "cache": Cache().using(["database"]),
})

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with system:  # calls await system.astart() / await system.ashutdown()
        app.state.system = system
        yield

app = FastAPI(lifespan=lifespan)

Error handling

• Start failure: if a component's start() raises, the System shuts down every component that had already started, in reverse order, then raises ComponentStartError. The failing component's name is available as error.component_name, the original exception as error.__cause__, and any exceptions raised while rolling back as error.rollback_errors.
• Shutdown failure: shutdown()/ashutdown() always attempts every started component, even if some raise. If any fail, an ExceptionGroup containing all of the shutdown exceptions is raised after every component has had a chance to clean up.
• Cycles: a circular dependency raises CycleError (re-exported from the standard library's graphlib; it's a subclass of ValueError).
• Missing dependency: referencing a system-map key that doesn't exist raises KeyError.
• Injection safety: all injections are validated before any component is mutated, so a failure never leaves the system partially injected. Injecting under a name that collides with a class attribute or method, a framework-reserved attribute (dependencies; plus system_map/state/_started on a nested System), or a target a __slots__-restricted component doesn't declare raises AttributeError.
• Sync/async mismatch: on the pure-sync path, a plain def start()/shutdown() that returns an awaitable raises TypeError instead of silently dropping it — declare the method with async def so the System awaits it.
• Context managers and exceptions: if the with block raises and shutdown also fails, the original exception still propagates; the shutdown failure is attached to it as a note instead of replacing it.
• State and restart: system.state is a SystemState enum (SystemState.INITIALIZED, SystemState.STARTED, SystemState.STOPPED). Starting an already-started system raises RuntimeError. Shutdown is idempotent — calling it on a system that isn't started is a no-op. A stopped system can be started again (restart).

from python_components import ComponentStartError, CycleError

try:
    system.start()
except ComponentStartError as exc:
    print(f"{exc.component_name} failed to start: {exc.__cause__}")
    for rollback_exc in exc.rollback_errors:
        print(f"  rollback also failed: {rollback_exc!r}")
except CycleError as exc:
    print(f"circular dependency: {exc}")

Example: Complete Application

from python_components import Component, System

class Database(Component):
    def __init__(self, connection_string: str):
        super().__init__()
        self.connection_string = connection_string
        self.connection = None
    
    def start(self):
        self.connection = create_connection(self.connection_string)
        print("✓ Database connected")
    
    def shutdown(self):
        self.connection.close()
        print("✓ Database disconnected")

class Cache(Component):
    def __init__(self):
        super().__init__()
        self.data = {}
    
    def start(self):
        print("✓ Cache initialized")
    
    def shutdown(self):
        self.data.clear()
        print("✓ Cache cleared")

class ApiService(Component):
    def __init__(self):
        super().__init__()
        self.server = None
    
    def start(self):
        self.server = start_api_server()
        print("✓ API service started")
    
    def shutdown(self):
        self.server.stop()
        print("✓ API service stopped")

# Compose the system
system = System({
    "database": Database("postgresql://localhost/mydb"),
    "cache": Cache().using(["database"]),
    "api": ApiService().using(["database", "cache"]),
})

# Lifecycle management: System is a context manager, so start/shutdown
# are handled for you, including on exceptions.
with system:
    pass  # Application logic here

# Equivalent, if you need more control over the shutdown path:
try:
    system.start()
    # Application logic here
finally:
    system.shutdown()

Features

• ✨ Simple API: Just implement start() and shutdown()
• 🔗 Explicit Dependencies: Clear, declarative dependency management, injected as attributes
• 📊 Automatic Ordering: Topological sorting ensures correct initialization order
• 🔄 Lifecycle Management: Consistent start/shutdown across all components, with restart support
• ⚡ Async Support: Mix sync and async components; astart()/ashutdown() and context managers included
• 🛟 Automatic Rollback: A failed start cleanly shuts down whatever already started
• 🚨 Cycle Detection: Catches circular dependencies at runtime
• 📦 Zero Dependencies: No runtime dependencies, built on the standard library
• 🐍 Pythonic & Typed: Uses type hints, ships py.typed, and follows Python best practices

Why Use Components?

Problem: Global State and Initialization Order

Without a component system, applications often struggle with:

• Global mutable state scattered throughout the codebase
• Unclear initialization order leading to subtle bugs
• Difficulty testing components in isolation
• Complex shutdown logic that's easy to forget

Solution: Managed Components

The Component pattern provides:

• Explicit lifecycle: Every stateful resource has clear start/shutdown methods
• Dependency injection: Components receive their dependencies explicitly
• Testability: Easy to create test doubles and inject them
• Reliability: Guaranteed correct initialization and cleanup order

Development

Running Tests

This project uses pytest for testing. To run the tests:

uv run pytest

To run tests with coverage:

uv run pytest --cov=python_components --cov-report=term-missing

Installing Development Dependencies

uv add pytest ruff --group=dev

Requirements

• Python >= 3.11
• No runtime dependencies

License

MIT License - see LICENSE file for details

Acknowledgments

This library is inspired by Alexandra Sierra's Component library for Clojure, which pioneered the pattern of explicit lifecycle management and dependency injection using immutable data structures.