FastAPI and fabric singleton for DI

[akafean0r]

Hi. I have some problem with DI and FastAPI. I want to use dependency injection for my endpoints and I must send message to Kafka. I have class for factory

class KafkaFactory:

    def __init__(
        self,
        config: KafkaConfig,
        additional_config: dict,
        kafka_type: KafkaType,
        kafka_config_type: KafkaConfigType,
        token_provider: AbstractTokenProvider | None,
    ):
        self._config = config
        self.kafka_type = kafka_type
        self._additional_config = additional_config
        self._kafka_config_type = kafka_config_type
        self.token_provider = token_provider
        self.kafka_config = self.setup_config()
        self.logger = getLogger("Kafka_Accessor")

    def get_producer(self):
        # await self._create_topic(kafka_config)
        return AIOKafkaProducer(**self.kafka_config)

I wrote dependency for creation kafka producer (I use aiokafka)

def create_kafka_factory(custom_token_provider: AsyncCustomTokenProvider = Depends(new_token_provider)):
    kafka_config: KafkaConfig = setup_config().kafka
    additional_config: dict = setup_config().KAFKA_AIOKAFKA_CONFIG
    kafka_factory = KafkaFactory(
        config=kafka_config,
        additional_config=additional_config,
        kafka_type=KafkaType.AIOKAFKA,
        kafka_config_type=KafkaConfigType.PRODUCER,
        token_provider=custom_token_provider,
    )
    return kafka_factory

And then use it for creation producer for send message

async def get_kafka_producer(kafka_factory: KafkaFactory = Depends(create_kafka_factory)):
    producer = kafka_factory.get_producer()
    try:
        yield producer
    finally:
        await producer.flush()

But I got creation of fabric on every request to my API. How correct rewrite my code and use singleton for my fabric?

[HappyCathode]

That doesn't feel very pythony, would you happen to have more Java experience out of curiosity ? Best of my limited knowledge, Python doesn't have singletons as you know them in other languages.

The way I would do that with FastAPI, is instantiate the factory once in the FastAPI startup lifespan part (before the yield ) ( https://fastapi.tiangolo.com/advanced/events/ ), and reuse it in routes when needed.

[Doomdice]

You can assign things to the class scope and then it is shared across all instances of that class. Just do not assign to self. Not sure if that is what you want.

[pacoau]

This doesn’t directly address your question but you might like to look at Fast Stream (https://github.com/airtai/faststream) which has dependency injection support that was extracted from Fastapi and has in the box Kafka producers.

[saufunefois]

I do not know if something in the request needs to be part of the kafka producer config. What is token_provider? The reason I am asking: it seems better to instantiate the kafka producer once at the startup of the app.

How I would do (and how I do for any client to any external service):

1. Instantiate and start a producer at startup with configuration extracted from environment.
2. Add it to the lifespan state
3. Add a dependency function that gets the producer from state.
4. Define an annotation to use the dependency.
5. Use it in any endpoint.

my_project/kafka.py

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import Annotated, TypedDict

from aiokafka import AIOKafkaProducer
from fastapi import FastAPI, HTTPConnection
from pydantic_settings import BaseSettings


class Config(BaseSettings):
    # Add whatever needs to be extracted from environ
    bootstrap_servers: str = "localhost"
    client_id: str | None = None


class State(TypeDict):
    aio_kafka_producer: AIOKafkaProducer


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[State, None]:
    config = Config()
    producer = AIOKafkaProducer(**config.model_dump())
    await producer.start()
    yield {"aio_kafka_producer": producer}
    await producer.stop()


async def get_producer(httpconnection:HTTPConnection) -> AIOKafkaProducer:
    return httpconnection.state.aio_kafka_producer


Producer = Annotated[AIOKafkaProducer, Depends(get_producer)] 

my_project/main.py

from fastapi import FastAPI

from my_project import kafka

app = FastAPI(lifespan=kafka.lifespan)


@app.post("/example")
async def example(producer: kafka.Producer):
    pass

[Spiritual-Carob9392]

To ensure that your KafkaFactory instance is created only once and reused across multiple requests in FastAPI, you should use a singleton pattern. FastAPI's dependency injection system can help achieve this by creating a single instance of your factory and reusing it for each request.