API Reference¶

@dataclass(frozen=True)
class TopicConfig:
    """Kafka topic configuration."""

    name: str
    num_partitions: int
    replication_factor: int
    replica_assignment: dict[int, list[int]] | None = None
    topic_configs: dict[str, str] | None = None

class KafkaSettingsProtocol(KafkaSaslSettingsProtocol, KafkaSslSettingsProtocol, Protocol):
    """Protocol for base Kafka settings (connection + SASL + SSL)."""

class ConsumerSettingsProtocol(KafkaSettingsProtocol):
    """Protocol for Kafka consumer settings."""

    group_id: str
    auto_offset_reset: KafkaOffsetReset
    enable_auto_commit: bool
    session_timeout_ms: int
    heartbeat_interval_ms: int
    max_poll_records: int
    max_poll_interval_ms: int
    fetch_max_wait_ms: int
    fetch_min_bytes: int
    fetch_max_bytes: int

class ProducerSettingsProtocol(KafkaSettingsProtocol):
    """Protocol for Kafka producer settings."""

    acks: KafkaAcks
    compression_type: KafkaCompressionType | None
    enable_idempotence: bool
    max_batch_size: int
    linger_ms: int
    request_timeout_ms: int

class TopicConfigProtocol(Protocol):
    """Protocol for topic configuration."""

    name: str
    num_partitions: int
    replication_factor: int
    replica_assignment: dict[int, list[int]] | None
    topic_configs: dict[str, str] | None

class KafkaConnectionSettingsProtocol(Protocol):
    """Protocol for core Kafka connection fields only."""

    bootstrap_servers: str
    client_id: str | None
    security_protocol: KafkaSecurityProtocol
    metadata_max_age_ms: int

    def get_sasl_password(self) -> str | None: ...

class KafkaSaslSettingsProtocol(KafkaConnectionSettingsProtocol, Protocol):
    """Protocol for SASL credentials. Relevant when ``security_protocol`` is ``SASL_*``."""

    sasl_mechanism: KafkaSaslMechanism | None
    sasl_username: str | None

class KafkaSslSettingsProtocol(KafkaConnectionSettingsProtocol, Protocol):
    """Protocol for TLS material. Relevant when ``security_protocol`` is ``SSL`` or ``SASL_SSL``."""

    ssl_cafile: str | None
    ssl_certfile: str | None
    ssl_keyfile: str | None
    ssl_check_hostname: bool

class ProducerLifecycleSettingsProtocol(ProducerSettingsProtocol):
    """Producer settings that also carry topic auto-creation policy.

    Used by lifecycle/DI helpers that need to know whether to call
    :func:`ensure_topics_async` before starting the producer.
    """

    auto_create_topics: bool

class BaseKafkaSettings(KafkaConnectionMixin, KafkaSaslMixin, KafkaSslMixin):
    """Base Kafka configuration shared between producer and consumer."""

    security_protocol: KafkaSecurityProtocol = Field(
        default="PLAINTEXT",
        description="Security protocol: PLAINTEXT, SASL_PLAINTEXT, SSL, SASL_SSL",
    )

    @model_validator(mode="after")
    def _validate_sasl(self) -> BaseKafkaSettings:
        if self.security_protocol not in {"SASL_PLAINTEXT", "SASL_SSL"}:
            return self

        if not self.sasl_mechanism:
            msg = f"Kafka sasl_mechanism is required for {self.security_protocol}"
            raise ValueError(msg)
        if not self.sasl_username:
            msg = f"Kafka sasl_username is required for {self.security_protocol}"
            raise ValueError(msg)
        if self.sasl_password is None or not self.sasl_password.get_secret_value():
            msg = f"Kafka sasl_password is required for {self.security_protocol}"
            raise ValueError(msg)

        return self

    @model_validator(mode="after")
    def _validate_ssl(self) -> BaseKafkaSettings:
        if self.security_protocol not in {"SSL", "SASL_SSL"}:
            return self

        if not self.ssl_cafile:
            msg = f"Kafka ssl_cafile is required for {self.security_protocol}"
            raise ValueError(msg)

        return self

class KafkaConnectionMixin(BaseModel):
    """Kafka transport connection parameters."""

    bootstrap_servers: str = Field(description="Kafka bootstrap servers (comma-separated)")
    client_id: str | None = Field(default=None, description="Client ID for Kafka")
    metadata_max_age_ms: int = Field(default=300000, ge=0, description="Metadata max age in ms")

class KafkaSaslMixin(BaseModel):
    """SASL credentials. Required when ``security_protocol`` selects SASL_*."""

    sasl_mechanism: KafkaSaslMechanism | None = Field(
        default=None,
        description="SASL mechanism: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512",
    )
    sasl_username: str | None = Field(default=None, description="SASL username")
    sasl_password: SecretStr | None = Field(default=None, description="SASL password")

    def get_sasl_password(self) -> str | None:
        """Plaintext SASL password for Kafka client config. Do NOT log."""
        if self.sasl_password is None:
            return None
        secret: str = self.sasl_password.get_secret_value()
        return secret

