tessl/pypi-dishka

Cute DI framework with scopes and agreeable API for Python dependency injection

—

Pending

Overview

Eval results

Files

Framework Integrations

Name: tessl/pypi-dishka
Author: tessl

Ready-to-use integrations for popular Python frameworks with automatic dependency injection setup and middleware configuration. These integrations handle the boilerplate of connecting Dishka to framework request lifecycles.

Capabilities

Common Integration Pattern

All framework integrations follow a consistent pattern with setup and injection functions.

def setup_dishka(container: Container | AsyncContainer, app: Any) -> None:
    """
    Set up Dishka integration with a framework application.
    
    Parameters:
    - container: Dishka container (sync or async depending on framework)
    - app: Framework application instance
    
    Side Effects:
    - Registers middleware/hooks for dependency injection
    - Configures container lifecycle management
    - Sets up error handling for dependency resolution
    """

def inject(func: Callable) -> Callable:
    """
    Decorator for automatic dependency injection in framework handlers.
    
    Parameters:
    - func: Handler function with FromDishka annotations
    
    Returns:
    Wrapped function with dependencies automatically injected
    
    Usage:
    Functions decorated with @inject will have parameters marked with
    FromDishka[Type] automatically resolved from the container.
    """

FastAPI Integration

Comprehensive integration for FastAPI with custom routing and automatic dependency injection.

from dishka.integrations.fastapi import setup_dishka, inject, FromDishka, DishkaRoute

def setup_dishka(
    container: Container | AsyncContainer, 
    app: FastAPI,
    *,
    finalize_container: bool = True
) -> None:
    """
    Set up Dishka integration with FastAPI.
    
    Parameters:
    - container: Dishka container
    - app: FastAPI application instance
    - finalize_container: Whether to close container on app shutdown
    """

class DishkaRoute(APIRoute):
    """Custom FastAPI route class with built-in dependency injection"""
    
    def get_route_handler(self) -> Callable:
        """Override to inject dependencies automatically"""

# Re-exported for convenience
FromDishka = FromDishka

Usage Example:

from fastapi import FastAPI, APIRouter
from dishka import make_async_container, Provider, Scope
from dishka.integrations.fastapi import setup_dishka, inject, FromDishka, DishkaRoute

# Set up dependencies
provider = Provider()
provider.provide(UserService, scope=Scope.REQUEST)
container = make_async_container(provider)

# Create FastAPI app
app = FastAPI()
setup_dishka(container, app)

# Using inject decorator
@app.get("/users/{user_id}")
@inject
async def get_user(
    user_id: int,  # From path parameter
    service: FromDishka[UserService]  # Injected dependency
) -> dict:
    return await service.get_user(user_id)

# Using custom route class
router = APIRouter(route_class=DishkaRoute)

@router.get("/users")
async def list_users(service: FromDishka[UserService]) -> list:
    return await service.list_users()

app.include_router(router)

Flask Integration

Integration for Flask with request-scoped dependency injection and cleanup.

from dishka.integrations.flask import setup_dishka, inject, FromDishka, FlaskProvider

def setup_dishka(container: Container, app: Flask) -> None:
    """
    Set up Dishka integration with Flask.
    
    Parameters:
    - container: Dishka container (must be synchronous)
    - app: Flask application instance
    """

class FlaskProvider(Provider):
    """Flask-specific provider with framework dependencies"""
    
    request = from_context(Request, scope=Scope.REQUEST)
    """Flask request object available as dependency"""

# Re-exported for convenience
FromDishka = FromDishka

Usage Example:

from flask import Flask
from dishka import make_container, Provider, Scope
from dishka.integrations.flask import setup_dishka, inject, FromDishka, FlaskProvider

# Set up dependencies
provider = Provider()
provider.provide(UserService, scope=Scope.REQUEST)

flask_provider = FlaskProvider()

container = make_container(provider, flask_provider)

# Create Flask app
app = Flask(__name__)
setup_dishka(container, app)

