### Full Dishka Example Source: https://github.com/reagento/dishka/blob/develop/docs/quickstart.md A complete example demonstrating Dishka installation, class definitions, provider setup, container creation, and dependency access. ```python import sqlite3 from collections.abc import Iterable from sqlite3 import Connection from typing import Protocol from dishka import Provider, Scope, make_container, provide class APIClient: ... class UserDAO(Protocol): ... class SQLiteUserDAO(UserDAO): def __init__(self, connection: Connection): ... class Service: def __init__(self, client: APIClient, user_dao: UserDAO): ... service_provider = Provider(scope=Scope.REQUEST) service_provider.provide(Service) service_provider.provide(SQLiteUserDAO, provides=UserDAO) service_provider.provide(APIClient, scope=Scope.APP) # override provider's scope class ConnectionProvider(Provider): @provide(scope=Scope.REQUEST) def new_connection(self) -> Iterable[Connection]: connection = sqlite3.connect(":memory:") yield connection connection.close() container = make_container(service_provider, ConnectionProvider()) # APIClient is bound to Scope.APP, so it can be accessed here # or from any scope inside including Scope.REQUEST client = container.get(APIClient) client = container.get(APIClient) # the same APIClient instance as above ``` -------------------------------- ### Full Dishka Quickstart Example Source: https://github.com/reagento/dishka/blob/develop/README.md A complete example combining all steps: installation, class definitions, provider configuration with scopes, container creation, and dependency access. ```python import sqlite3 from collections.abc import Iterable from sqlite3 import Connection from typing import Protocol from dishka import Provider, Scope, make_container, provide class APIClient: ... class UserDAO(Protocol): ... class SQLiteUserDAO(UserDAO): def __init__(self, connection: Connection): ... class Service: def __init__(self, client: APIClient, user_dao: UserDAO): ... service_provider = Provider(scope=Scope.REQUEST) service_provider.provide(Service) service_provider.provide(SQLiteUserDAO, provides=UserDAO) service_provider.provide(APIClient, scope=Scope.APP) # override provider's scope class ConnectionProvider(Provider): @provide(scope=Scope.REQUEST) def new_connection(self) -> Iterable[Connection]: connection = sqlite3.connect(":memory:") yield connection connection.close() container = make_container(service_provider, ConnectionProvider()) # APIClient is bound to Scope.APP, so it can be accessed here ``` -------------------------------- ### Combine Various Provider Registrations with provide_all Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide_all.md Demonstrates combining different provider registration methods like provide, alias, decorate, from_context, and provide_all within a single Provider class. This example shows an equivalent setup using individual registrations versus a combined approach. ```python from dishka import alias, decorate, provide, provide_all, Provider, Scope class OneByOne(Provider): scope = Scope.APP config = from_context(Config) user_dao = provide(UserDAOImpl, provides=UserDAO) post_dao = provide(PostDAOImpl, provides=PostDAO) post_reader = alias(source=PostDAOImpl, provides=PostReader) decorator = decorate(SomeDecorator, provides=SomeClass) class AllAtOnce(Provider): scope = Scope.APP provides = ( provide(UserDAOImpl, provides=UserDAO) + provide(PostDAOImpl, provides=PostDAO) + alias(source=PostDAOImpl, provides=PostReader) + decorate(SomeDecorator, provides=SomeClass) + from_context(Config) ) ``` -------------------------------- ### Install documentation requirements Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Install the necessary Python packages for building the documentation. ```default pip install -r requirements_doc.txt ``` -------------------------------- ### Install development dependencies Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Install the necessary development tools and the project itself in editable mode. ```default pip install --group dev uv pip install -e . ``` -------------------------------- ### Define Basic Classes Source: https://github.com/reagento/dishka/blob/develop/docs/provider/index.md Define simple classes that will be used as dependencies. These serve as examples for dependency injection setup. ```python class Connection: pass class DAO: def __init__(self, conn: Connection): pass ``` -------------------------------- ### Install Dishka Source: https://github.com/reagento/dishka/blob/develop/README.md Install the dishka library using pip. ```shell pip install dishka ``` -------------------------------- ### Build documentation with sphinx-build Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Compile the project documentation into HTML format using sphinx-build. Ensure 'make' is installed. ```default sphinx-build -M html docs docs-build -W ``` -------------------------------- ### FastAPI integration with Dishka Source: https://github.com/reagento/dishka/blob/develop/README.md Shows how to integrate Dishka with a FastAPI application using decorators and setup functions. This example includes setting up an async container, including providers, and injecting dependencies into route handlers. ```python from fastapi import APIRouter, FastAPI from dishka import make_async_container from dishka.integrations.fastapi import ( FastapiProvider, FromDishka, inject, setup_dishka, ) app = FastAPI() router = APIRouter() app.include_router(router) container = make_async_container( service_provider, ConnectionProvider(), FastapiProvider(), ) setup_dishka(container, app) @router.get("/") @inject async def index(service: FromDishka[Service]) -> str: ... ``` -------------------------------- ### Setup Dishka Integration with Taskiq Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/taskiq.md Call setup_dishka with the configured container and Taskiq broker to finalize the integration. ```python setup_dishka(container=container, broker=broker) ``` -------------------------------- ### dishka Async Container Setup with Request Scope Provider Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/testing/index.md Demonstrates setting up an asynchronous dishka container with a custom provider for database connections that are scoped to the request. This setup is for production. ```python from collections.abc import Iterable from sqlite3 import connect, Connection from dishka import Provider, Scope, provide, make_async_container from dishka.integrations.fastapi import setup_dishka class ConnectionProvider(Provider): def __init__(self, uri): super().__init__() self.uri = uri @provide(scope=Scope.REQUEST) def get_connection(self) -> Iterable[Connection]: conn = connect(self.uri) yield conn conn.close() container = make_async_container(ConnectionProvider("sqlite:///")) setup_dishka(container, app) ``` -------------------------------- ### Parameter Injection Example Source: https://github.com/reagento/dishka/blob/develop/docs/di_intro.md Demonstrates injecting a dependency (Client) as a parameter to a method. This approach works but can become cumbersome with many methods or complex call chains. ```python class Service: def action(self, client: Client): client.get_data() service = Service() client = Client(token="1234567890") service.action(client) ``` -------------------------------- ### FastStream with Dishka Setup and Test Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md Demonstrates setting up a FastStream application with Dishka for dependency injection and includes a test case to verify handler functionality with injected dependencies. ```python from collections.abc import AsyncIterator import pytest from dishka import AsyncContainer, make_async_container from dishka import Provider, Scope, provide from dishka.integrations import faststream as faststream_integration from dishka.integrations.base import FromDishka as Depends from faststream import FastStream, TestApp from faststream.rabbit import RabbitBroker, TestRabbitBroker, RabbitRouter router = RabbitRouter() @router.subscriber("test-queue") async def handler(msg: str, some_dependency: Depends[int]) -> int: print(f"{msg=}") return some_dependency @pytest.fixture async def broker() -> RabbitBroker: broker = RabbitBroker() broker.include_router(router) return broker @pytest.fixture def mock_provider() -> Provider: class MockProvider(Provider): @provide(scope=Scope.REQUEST) async def get_some_dependency(self) -> int: return 42 return MockProvider() @pytest.fixture def container(mock_provider: Provider) -> AsyncContainer: return make_async_container(mock_provider) @pytest.fixture async def app(broker: RabbitBroker, container: AsyncContainer) -> FastStream: app = FastStream(broker) faststream_integration.setup_dishka(container, app, auto_inject=True) return FastStream(broker) @pytest.fixture async def client(app: FastStream) -> AsyncIterator[RabbitBroker]: async with TestRabbitBroker(app.broker) as br, TestApp(app): yield br @pytest.mark.asyncio async def test_handler(client: RabbitBroker) -> None: result = await client.request("hello", "test-queue") assert await result.decode() == 42 ``` -------------------------------- ### Setup Dishka Integration Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/litestar.md Call setup_dishka to integrate Dishka with the Litestar application. ```python setup_dishka(container=container, app=app) ``` -------------------------------- ### Install dishka-faststream Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md Install the dishka-faststream package using pip. This package provides the integration for FastStream. ```shell pip install dishka-faststream ``` -------------------------------- ### Configure Provider with Methods Source: https://github.com/reagento/dishka/blob/develop/docs/provider/index.md Configure a Provider by directly calling its methods to register factories and dependencies. This approach is suitable for straightforward dependency setups. ```python from dishka import make_container, Provider, Scope def get_connection() -> Iterable[Connection]: conn = connect(uri) yield conn conn.close() provider = Provider(scope=Scope.APP) provider.provide(get_connection) provider.provide(DAO) container = make_container(provider) ``` -------------------------------- ### Run RabbitMQ Docker Container Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md This command starts a RabbitMQ container with management UI enabled. Ensure RabbitMQ is running before proceeding with the FastStream examples. ```shell docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USER=guest \ -e RABBITMQ_DEFAULT_PASS=guest \ rabbitmq:management ``` -------------------------------- ### Basic Provider Setup Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide.md Defines a provider for an external API client, specifying the implementation, the interface it provides, and its scope. Use this to register services with Dishka. ```python from dishka import provide, Provider, Scope class ExternalAPIClient(Protocol): ... class ExternalAPIClientImpl(UserDAO): def __init__(self, settings: APISettings): ... class MyProvider(Provider): external_api_client = provide( ExternalAPIClientImpl, provides=ExternalAPIClient, scope=Scope.REQUEST, recursive=True ) ``` -------------------------------- ### Install Starlette Integration Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/starlette.md Install the starlette-dishka package using pip. This is the recommended way to use the integration. ```shell pip install starlette-dishka ``` -------------------------------- ### Application Factory Pattern for Production and Testing Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/testing/index.md Illustrates separating application creation from container setup using factory functions. This allows for different container configurations for production and testing environments. ```python from fastapi import FastAPI from dishka import make_async_container from dishka.integrations.fastapi import setup_dishka def create_app() -> FastAPI: app = FastAPI() app.include_router(router) return app def create_production_app(): app = create_app() container = make_async_container(ConnectionProvider("sqlite:///")) setup_dishka(container, app) return app ``` -------------------------------- ### Setup dishka integration with the container and bot instance Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/telebot.md Call setup_dishka to initialize the integration, passing your configured Dishka container and the pyTelegramBotAPI bot instance. ```python setup_dishka(container=container, bot=bot) ``` -------------------------------- ### Constructor Injection Example Source: https://github.com/reagento/dishka/blob/develop/docs/di_intro.md Illustrates constructor injection, where a dependency (Client) is passed to the service's constructor. This is the primary and recommended way to implement DI, making dependencies readily available to all methods. ```python class Service: def __init__(self, client: Client): self.client = client def action(self): self.client.get_data() token = os.getenv("TOKEN") client = Client(token) service = Service(client) service.action() ``` -------------------------------- ### Service without DI Source: https://github.com/reagento/dishka/blob/develop/docs/di_intro.md This example shows a service creating its own dependency (Client) internally, which leads to several issues related to configuration, testing, and reuse. ```python class Service: def action(self): client = Client(token) client.get_data() service = Service() service.action() ``` -------------------------------- ### FastAPI Integration and Testing Setup Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/testing/index.md This snippet shows how to set up a FastAPI application with Dishka for dependency injection and how to create a test client for integration testing. It includes defining a router, an endpoint that injects a dependency, and functions to create production and test applications. ```python from collections.abc import Iterable from sqlite3 import Connection, connect from unittest.mock import Mock import pytest import pytest_asyncio from fastapi import APIRouter, FastAPI from fastapi.testclient import TestClient from dishka import Provider, Scope, make_async_container, provide from dishka.integrations.fastapi import FromDishka, inject, setup_dishka router = APIRouter() @router.get("/") @inject async def index(connection: FromDishka[Connection]) -> str: connection.execute("select 1") return "Ok" def create_app() -> FastAPI: app = FastAPI() app.include_router(router) return app class ConnectionProvider(Provider): def __init__(self, uri): super().__init__() self.uri = uri @provide(scope=Scope.REQUEST) def get_connection(self) -> Iterable[Connection]: conn = connect(self.uri) yield conn conn.close() def create_production_app(): app = create_app() container = make_async_container(ConnectionProvider("sqlite:///")) setup_dishka(container, app) return app class MockConnectionProvider(Provider): @provide(scope=Scope.APP) def get_connection(self) -> Connection: connection = Mock() connection.execute = Mock(return_value="1") return connection @pytest_asyncio.fixture async def container(): container = make_async_container(MockConnectionProvider()) yield container await container.close() @pytest.fixture def client(container): app = create_app() setup_dishka(container, app) with TestClient(app) as client: yield client @pytest_asyncio.fixture async def connection(container): return await container.get(Connection) @pytest.mark.asyncio async def test_controller(client: TestClient, connection: Mock): response = client.get("/") assert response.status_code == 200 connection.execute.assert_called() ``` -------------------------------- ### Run all tests with nox Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Use nox to run all types of tests, including unit, integration, and example app tests. ```default nox ``` -------------------------------- ### Dependency Creation with a Custom Container Source: https://github.com/reagento/dishka/blob/develop/docs/di_intro.md This example demonstrates using a custom container class to manage dependency creation. It allows for lazy instantiation of dependencies, creating them only when needed within a request handler. ```python from dishka import Client, Service import os class Container: def get_client(self) -> Client: return Client(os.getenv("TOKEN")) def get_service(self) -> Service: return Service(self.get_client()) container = Container() def container_middleware(request): request.state.container = container app.setup_middleware(container_middleware) @app.get("/") def index(request): service = request.state.container.get_service() service.action() ``` -------------------------------- ### Setup Dishka with Auto-Injection Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/click.md Configure the Click context with a Dishka container. Set `auto_inject=True` for automatic injection of dependencies into command handlers. ```python @click.group() @click.pass_context def main(context: click.Context): container = make_container(MyProvider()) setup_dishka(container=container, context=context, auto_inject=True) ``` -------------------------------- ### FastAPI Service with dishka Integration Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/testing/index.md Example of a FastAPI service using dishka for dependency injection, demonstrating how to inject a database connection into a route handler. ```python from sqlite3 import Connection from fastapi import FastAPI, APIRouter from dishka.integrations.fastapi import FromDishka, inject router = APIRouter() @router.get("/") @inject async def index(connection: FromDishka[Connection]) -> str: connection.execute("select 1") return "Ok" app = FastAPI() app.include_router(router) ``` -------------------------------- ### Manual Dependency Wiring in Web Handlers Source: https://github.com/reagento/dishka/blob/develop/docs/di_intro.md This example shows direct dependency creation within request handlers, which can lead to code duplication and complexity in larger applications. ```python from dishka import Client, Service import os @app.get("/") def index(request): client = Client(os.getenv("TOKEN")) service = Service(client) service.action() @app.get("/foo") def get_foo(request): client = Client(os.getenv("TOKEN")) service = Service(client) service.action() ``` -------------------------------- ### Setup Dishka with a custom inject decorator Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/flask.md Alternatively, provide your own inject decorator to `setup_dishka` if you have custom injection logic or prefer a different decorator name. ```python setup_dishka(container=container, app=app, auto_inject=my_inject) ``` -------------------------------- ### FastStream + Litestar Integration with Dishka Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md Example demonstrating the integration of FastStream with Litestar using Dishka for dependency injection. It sets up a RabbitMQ broker, an HTTP handler, and an AMQP subscriber, all utilizing a shared dependency. ```python import uvicorn from dishka import Provider, Scope, provide from dishka import make_async_container from dishka.integrations import faststream as faststream_integration from dishka.integrations import litestar as litestar_integration from dishka.integrations.base import FromDishka from dishka.integrations.faststream import inject as faststream_inject from dishka.integrations.litestar import inject as litestar_inject from faststream.rabbit import RabbitBroker, RabbitRouter from litestar import Litestar, route, HttpMethod class SomeDependency: async def do_something(self) -> int: print("Hello world") return 42 class SomeProvider(Provider): @provide(scope=Scope.REQUEST) def some_dependency(self) -> SomeDependency: return SomeDependency() @route(http_method=HttpMethod.GET, path="/", status_code=200) @litestar_inject async def http_handler(some_dependency: FromDishka[SomeDependency]) -> None: await some_dependency.do_something() amqp_router = RabbitRouter() @amqp_router.subscriber("test-queue") @faststream_inject async def amqp_handler(some_dependency: FromDishka[SomeDependency]) -> None: await some_dependency.do_something() def create_app() -> Litestar: container = make_async_container(SomeProvider()) broker = RabbitBroker(url="amqp://guest:guest@localhost:5672/") broker.include_router(amqp_router) faststream_integration.setup_dishka(container, broker=broker) http = Litestar( route_handlers=[http_handler], on_startup=[broker.start], on_shutdown=[broker.stop], ) litestar_integration.setup_dishka(container, http) return http if __name__ == "__main__": uvicorn.run(create_app(), host="0.0.0.0", port=8000) ``` -------------------------------- ### Setup Dishka Integration with Custom Inject Decorator Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md Alternatively, provide a custom inject decorator to `setup_dishka` if not using the default auto-injection. ```python setup_dishka(container=container, broker=broker, auto_inject=my_inject) ``` -------------------------------- ### Setup Lifespan for Container Closing Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/litestar.md Configure the application lifespan to ensure the Dishka container is closed upon app termination. ```python @asynccontextmanager async def lifespan(app: Litestar): yield await app.state.dishka_container.close() app = Litestar([endpoint], lifespan=[lifespan]) ``` -------------------------------- ### Create Providers with Scopes Source: https://github.com/reagento/dishka/blob/develop/README.md Set up Dishka providers to define how dependencies are created. Use Scope.REQUEST for per-request dependencies and Scope.APP for application-wide singletons. This example includes a custom provider for a releasable Connection. ```python import sqlite3 from collections.abc import Iterable from sqlite3 import Connection from dishka import Provider, Scope, provide service_provider = Provider(scope=Scope.REQUEST) service_provider.provide(Service) service_provider.provide(SQLiteUserDAO, provides=UserDAO) service_provider.provide(APIClient, scope=Scope.APP) # override provider's scope class ConnectionProvider(Provider): @provide(scope=Scope.REQUEST) def new_connection(self) -> Iterable[Connection]: connection = sqlite3.connect(":memory:") yield connection connection.close() ``` -------------------------------- ### FastStream + FastAPI Integration with Dishka Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md This example shows how to integrate FastStream with FastAPI using Dishka for dependency injection. It configures a RabbitMQ broker, an HTTP endpoint, and an AMQP consumer, all sharing a common dependency. ```python from collections.abc import AsyncIterator from contextlib import asynccontextmanager import uvicorn from fastapi import APIRouter, FastAPI from faststream.rabbit import RabbitBroker, RabbitRouter from dishka import Provider, Scope, make_async_container, provide from dishka.integrations import fastapi as fastapi_integration from dishka.integrations import faststream as faststream_integration from dishka.integrations.base import FromDishka from dishka.integrations.fastapi import DishkaRoute from dishka.integrations.faststream import inject as faststream_inject class SomeDependency: async def do_something(self) -> int: print("Hello world") return 42 class SomeProvider(Provider): @provide(scope=Scope.REQUEST) def some_dependency(self) -> SomeDependency: return SomeDependency() router = APIRouter(route_class=DishkaRoute) @router.get("/") async def http_handler(some_dependency: FromDishka[SomeDependency]) -> None: await some_dependency.do_something() amqp_router = RabbitRouter() @amqp_router.subscriber("test-queue") @faststream_inject async def amqp_handler(some_dependency: FromDishka[SomeDependency]) -> None: await some_dependency.do_something() def create_app() -> FastAPI: container = make_async_container(SomeProvider()) broker = RabbitBroker(url="amqp://guest:guest@localhost:5672/") broker.include_router(amqp_router) faststream_integration.setup_dishka(container, broker=broker) @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: async with broker: await broker.start() yield http = FastAPI(lifespan=lifespan) http.include_router(router) fastapi_integration.setup_dishka(container, http) return http if __name__ == "__main__": uvicorn.run(create_app(), host="0.0.0.0", port=8000) ``` -------------------------------- ### Import necessary components from Dishka aiogram integration Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/aiogram.md Import the required classes and functions for setting up Dishka with aiogram. This includes providers, decorators, and setup utilities. ```python from dishka.integrations.aiogram import ( AiogramProvider, FromDishka, inject, setup_dishka, ) from dishka import make_async_container, Provider, provide, Scope ``` -------------------------------- ### Setup Dishka with FastAPI (Sync Container) Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/fastapi.md Configure Dishka for use with a synchronous container in FastAPI. This includes setting up the container, including providers, and integrating with the FastAPI application lifecycle. ```python from dishka.integrations.fastapi import ( FromDishka, FastapiProvider, inject_sync, setup_dishka, ) from dishka import make_container, Provider, provide, Scope router = APIRouter() @router.get('/') @inject_sync def endpoint( gateway: FromDishka[Gateway], ) -> ResponseModel: ... @asynccontextmanager async def lifespan(app: FastAPI): yield app.state.dishka_container.close() app = FastAPI(lifespan=lifespan) app.include_router(router) container = make_container(YourProvider(), FastapiProvider()) setup_dishka(container=container, app=app) ``` -------------------------------- ### Setup Dishka Interceptor for gRPC Server Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/grpcio.md Configure the gRPC server with Dishka interceptors. Use DishkaInterceptor for synchronous services and DishkaAioInterceptor for asynchronous services. ```python from concurrent.futures import ThreadPoolExecutor from dishka.integrations.grpcio import DishkaInterceptor from grpc import make_server server = make_server( ThreadPoolExecutor(max_workers=10), interceptors=[ DishkaInterceptor(container), ], ) ``` -------------------------------- ### Define Classes with Type Hints Source: https://github.com/reagento/dishka/blob/develop/README.md Define service and dependency classes using Python type hints. This example shows a Service class with APIClient and UserDAO dependencies, where UserDAO is implemented by SQLiteUserDAO. ```python from sqlite3 import Connection from typing import Protocol class APIClient: ... class UserDAO(Protocol): ... class SQLiteUserDAO(UserDAO): def __init__(self, connection: Connection): ... class Service: def __init__(self, client: APIClient, user_dao: UserDAO): ... ``` -------------------------------- ### Configure Validation Settings Source: https://github.com/reagento/dishka/blob/develop/docs/errors.md Disable validation during container creation by passing `skip_validation=True` or configure specific validation options using `ValidationSettings`. This example enables all currently possible validations. ```python from dishka import make_container, ValidationSettings settings = ValidationSettings( nothing_overridden = True, implicit_override = True, nothing_decorated = True, ) container = make_container(provider, validation_settings=settings) ``` -------------------------------- ### Setup Dishka integration with aiogram dispatcher Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/aiogram.md Initialize the Dishka integration with your aiogram dispatcher using `setup_dishka`. Set `auto_inject=True` for automatic injection or provide a custom inject decorator. ```python setup_dishka(container=container, router=dp, auto_inject=True) ``` ```python setup_dishka(container=container, router=dp, auto_inject=my_inject) ``` -------------------------------- ### Setup Dishka with Custom Inject Decorator Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/click.md Configure the Click context with a Dishka container and a custom inject decorator. This allows for more control over the injection process. ```python @click.group() @click.pass_context def main(context: click.Context): container = make_container(MyProvider()) setup_dishka(container=container, context=context, auto_inject=my_inject) ``` -------------------------------- ### Isolating Providers with Components for Different Databases Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/components.md Illustrates how components isolate providers, preventing implicit requests across components. This example shows distinct `DBConnection` implementations for 'user' and 'comment' components. ```python from dishka import Provider, Scope, make_container, provide from typing import Protocol class DBConnection(Protocol): ... class UserDBConnection(DBConnection): ... class CommentDBConnection(DBConnection): ... class UserDAO: def __init__(self, db: DBConnection): ... class CommentDAO: def __init__(self, db: DBConnection): ... class UserProvider(Provider): component = "user" scope = Scope.APP # should be REQUEST, but set to APP for the sake of simplicity db_connection = provide(UserDBConnection, provides=DBConnection) dao = provide(UserDAO) class CommentProvider(Provider): component = "comment" scope = Scope.APP # should be REQUEST, but set to APP for the sake of simplicity db_connection = provide(CommentDBConnection, provides=DBConnection) dao = provide(CommentDAO) container = make_container(UserProvider(), CommentProvider()) container.get(DBConnection, component="user") # UserDBConnection container.get(DBConnection, component="comment") # CommentDBConnection ``` -------------------------------- ### Create and activate virtual environment Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Set up a Python virtual environment for development and activate it. ```default python3 -m venv .venv source .venv/bin/activate ``` -------------------------------- ### Create and Use a Container Source: https://github.com/reagento/dishka/blob/develop/docs/concepts.md Demonstrates creating an application container, retrieving an APP-scoped dependency, and then entering a REQUEST scope to retrieve both REQUEST and APP-scoped dependencies. ```python app_container = make_container(provider1, provider2) # enter APP scope config = app_container.get(Config) # APP-scoped object with app_container() as request_container: # enter REQUEST scope connection = request_container.get(Connection) # REQUEST-scoped object config = request_container.get(Config) # APP-scoped object ``` -------------------------------- ### Create Providers and Specify Dependencies Source: https://github.com/reagento/dishka/blob/develop/docs/quickstart.md Set up providers to define how dependencies are created. Use Scope.APP for application-lifetime dependencies and Scope.REQUEST for per-request dependencies. Custom factories can be used for resources requiring finalization, like database connections. ```python import sqlite3 from collections.abc import Iterable from sqlite3 import Connection from dishka import Provider, Scope, provide service_provider = Provider(scope=Scope.REQUEST) service_provider.provide(Service) service_provider.provide(SQLiteUserDAO, provides=UserDAO) service_provider.provide(APIClient, scope=Scope.APP) # override provider's scope class ConnectionProvider(Provider): @provide(scope=Scope.REQUEST) def new_connection(self) -> Iterable[Connection]: connection = sqlite3.connect(":memory:") yield connection connection.close() ``` -------------------------------- ### Setup Dishka Integration with Celery App Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/celery.md Initialize the Dishka integration with your Celery application and the created container. ```python setup_dishka(container=container, app=celery_app) ``` -------------------------------- ### FastAPI Integration with Dishka Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/index.md Demonstrates how to set up and use Dishka with FastAPI. Ensure `setup_dishka` is called with a configured container and the FastAPI app. ```python from dishka.integrations.fastapi import FromDishka, inject, setup_dishka, FastapiProvider @router.get("/") @inject async def index(interactor: FromDishka[Interactor]) -> str: result = interactor() return result app = FastAPI() container = make_async_container(your_provider, FastapiProvider()) setup_dishka(container, app) ``` -------------------------------- ### Provider with Initialization and Configuration Source: https://github.com/reagento/dishka/blob/develop/docs/provider/index.md Create a custom Provider class that accepts configuration in its constructor. This is useful for passing settings like database URIs to your dependency factories. ```python class MyProvider(Provider): def __init__(self, uri: str, scope: Scope): super().__init__(scope=scope) # do not forget `super` self.uri = uri @provide def get_connection(self) -> Iterable[Connection]: conn = connect(self.uri) # use passed configuration yield conn conn.close() dao = provide(DAO) provider = MyProvider(uri=os.getenv("DB_URI"), scope=Scope.APP) container = make_container(provider) ``` -------------------------------- ### Setup Signal for Container Closure Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/celery.md Connect a signal handler to close the Dishka container when a Celery worker process shuts down. ```python from celery import current_app from celery.signals import worker_process_shutdown from dishka import Container @worker_process_shutdown.connect() def close_dishka(*args, **kwargs): container: Container = current_app.conf["dishka_container"] container.close() ``` -------------------------------- ### Provide a Class Instance Directly Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide.md If you just need to create an instance of a class using its constructor, you can pass the class directly to the `provide` function as an attribute of your `Provider` class. ```python from dishka import provide, Provider, Scope class MyProvider(Provider): service = provide(Service, scope=Scope.REQUEST) ``` -------------------------------- ### Setup Dishka with ARQ Worker Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/arq.md Configure Dishka integration with your ARQ Worker or WorkerSettings by providing the container and worker settings. ```python setup_dishka(container=container, worker_settings=WorkerSettings) ``` -------------------------------- ### NoActiveFactoryError Example Source: https://github.com/reagento/dishka/blob/develop/docs/errors.md This error occurs when no active factory can be selected for a component, meaning all provided variants are inactive. Check the logic of marker activation. ```default dishka.exceptions.NoActiveFactoryError: Cannot select active factory for (Cache, component=''). All variants are not active. │ ◈ Scope.REQUEST, component='' ◈ ▼ __main__.Service MyProvider.service ╰─> __main__.Cache select ╰─× MyProvider.redis_cache: Has(RedisCache) ╰─× MyProvider.cache: Marker("a") ``` -------------------------------- ### Create Container with FastStreamProvider Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md Instantiate an async container, including `FastStreamProvider` if your providers use `StreamMessage` or `ContextRepo`. ```python container = make_async_container(YourProvider(), FastStreamProvider()) ``` -------------------------------- ### Register Multiple Providers using provide_all Method Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide_all.md The provide_all functionality is also available as a method on a Provider instance. This offers an alternative way to group provider registrations. ```python provider = Provider(scope=Scope.APP) provider.provide_all( RegisterUserInteractor, UpdateProfilePicInteractor ) ``` -------------------------------- ### Combining Provider and Source `when` Conditions Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/when.md When both the provider and an individual source have `when=` conditions, they are combined using AND logic. This allows for more granular control over factory activation. ```python from dishka import Provider, provide, Scope, Has class FeatureProvider(Provider): when = Marker("prod") # prerequisite scope = Scope.APP @provide(when=Has(RedisConfig)) # additional condition def redis_cache(self, config: RedisConfig) -> Cache: return RedisCache(config) # Effective: Marker("prod") & Has(RedisConfig) ``` -------------------------------- ### Create container with SanicProvider Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/sanic.md Instantiate the container using `make_async_container`, including `SanicProvider()` if you intend to use `sanic.Request` within your providers. ```python container = make_async_container(YourProvider(), SanicProvider()) ``` -------------------------------- ### Decorate handlers with @inject Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/flask.md Optionally, decorate your handler functions with the `@inject` decorator to explicitly enable dependency injection. This is an alternative to relying solely on `auto_inject=True` during setup. ```python @router.get('/') @inject async def endpoint( gateway: FromDishka[Gateway], ) -> Response: ... ``` -------------------------------- ### Defining and Using Components in Dishka Providers Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/components.md Demonstrates how to define components for providers, either at the class level or during instance creation. A provider can also be cast to use a different component. ```python from dishka import make_container, Provider # default component is used when not specified provider0 = Provider() class MyProvider(Provider): # component can be set in class component = "component_name" provider1 = MyProvider() # component can be set on instance creation provider2 = MyProvider(component="other") # same provider instance is casted to use with different component provider3 = provider2.to_component("additional") container = make_container(provider0, provider1, provider2, provider3) ``` -------------------------------- ### Import Dishka Click Components Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/click.md Import the necessary components from dishka and dishka.integrations.click to enable dependency injection. ```python from dishka import make_container from dishka.integrations.click import setup_dishka, inject ``` -------------------------------- ### Setup FastAPI lifespan for container management Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/fastapi.md Implement a lifespan context manager in your FastAPI app to ensure the Dishka container is properly closed upon application termination. ```python @asynccontextmanager async def lifespan(app: FastAPI): yield await app.state.dishka_container.close() app = FastAPI(lifespan=lifespan) ``` -------------------------------- ### Get Dependency from Another Component Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/components.md Demonstrates retrieving a dependency from a specific component using `FromComponent` annotation. This is useful when a factory needs a dependency that resides in a different component. ```python from typing import Annotated from dishka import FromComponent, make_container, Provider, provide, Scope class MainProvider(Provider): @provide(scope=Scope.APP) def foo(self, a: Annotated[int, FromComponent("X")]) -> float: return a / 10 class AdditionalProvider(Provider): component = "X" @provide(scope=Scope.APP) def foo(self) -> int: return 1 container = make_container(MainProvider(), AdditionalProvider()) container.get(float) # returns 0.1 ``` -------------------------------- ### Using wrap_injection with Custom Scopes Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/adding_new.md Demonstrates how to use `wrap_injection` with specific scopes like `Scope.STEP` and `Scope.ACTION` for targeted dependency injection. ```python from dishka import inject, Scope from dishka.dependency_source import FromDishka @inject(scope=Scope.STEP) async def handler(step_dep: StepDep = FromDishka[StepDep]): ... @inject(scope=Scope.ACTION) def process_action(action_dep: ActionDep = FromDishka[ActionDep]): ... ``` -------------------------------- ### Provider with Generic Classes Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide.md Shows how to use a provider factory with generic classes. The factory can accept a type parameter `type_` to create instances of generic types. ```python from dishka import Provider, provide class MyProvider(Provider): @provide def make_a(self, type_: type[T]) -> A[T]: ... ``` -------------------------------- ### ImplicitOverrideDetectedError Example Source: https://github.com/reagento/dishka/blob/develop/docs/errors.md This error is raised when multiple factories are detected for the same type without the `override` flag set, and `implicit_override` is enabled. Consider setting `override=True` or removing one of the factories. ```default dishka.exceptions.ImplicitOverrideDetectedError: Detected multiple factories for (, component='') while `override` flag is not set. Hint: * Try specifying `override=True` for SecondProvider.get_a * Try removing factory FirstProvider.get_a or SecondProvider.get_a ``` -------------------------------- ### Import Taskiq Integration Components Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/taskiq.md Import necessary components from dishka.integrations.taskiq and dishka for setting up the integration. ```python from dishka.integrations.taskiq import ( FromDishka, inject, setup_dishka, TaskiqProvider, ) from dishka import make_async_container, Provider, provide, Scope ``` -------------------------------- ### Import Dishka FastStream Components Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/faststream.md Import necessary components from dishka.integrations.faststream and dishka for setting up dependency injection. ```python from dishka.integrations.faststream import ( FromDishka, inject, setup_dishka, FastStreamProvider, ) from dishka import make_async_container, Provider, provide, Scope ``` -------------------------------- ### IndependentDecoratorError Example Source: https://github.com/reagento/dishka/blob/develop/docs/errors.md This error indicates that a decorator does not depend on a provided type, suggesting it might be intended as a simple `provide` instead. Use `decorate` only when modifying an object created by another factory. ```default dishka.provider.exceptions.IndependentDecoratorError: Decorator __main__.FirstProvider.get_a does not depend on provided type. Did you mean @provide instead of @decorate? ``` -------------------------------- ### Define and Use Custom Scopes Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/scopes.md Demonstrates how to define custom scopes by inheriting from `BaseScope` and providing them to `make_container`. This allows for tailored dependency management strategies. ```python from dishka import BaseScope, Provider, make_container, new_scope class MyScope(BaseScope): APPLICATION = new_scope("APPLICATION") SESSION = new_scope("SESSION", skip=True) EVENT = new_scope("EVENT") provider = Provider(scope=MyScope.EVENT) make_container(provider, scopes=MyScope) ``` -------------------------------- ### Setup Dishka integration with Flask app Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/flask.md Call `setup_dishka` with your container, Flask app, and `auto_inject=True` to enable automatic dependency injection for handlers. Ensure this is called after registering all views and blueprints. ```python setup_dishka(container=container, app=app, auto_inject=True) ``` -------------------------------- ### Create Synchronous Container Source: https://github.com/reagento/dishka/blob/develop/docs/container/index.md Use `make_container` to create a synchronous container. Pass one or more providers to it. ```python from dishka import make_container container = make_container(provider) ``` -------------------------------- ### Import necessary components from dishka.integrations.aiohttp Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/aiohttp.md Import the required classes and functions for setting up Dishka with aiohttp. ```python from dishka.integrations.aiohttp import ( DISHKA_CONTAINER_KEY, FromDishka, inject, setup_dishka, AiohttpProvider, ) from dishka import make_async_container, Provider, provide, Scope ``` -------------------------------- ### Clone the dishka repository Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Clone the project repository to your local machine and navigate into the project directory. ```default git clone git@github.com:reagento/dishka.git cd dishka ``` -------------------------------- ### Attribute Injection Example Source: https://github.com/reagento/dishka/blob/develop/docs/di_intro.md Shows attribute injection, where a dependency (Client) is assigned directly to an attribute of the service object after its creation. This is often used in conjunction with constructor injection or to break circular dependencies. ```python class Service: client: Client def action(self): self.client.get_data() service = Service() service.client = Client(token) service.action() ``` -------------------------------- ### Run integration tests for aiohttp with nox Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Execute integration tests specifically for the aiohttp library using nox. ```default nox -t aiohttp ``` -------------------------------- ### Create an async container with FastapiProvider Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/fastapi.md Instantiate an `AsyncContainer` and include `FastapiProvider` to enable the use of `fastapi.Request` and `fastapi.WebSocket` within providers. ```python container = make_async_container(YourProvider(), FastapiProvider()) ``` -------------------------------- ### dishka Test Controller with Mocked Dependencies Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/testing/index.md Example of a pytest test case for a FastAPI controller that uses dishka. It verifies the controller's response and checks if a mocked dependency (database connection) was called. ```python from unittest.mock import Mock async def test_controller(client: TestClient, connection: Mock): response = client.get("/") assert response.status_code == 200 connection.execute.assertCalled() ``` -------------------------------- ### Create Container with StarletteProvider Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/starlette.md Instantiate an async container, including `StarletteProvider` if you intend to use `starlette.Request` or `starlette.WebSocket` within your providers. ```python container = make_async_container(YourProvider(), StarletteProvider()) ``` -------------------------------- ### NothingOverriddenError Example Source: https://github.com/reagento/dishka/blob/develop/docs/errors.md This error occurs when an overriding factory is found but there is nothing to override, often due to incorrect provider order or missing factories. Ensure providers are correctly specified or remove the `override=True` flag. ```default dishka.exceptions.NothingOverriddenError: Overriding factory found for (, component=''), but there is nothing to override. Hint: * Try removing override=True from FirstProvider.get_a * Check the order of providers ``` -------------------------------- ### Run integration tests with latest aiohttp version Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Run integration tests for aiohttp using the latest available version of the library by specifying it explicitly. ```default nox -s aiohttp_latest ``` -------------------------------- ### Create a Custom Provider Source: https://github.com/reagento/dishka/blob/develop/docs/concepts.md Inherit from the `Provider` class to create custom dependency providers. Instantiate your provider when creating a container. ```python class MyProvider(Provider): pass container = make_container(MyProvider()) ``` -------------------------------- ### Configuring Collection Options Source: https://github.com/reagento/dishka/blob/develop/docs/advanced/collect.md Customize the collection behavior using options like `provides`, `scope`, and `cache`. Note that `provides` affects dependency resolution, not the creation of the list itself. ```python handlers = collect( Handler, provides=Sequence[Handler], scope=Scope.REQUEST, cache=True, ) ``` -------------------------------- ### Provide a Child Class for a Parent Interface Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide.md Use the `source` and `provides` arguments in `provide` to register a specific implementation class for a requested parent interface. ```python from dishka import provide, Provider, Scope class UserDAO(Protocol): ... class UserDAOImpl(UserDAO): ... class MyProvider(Provider): user_dao = provide(source=UserDAOImpl, scope=Scope.REQUEST, provides=UserDAO) ``` -------------------------------- ### Configure Provider with Inheritance Source: https://github.com/reagento/dishka/blob/develop/docs/provider/index.md Configure a Provider by inheriting from the base Provider class and using decorators or direct assignments for dependency registration. This allows for a more organized and class-based approach to dependency management. ```python from dishka import make_container, Provider, provide, Scope class MyProvider(Provider): @provide def get_connection(self) -> Iterable[Connection]: conn = connect(uri) yield conn conn.close() dao = provide(DAO) container = make_container(MyProvider(scope=Scope.APP)) ``` -------------------------------- ### Import ARQ Integration Components Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/arq.md Import necessary components from dishka.integrations.arq for setting up dependency injection. ```python from dishka.integrations.arq import ( FromDishka, inject, setup_dishka, ) ``` -------------------------------- ### Run ast-grep linter Source: https://github.com/reagento/dishka/blob/develop/docs/contributing.md Use ast-grep to scan the codebase for potential issues. ```default sg scan ``` -------------------------------- ### Configure Container with TaskiqProvider Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/taskiq.md Include TaskiqProvider when creating the Dishka container if TaskiqMessage is used in providers. ```python container = make_async_container(YourProvider(), TaskiqProvider()) ``` -------------------------------- ### Define Providers with `from_context` Source: https://github.com/reagento/dishka/blob/develop/docs/provider/from_context.md Use `from_context` to mark dependencies that should be retrieved from the scope's context. Set the `context=` argument when entering the scope to provide these values. ```python from dishka import from_context, Provider, provide, Scope class MyProvider(Provider): scope = Scope.REQUEST app = from_context(provides=App, scope=Scope.APP) request = from_context(provides=RequestClass) @provide def get_a(self, request: RequestClass, app: App) -> A: ... container = make_container(MyProvider(), context={App: app}) with container(context={RequestClass: request_instance}) as request_container: pass ``` -------------------------------- ### Import necessary modules for FastAPI integration Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/fastapi.md Import the required components from dishka.integrations.fastapi and dishka for setting up the integration. ```python from dishka.integrations.fastapi import ( DishkaRoute, FromDishka, FastapiProvider, inject, setup_dishka, ) from dishka import make_async_container, Provider, provide, Scope ``` -------------------------------- ### Create a Dishka Container Source: https://github.com/reagento/dishka/blob/develop/README.md Create a Dishka container by passing the configured providers. The container manages the cache and retrieval of dependencies. ```python from dishka import make_container container = make_container(service_provider, ConnectionProvider()) ``` -------------------------------- ### Overriding Provider Factories Source: https://github.com/reagento/dishka/blob/develop/docs/provider/provide.md Demonstrates how to override a default provider implementation with a mock or alternative implementation using `override=True`. This is useful for testing or specific configurations. ```python from dishka import provide, Provider, Scope, make_container class UserDAO(Protocol): ... class UserDAOImpl(UserDAO): ... class UserDAOMock(UserDAO): ... class MyProvider(Provider): scope = Scope.APP user_dao = provide(UserDAOImpl, provides=UserDAO) user_dao_mock = provide( UserDAOMock, provides=UserDAO, override=True ) container = make_container(MyProvider()) dao = container.get(UserDAO) # UserDAOMock ``` -------------------------------- ### Import Dishka Litestar Components Source: https://github.com/reagento/dishka/blob/develop/docs/integrations/litestar.md Import necessary components from dishka.integrations.litestar and dishka for integration. ```python from dishka.integrations.litestar import ( FromDishka, LitestarProvider, inject, setup_dishka, ) from dishka import make_async_container, Provider, provide, Scope ```