FastAPI + Redis example¶

This example shows how to useDependencyInjector withFastAPI andRedis.

The source code is available on theGithub.

Application structure¶

Application has next structure:

./├──fastapiredis/│├──__init__.py│├──application.py│├──containers.py│├──redis.py│├──services.py│└──tests.py├──docker-compose.yml├──Dockerfile└──requirements.txt

Redis¶

Moduleredis defines Redis connection pool initialization and shutdown. Seefastapiredis/redis.py:

fromtypingimportAsyncIteratorfromredis.asyncioimportfrom_url,Redisasyncdefinit_redis_pool(host:str,password:str)->AsyncIterator[Redis]:session=from_url(f"redis://{host}",password=password,encoding="utf-8",decode_responses=True)yieldsessionsession.close()awaitsession.wait_closed()

Service¶

Moduleservices contains example service. Service has a dependency on Redis connection pool.It uses it for getting and setting a key asynchronously. Real life service will do something more meaningful.Seefastapiredis/services.py:

"""Services module."""fromredis.asyncioimportRedisclassService:def__init__(self,redis:Redis)->None:self._redis=redisasyncdefprocess(self)->str:awaitself._redis.set("my-key","value")returnawaitself._redis.get("my-key")

Container¶

Declarative container wires example service with Redis connection pool. Seefastapiredis/containers.py:

"""Containers module."""fromdependency_injectorimportcontainers,providersfrom.importredis,servicesclassContainer(containers.DeclarativeContainer):config=providers.Configuration()redis_pool=providers.Resource(redis.init_redis_pool,host=config.redis_host,password=config.redis_password,)service=providers.Factory(services.Service,redis=redis_pool,)

Application¶

Moduleapplication createsFastAPI app, setup endpoint, and init container.

Endpointindex has a dependency on example service. The dependency is injected usingWiring feature.

Listing offastapiredis/application.py:

"""Application module."""fromtypingimportAnnotatedfromfastapiimportDepends,FastAPIfromdependency_injector.wiringimportProvide,injectfrom.containersimportContainerfrom.servicesimportServiceapp=FastAPI()@app.api_route("/")@injectasyncdefindex(service:Annotated[Service,Depends(Provide[Container.service])])->dict[str,str]:value=awaitservice.process()return{"result":value}container=Container()container.config.redis_host.from_env("REDIS_HOST","localhost")container.config.redis_password.from_env("REDIS_PASSWORD","password")container.wire(modules=[__name__])

Tests¶

Tests useProvider overriding feature to replace example service with a mock. Seefastapiredis/tests.py:

"""Tests module."""fromunittestimportmockimportpytestfromhttpximportASGITransport,AsyncClientfrom.applicationimportapp,containerfrom.servicesimportService@pytest.fixturedefclient(event_loop):client=AsyncClient(transport=ASGITransport(app=app),base_url="http://test",)yieldclientevent_loop.run_until_complete(client.aclose())@pytest.mark.asyncioasyncdeftest_index(client):service_mock=mock.AsyncMock(spec=Service)service_mock.process.return_value="Foo"withcontainer.service.override(service_mock):response=awaitclient.get("/")assertresponse.status_code==200assertresponse.json()=={"result":"Foo"}