@app.route("/users/<int:user_id>")
@inject
def get_user(user_id: int, service: FromDishka[UserService]) -> dict:
    return service.get_user(user_id)

@app.route("/current-request")
@inject
def get_request_info(request: FromDishka[Request]) -> dict:
    return {"method": request.method, "path": request.path}

AIOHttp Integration

Integration for AIOHttp with async middleware and dependency injection.

from dishka.integrations.aiohttp import setup_dishka, inject, FromDishka, AioHttpProvider

def setup_dishka(container: AsyncContainer, app: Application) -> None:
    """
    Set up Dishka integration with AIOHttp.
    
    Parameters:
    - container: Dishka async container
    - app: AIOHttp Application instance
    """

class AioHttpProvider(Provider):
    """AIOHttp-specific provider with framework dependencies"""
    
    request = from_context(Request, scope=Scope.REQUEST)
    """AIOHttp request object available as dependency"""

# Re-exported for convenience
FromDishka = FromDishka

Usage Example:

from aiohttp import Application, web
from dishka import make_async_container, Provider, Scope
from dishka.integrations.aiohttp import setup_dishka, inject, FromDishka, AioHttpProvider

# Set up dependencies
provider = Provider()
provider.provide(UserService, scope=Scope.REQUEST)

aiohttp_provider = AioHttpProvider()

container = make_async_container(provider, aiohttp_provider)

# Create AIOHttp app
app = Application()
setup_dishka(container, app)

@inject
async def get_user(
    request: FromDishka[web.Request],
    service: FromDishka[UserService]
) -> web.Response:
    user_id = int(request.match_info['user_id'])
    user_data = await service.get_user(user_id)
    return web.json_response(user_data)

app.router.add_get("/users/{user_id}", get_user)

Starlette Integration

Integration for Starlette with middleware support and dependency injection.

from dishka.integrations.starlette import setup_dishka, inject, FromDishka, DishkaMiddleware

def setup_dishka(container: AsyncContainer, app: Starlette) -> None:
    """
    Set up Dishka integration with Starlette.
    
    Parameters:
    - container: Dishka async container
    - app: Starlette application instance
    """

class DishkaMiddleware:
    """Starlette middleware for dependency injection"""
    
    def __init__(self, app: ASGIApp, container: AsyncContainer): ...
    async def __call__(self, scope: Scope, receive: Receive, send: Send): ...

# Re-exported for convenience
FromDishka = FromDishka

Other Web Framework Integrations

Additional web framework integrations following similar patterns.

Sanic Integration:

from dishka.integrations.sanic import setup_dishka, inject, FromDishka, SanicProvider

def setup_dishka(container: AsyncContainer, app: Sanic) -> None: ...

class SanicProvider(Provider):
    request = from_context(Request, scope=Scope.REQUEST)

Litestar Integration:

from dishka.integrations.litestar import setup_dishka, inject, FromDishka

def setup_dishka(container: AsyncContainer, app: Litestar) -> None: ...

Task Queue Integrations

Integrations for task queue and job processing frameworks.

Celery Integration:

from dishka.integrations.celery import setup_dishka, inject, FromDishka

def setup_dishka(container: Container, app: Celery) -> None:
    """
    Set up Dishka integration with Celery.
    
    Parameters:
    - container: Dishka container
    - app: Celery application instance
    """

Usage Example:

from celery import Celery
from dishka import make_container, Provider, Scope
from dishka.integrations.celery import setup_dishka, inject, FromDishka

provider = Provider()
provider.provide(TaskProcessor, scope=Scope.REQUEST)

container = make_container(provider)
app = Celery('tasks')
setup_dishka(container, app)

@app.task
@inject
def process_task(task_data: dict, processor: FromDishka[TaskProcessor]) -> str:
    return processor.process(task_data)

TaskIQ Integration:

from dishka.integrations.taskiq import setup_dishka, inject, FromDishka

def setup_dishka(container: AsyncContainer, broker: AsyncBroker) -> None: ...