class KafkaSslMixin(BaseModel):
    """TLS material. Required when ``security_protocol`` selects SSL/SASL_SSL."""

    ssl_cafile: str | None = Field(default=None, description="Path to CA certificate")
    ssl_certfile: str | None = Field(default=None, description="Path to client certificate")
    ssl_keyfile: str | None = Field(default=None, description="Path to client key")
    ssl_check_hostname: bool = Field(default=True, description="Verify hostname")

class BaseKafkaConsumerSettings(BaseKafkaSettings):
    """Kafka consumer configuration.

    ``enable_auto_commit`` defaults to ``False`` because at-least-once consumers
    must commit offsets manually; flip it to ``True`` only for fire-and-forget
    or idempotent consumers.
    """

    group_id: str = Field(description="Consumer group ID")
    auto_offset_reset: KafkaOffsetReset = Field(default="earliest", description="Auto offset reset")
    enable_auto_commit: bool = Field(
        default=False,
        description="Enable broker-side auto-commit. Keep False for at-least-once delivery.",
    )
    session_timeout_ms: int = Field(default=30000, ge=0, description="Session timeout in ms")
    heartbeat_interval_ms: int = Field(default=3000, ge=0, description="Heartbeat interval in ms")
    max_poll_records: int = Field(default=500, ge=1, description="Max records per poll")
    max_poll_interval_ms: int = Field(default=300000, ge=1, description="Maximum poll interval in ms")
    fetch_max_wait_ms: int = Field(default=500, ge=0, description="Maximum wait time for fetch response")
    fetch_min_bytes: int = Field(default=1, ge=0, description="Minimum bytes server should return for fetch")
    fetch_max_bytes: int = Field(default=52428800, ge=1, description="Maximum bytes fetched per request")

class BaseKafkaProducerSettings(BaseKafkaSettings, KafkaAutoCreateMixin):
    """Kafka producer configuration."""

    acks: KafkaAcks = Field(default="all", description="Acks: 0, 1, all")
    compression_type: KafkaCompressionType | None = Field(
        default="gzip",
        description="Compression: gzip, snappy, lz4, zstd, None",
    )
    enable_idempotence: bool = Field(default=True, description="Enable idempotent producer")
    max_batch_size: int = Field(default=16384, ge=0, description="Max batch size in bytes")
    linger_ms: int = Field(default=5, ge=0, description="Linger time in ms for batching")
    request_timeout_ms: int = Field(default=30000, ge=0, description="Request timeout in ms")

    @model_validator(mode="after")
    def _validate_auto_create(self) -> BaseKafkaProducerSettings:
        if not self.auto_create_topics:
            return self

        if self.default_replication_factor is None:
            msg = "Kafka default_replication_factor is required when auto_create_topics is enabled"
            raise ValueError(msg)

        return self

class KafkaAutoCreateMixin(BaseModel):
    """Mixin with topic auto-creation settings (kept separate from transport config)."""

    auto_create_topics: bool = Field(
        default=False,
        description="Auto-create topics on startup",
    )
    default_partitions: int = Field(
        default=3,
        ge=1,
        description="Default partitions for new topics",
    )
    default_replication_factor: int | None = Field(
        default=None,
        ge=1,
        description="Default replication factor for new topics",
    )

class BaseKafkaInfraSettings(BaseModel):
    """Shared Kafka infrastructure settings."""

    topic_prefix: str | None = Field(default=None, description="Global prefix for all service topics")
    topic_catalog: dict[str, KafkaTopicSettings] | None = Field(
        default=None,
        description="Map of logical topic names to physical settings for topic creation",
    )
    consumer_subscriptions: list[str] | None = Field(
        default=None,
        description="Logical topic names to subscribe for inbox consumers",
    )

    @field_validator("topic_prefix", mode="before")
    @classmethod
    def _coerce_topic_prefix(cls, value: object) -> str | None:
        return normalize_kafka_topic_prefix_value(value)

class KafkaTopicSettings(BaseModel):
    """Physical parameters for a specific Kafka topic."""

    num_partitions: int = Field(default=3, ge=1, description="Number of partitions for the topic")
    replication_factor: int = Field(default=3, ge=1, description="Replication factor for the topic")
    topic_configs: dict[str, str] | None = Field(default=None, description="Topic-specific configuration overrides")

class AsyncKafkaProducerProvider(Provider):
    """Dishka provider exposing an ``AIOKafkaProducer`` for ``APP`` scope.

    Topics passed via the container are auto-created when
    ``settings.auto_create_topics`` is enabled.
    """

    scope = Scope.APP

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        check_dishka()
        super().__init__(*args, **kwargs)

    @provide
    async def get_kafka_producer(
        self,
        kafka_settings: ProducerLifecycleSettingsProtocol,
        topics: Sequence[TopicConfigProtocol] | None = None,
    ) -> AsyncIterator[AIOKafkaProducer]:
        """Provide Kafka producer as a singleton for APP scope."""
        async with producer_lifecycle(
            kafka_settings,
            topics=topics,
            auto_create_topics=kafka_settings.auto_create_topics,
        ) as producer:
            yield producer

