canary-framework 0.4.13

Lightweight decorator-driven Python async service framework with dependency injection

Canary Framework

Lightweight Python Async Service Framework — Decorator-Driven, Annotation-Based DI

---

Canary Framework is a decorator-driven async service framework for Python. Core philosophy: Services are the smallest unit, modules compose services, and modules themselves are services.

Core Features

• Decorator-Driven — Use @service, @module, @router decorators with explicit base class inheritance
• Annotation-Based DI — Declare dependencies with type annotations: db: DatabaseService, no boilerplate
• Topological Startup — Kahn's algorithm ensures dependencies start first
• Lifecycle Management — @after_config/@after_init/@before_startup/@before_shutdown hooks
• ASGI Compatible — Built on Starlette, works with uvicorn and other ASGI servers
• Modular Architecture — Hierarchical composition with nested modules
• OpenAPI Support — Auto-generated Swagger UI and ReDoc documentation

Installation

pip install canary-framework

Quick Start

from canary_framework import module, service, router, get, post, after_config
from canary_framework.core.service import ServiceBase
from canary_framework.core.module import ModuleBase
from canary_framework.core.router import RouterBase

@service()
class DatabaseService(ServiceBase):
    @after_config
    async def connect(self):
        self.conn = "connected"

@service()
class UserService(ServiceBase):
    db: DatabaseService

    async def get_user(self, user_id: int):
        return {"id": user_id, "name": "Alice"}

@router(prefix="/api", tags=["users"])
class ApiRouter(RouterBase):
    user_service: UserService

    @get("/users/{user_id}")
    async def get_user(self, user_id: int) -> dict:
        return self.user_service.get_user(user_id)

    @post("/users")
    async def create_user(self, body: dict) -> dict:
        return {"id": 1, **body}

@module(services=[DatabaseService, UserService, ApiRouter])
class App(ModuleBase):
    pass

# ---- Entry Point ----

async def setup():
    app = App()
    await app.configure()
    await app.init()
    return app

if __name__ == "__main__":
    import asyncio
    import uvicorn

    app = asyncio.run(setup())
    uvicorn.run(app, lifespan="on")

Configuration

Use @config with CanaryConfig to customize framework behavior:

from canary_framework import config
from canary_framework.common.config import CanaryConfig

@config
class AppConfig(CanaryConfig):
    host: str = "0.0.0.0"
    port: int = 8080
    openapi_title: str = "My API"
    log_level: str = "DEBUG"

async def setup():
    cfg = AppConfig()
    app = App()
    await app.configure(cfg)
    await app.init()
    return app, cfg

Web Example with OpenAPI

from canary_framework import module, router, get, post
from canary_framework.core.module import ModuleBase
from canary_framework.core.router import RouterBase
from pydantic import BaseModel, Field

class UserRequest(BaseModel):
    name: str = Field(description="User name")
    email: str = Field(description="User email")

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@router(prefix="/users", tags=["Users"])
class UsersRouter(RouterBase):
    @get("/", summary="List users", description="Get all users")
    async def list_users(self) -> list[UserResponse]:
        return []

    @post("/",
          summary="Create user",
          description="Create a new user",
          request_model=UserRequest,
          response_model=UserResponse)
    async def create_user(self, user: UserRequest) -> UserResponse:
        return UserResponse(id=1, name=user.name, email=user.email)

@module(services=[UsersRouter])
class App(ModuleBase):
    pass

OpenAPI Documentation

Access automatically generated documentation:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

Architecture

src/canary_framework/
├── common/              # Shared infrastructure
│   ├── errors.py        # Framework exceptions
│   ├── routing.py       # Route path parsing
│   └── types.py         # Data classes, markers, and type aliases
├── core/                # Base classes
│   ├── module.py        # ModuleBase — orchestration and DI
│   ├── service.py       # ServiceBase — lifecycle and ASGI
│   └── router.py        # RouterBase — ASGI routing + OpenAPI docs
├── decorators/          # Decorator implementations
│   ├── module.py        # @module
│   ├── service.py       # @service
│   ├── router.py        # @router, @get/@post/...
│   └── lifecycle.py     # @after_config, @after_init, etc.
└── engine/              # Runtime engine
    ├── registry.py      # Service registry
    ├── injector.py      # Topological sort
    ├── hooks.py         # Lifecycle hook discovery
    ├── openapi.py       # OpenAPI schema generation
    ├── params.py        # Route parameter resolution
    └── logging.py       # Framework logging

Dependency Injection Flow

@service() class MyService:
    db: DatabaseService      ←  1. User declares dependency via annotation

    ↓ configure phase

resolve_deps(MyService)
    → get_type_hints() reads {db: DatabaseService}
    → filters by CF_SERVICE_MARKER
    → returns {"db": DatabaseService}

    ↓ registration: recursively registers DatabaseService
    ↓ topological_sort: build dependency graph
    ↓ instantiation: creates instances in order
    ↓ wiring:

setattr(instance, "db", db_instance)   ←  3. Injected with annotation key name

Lifecycle Flow

app.configure(config_instance)
  ├── Register all services + transitive deps
  ├── Topological sort (Kahn's algorithm)
  ├── Instantiate services
  ├── Inject dependencies (annotation-driven)
  ├── Call configure() on each service (topological order)
  └── Invoke @after_config hooks

app.init()
  ├── Invoke @after_init hook
  └── Call init() on each service (topological order)

app.startup()
  ├── Invoke @before_startup hook
  └── Call startup() on each service (topological order)

app.shutdown()
  ├── Invoke @before_shutdown hook
  └── Call shutdown() on each service (reverse topological order)

Testing

# Run all tests
pytest

# Run unit tests
pytest tests/unit/

# Run integration tests
pytest tests/integration/

Community

• 💬 Discussions
• 🐛 Issues
• 📖 Docs

Contributing

See CONTRIBUTING.md.

License

Apache 2.0 · Copyright 2026 Zhang Wenbo (Canary)