Configuration¶

Design: protocol-based settings¶

All factory functions and lifecycle helpers accept protocols rather than concrete types. A typing.Protocol describes only the attributes and methods a settings object must expose — it imposes no base class requirement.

This means you can pass:

• A BaseKafkaProducerSettings Pydantic model (provided by the models extra).
• Your own dataclass or plain class that matches the shape.
• A MagicMock in tests.

The protocol hierarchy is:

KafkaConnectionSettingsProtocol   ← bootstrap_servers, client_id, security_protocol,
│                                    metadata_max_age_ms, get_sasl_password()
├── KafkaSaslSettingsProtocol     ← + sasl_mechanism, sasl_username
└── KafkaSslSettingsProtocol      ← + ssl_cafile, ssl_certfile, ssl_keyfile, ssl_check_hostname

KafkaSettingsProtocol             ← combines SASL + SSL (base for producer and consumer)
├── ConsumerSettingsProtocol      ← + group_id, auto_offset_reset, enable_auto_commit, …
└── ProducerSettingsProtocol      ← + acks, compression_type, enable_idempotence, …
    └── ProducerLifecycleSettingsProtocol ← + auto_create_topics

Pydantic models (recommended)¶

Install the models extra to get Pydantic implementations of the protocols:

pip install aiokafka-foundation-kit[models]

BaseKafkaSettings¶

Shared base for all settings. Combines connection, SASL, and SSL mixins.

Validators enforce consistency: SASL fields are required when security_protocol is SASL_*; ssl_cafile is required when security_protocol is SSL or SASL_SSL.

BaseKafkaConsumerSettings¶

Extends BaseKafkaSettings.

BaseKafkaProducerSettings¶

Extends BaseKafkaSettings + KafkaAutoCreateMixin.

BaseKafkaInfraSettings¶

Infrastructure-level settings used by KafkaInfraProvider (Dishka).

topic_prefix normalises "None", "null", and empty strings to None — useful when the value comes from an env var or a secrets vault.

Security modes¶

PLAINTEXT (default)¶

No authentication or encryption. Suitable for local development.

from aiokafka_foundation_kit.contrib.models import BaseKafkaProducerSettings

settings = BaseKafkaProducerSettings(bootstrap_servers="localhost:9092")

SASL_PLAINTEXT / SASL_SSL¶

SASL authentication over plaintext or TLS.

settings = BaseKafkaProducerSettings(
    bootstrap_servers="kafka.internal:9093",
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",
    sasl_username="service-account",
    sasl_password="s3cr3t",
    ssl_cafile="/etc/ssl/certs/ca-bundle.crt",
)

Supported mechanisms: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512.

SSL (mutual TLS)¶

Client certificate authentication.

settings = BaseKafkaProducerSettings(
    bootstrap_servers="kafka.internal:9093",
    security_protocol="SSL",
    ssl_cafile="/etc/certs/ca.crt",
    ssl_certfile="/etc/certs/client.crt",
    ssl_keyfile="/etc/certs/client.key",
)

How the TLS fields reach aiokafka¶

aiokafka takes a prepared ssl.SSLContext, not certificate paths. For SSL and SASL_SSL, build_kafka_common_config loads the three ssl_* paths through aiokafka.helpers.create_ssl_context and passes the result as ssl_context; the individual paths never reach the client.

Two consequences worth knowing:

• The certificate files are read when the client is built, so a missing or unreadable file raises FileNotFoundError (a malformed one ssl.SSLError) from create_async_kafka_producer / create_async_kafka_consumer / ensure_topics_async, not on the first connection attempt. With no ssl_cafile, the system trust store is used.
• ssl_check_hostname=False sets check_hostname = False on the context. The broker certificate is still verified against the CA — only the hostname match is skipped.

Loading settings from environment¶

Combine with pydantic-settings to load from environment variables or .env files:

from pydantic_settings import BaseSettings
from aiokafka_foundation_kit.contrib.models import BaseKafkaProducerSettings

class AppSettings(BaseSettings):
    kafka: BaseKafkaProducerSettings = BaseKafkaProducerSettings(
        bootstrap_servers="localhost:9092"
    )

    model_config = {"env_nested_delimiter": "__"}

# KAFKA__BOOTSTRAP_SERVERS=kafka:9092 KAFKA__GROUP_ID=my-group python app.py
settings = AppSettings()

Adding your own fields¶

The models carry what the library reads, plus the two caller-side defaults noted above. There is no observability flag on either of them: the library collects no metrics, on the producer side or the consumer side (the telemetry extra is tracing only), so a metrics_enabled field would be a switch wired to nothing. A flag like that belongs with the code that owns the collector. Keep it in a mixin of your own and compose it into both subclasses, the way the library's own Kafka*Mixins are composed — one Protocol with a metrics_enabled: bool member then describes producer and consumer settings alike:

from pydantic import BaseModel, ConfigDict

from aiokafka_foundation_kit.contrib.models import (
    BaseKafkaConsumerSettings,
    BaseKafkaProducerSettings,
)


class KafkaMetricsMixin(BaseModel):
    metrics_enabled: bool = False


class ProducerSettings(BaseKafkaProducerSettings, KafkaMetricsMixin):
    model_config = ConfigDict(extra="forbid")


class ConsumerSettings(BaseKafkaConsumerSettings, KafkaMetricsMixin):
    model_config = ConfigDict(extra="forbid")

Custom settings (protocol implementation)¶

If you prefer not to depend on Pydantic, implement the protocol directly:

from dataclasses import dataclass

from aiokafka_foundation_kit import producer_lifecycle
from aiokafka_foundation_kit.config.kafka import (
    KafkaAcks,
    KafkaCompressionType,
    KafkaSaslMechanism,
    KafkaSecurityProtocol,
)

@dataclass
class MyProducerSettings:
    bootstrap_servers: str
    client_id: str | None = None
    security_protocol: KafkaSecurityProtocol = "PLAINTEXT"
    metadata_max_age_ms: int = 300_000
    acks: KafkaAcks = "all"
    compression_type: KafkaCompressionType | None = "gzip"
    enable_idempotence: bool = True
    max_batch_size: int = 16_384
    linger_ms: int = 5
    request_timeout_ms: int = 30_000

    # SASL and TLS fields (unused when security_protocol == "PLAINTEXT")
    sasl_mechanism: KafkaSaslMechanism | None = None
    sasl_username: str | None = None
    ssl_cafile: str | None = None
    ssl_certfile: str | None = None
    ssl_keyfile: str | None = None
    ssl_check_hostname: bool = True

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

settings = MyProducerSettings(bootstrap_servers="localhost:9092")

async with producer_lifecycle(settings) as producer:
    await producer.send_and_wait("topic", b"data")

The plain str annotations a hand-written class usually starts with will not satisfy the protocol under a type checker: security_protocol, acks, compression_type, sasl_mechanism and auto_offset_reset are Literal aliases, and protocol attributes are invariant. Import the aliases from aiokafka_foundation_kit.config.kafka as above.