class AsyncKafkaConsumerProvider(Provider):
    """Dishka provider exposing an ``AIOKafkaConsumer`` for ``APP`` scope."""

    scope = Scope.APP

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        check_dishka()
        super().__init__(*args, **kwargs)

    @provide
    async def get_kafka_consumer(
        self,
        kafka_settings: ConsumerSettingsProtocol,
        topics: tuple[str, ...] | None = None,
    ) -> AsyncIterator[AIOKafkaConsumer]:
        """Provide Kafka consumer as a singleton for APP scope."""
        async with consumer_lifecycle(
            kafka_settings,
            topics=topics,
        ) as consumer:
            yield consumer

class KafkaInfraProvider(Provider):
    """Provide resolved topic configs for producer and subscriptions for consumers."""

    scope = Scope.APP

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        check_dishka()
        super().__init__(*args, **kwargs)

    @provide
    def get_topic_configs_for_catalog(
        self,
        settings: KafkaProducerInfraSettingsProtocol,
    ) -> Sequence[TopicConfig]:
        """Provide physical topic configs for topic creation based on topic catalog."""
        prefix = settings.topic_prefix
        topic_catalog = settings.topic_catalog or {}

        return [
            TopicConfig(
                name=_apply_topic_prefix(prefix, name),
                num_partitions=cfg.num_partitions,
                replication_factor=cfg.replication_factor,
                topic_configs=cfg.topic_configs,
            )
            for name, cfg in topic_catalog.items()
        ]

    @provide
    def get_consumer_subscription_topics(
        self,
        settings: KafkaConsumerInfraSettingsProtocol,
    ) -> tuple[str, ...]:
        """Provide fully resolved physical topic names for consumer subscriptions."""
        subscriptions = settings.consumer_subscriptions or []
        if not subscriptions:
            return ()

        prefix = settings.topic_prefix
        return tuple(_apply_topic_prefix(prefix, topic) for topic in subscriptions)

class KafkaProducerInfraSettingsProtocol(KafkaInfraBaseSettingsProtocol, Protocol):
    """Protocol for producer-side Kafka infrastructure settings.

    Defines what producer providers need: a topic prefix and a catalog of
    physical topic configs to ensure exist on startup.
    """

    topic_catalog: dict[str, KafkaTopicSettingsProtocol] | None

class KafkaConsumerInfraSettingsProtocol(KafkaInfraBaseSettingsProtocol, Protocol):
    """Protocol for consumer-side Kafka infrastructure settings.

    Defines what consumer providers need: a topic prefix and a list of
    logical topic names to subscribe to.
    """

    consumer_subscriptions: list[str] | None

class KafkaTopicSettingsProtocol(Protocol):
    """Physical parameters for one topic in a catalog.

    Unlike :class:`~aiokafka_foundation_kit.config.topic.TopicConfigProtocol`,
    this one omits ``name`` because the topic name comes from the catalog map
    key in :class:`KafkaProducerInfraSettingsProtocol.topic_catalog`.
    """

    num_partitions: int
    replication_factor: int
    topic_configs: dict[str, str] | None

class KafkaProducerContainer(containers.DeclarativeContainer):
    """Container for Kafka producer dependencies.

    Provides:
        - ``producer``: :class:`AIOKafkaProducer` with automatic lifecycle.

    Configuration:
        - ``kafka_settings``: Kafka producer settings (``ProducerSettingsProtocol``).
        - ``topics``: Optional list of topics to auto-create (``Sequence[TopicConfigProtocol]``).
        - ``auto_create_topics``: Whether to run ``ensure_topics_async`` on startup.
    """

    kafka_settings = providers.Dependency()  # type: ignore[var-annotated]
    topics = providers.Dependency(default=None)  # type: ignore[var-annotated]
    auto_create_topics = providers.Object(False)  # type: ignore[var-annotated]

    producer = providers.Resource(  # type: ignore[var-annotated]
        _producer_resource,
        kafka_settings=kafka_settings,
        topics=topics,
        auto_create_topics=auto_create_topics,
    )

    def __init__(self, *args: Any, **kwargs: Any) -> None:  # pragma: no cover
        check_dependency_injector()  # pragma: no cover
        super().__init__(*args, **kwargs)  # pragma: no cover

class KafkaConsumerContainer(containers.DeclarativeContainer):
    """Container for Kafka consumer dependencies.

    Provides:
        - ``consumer``: :class:`AIOKafkaConsumer` with automatic lifecycle.

    Configuration:
        - ``kafka_settings``: Kafka consumer settings (``ConsumerSettingsProtocol``).
        - ``topics``: Optional tuple of topic names to subscribe to.
    """

    kafka_settings = providers.Dependency()  # type: ignore[var-annotated]
    topics = providers.Dependency(default=None)  # type: ignore[var-annotated]

    consumer = providers.Resource(  # type: ignore[var-annotated]
        _consumer_resource,
        kafka_settings=kafka_settings,
        topics=topics,
    )

    def __init__(self, *args: Any, **kwargs: Any) -> None:  # pragma: no cover
        check_dependency_injector()  # pragma: no cover
        super().__init__(*args, **kwargs)  # pragma: no cover