ARQ Integration:

from dishka.integrations.arq import setup_dishka, inject, FromDishka

def setup_dishka(container: AsyncContainer, worker_settings: dict) -> None: ...

Bot Framework Integrations

Integrations for Telegram bot frameworks.

Aiogram Integration:

from dishka.integrations.aiogram import setup_dishka, inject, FromDishka

def setup_dishka(container: AsyncContainer, dispatcher: Dispatcher) -> None:
    """
    Set up Dishka integration with Aiogram.
    
    Parameters:
    - container: Dishka async container
    - dispatcher: Aiogram dispatcher instance
    """

Usage Example:

from aiogram import Bot, Dispatcher
from dishka import make_async_container, Provider, Scope
from dishka.integrations.aiogram import setup_dishka, inject, FromDishka

provider = Provider()
provider.provide(BotService, scope=Scope.REQUEST)

container = make_async_container(provider)
bot = Bot(token="TOKEN")
dp = Dispatcher()
setup_dishka(container, dp)

@dp.message()
@inject
async def handle_message(
    message: Message,  # From aiogram
    service: FromDishka[BotService]  # Injected
) -> None:
    await service.process_message(message)

Telebot Integration:

from dishka.integrations.telebot import setup_dishka, inject, FromDishka

def setup_dishka(container: Container, bot: TeleBot) -> None: ...

CLI Integration

Integration for Click command-line framework.

from dishka.integrations.click import setup_dishka, inject, FromDishka

def setup_dishka(container: Container, app: click.Group) -> None:
    """
    Set up Dishka integration with Click.
    
    Parameters:
    - container: Dishka container
    - app: Click Group/Command instance
    """

Usage Example:

import click
from dishka import make_container, Provider, Scope
from dishka.integrations.click import setup_dishka, inject, FromDishka

provider = Provider()
provider.provide(ConfigService, scope=Scope.APP)

container = make_container(provider)

@click.group()
def cli():
    pass

setup_dishka(container, cli)

@cli.command()
@click.argument('filename')
@inject
def process_file(filename: str, config: FromDishka[ConfigService]) -> None:
    """Process a file with injected configuration service."""
    processor = FileProcessor(config.get_settings())
    processor.process(filename)

gRPC Integration

Integration for gRPCio server applications.

from dishka.integrations.grpcio import setup_dishka, inject, FromDishka

def setup_dishka(container: AsyncContainer, server: grpc.Server) -> None:
    """
    Set up Dishka integration with gRPC server.
    
    Parameters:
    - container: Dishka async container
    - server: gRPC server instance
    """

Message Streaming Integration

Integration for FastStream message processing framework.

from dishka.integrations.faststream import setup_dishka, inject, FromDishka

def setup_dishka(container: AsyncContainer, broker: FastStream) -> None:
    """
    Set up Dishka integration with FastStream.
    
    Parameters:
    - container: Dishka async container
    - broker: FastStream broker instance
    """

Integration Error Handling

Common error handling patterns across integrations.

class IntegrationError(DishkaError):
    """Base class for integration-related errors"""

class SetupError(IntegrationError):
    """Raised when integration setup fails"""

class InjectionError(IntegrationError):
    """Raised when dependency injection fails during request handling"""

Custom Integration Pattern

Pattern for creating custom framework integrations.

Basic Integration Structure:

from dishka import Container, AsyncContainer, FromDishka
from typing import Callable, Any

def setup_custom_framework(
    container: Container | AsyncContainer,
    app: Any
) -> None:
    """
    1. Register middleware/hooks with framework
    2. Set up container lifecycle management
    3. Configure error handling
    """
    # Implementation specific to framework
    pass

def inject(func: Callable) -> Callable:
    """
    1. Inspect function signature for FromDishka annotations
    2. Extract dependencies from container during request
    3. Call original function with injected dependencies
    """
    # Implementation for dependency injection
    pass

Install with Tessl CLI