KafkaConnectionSettings

pydantic model safir.kafka.KafkaConnectionSettings

Settings for connecting to Kafka.

This settings model supports different authentication methods, which each have different sets of required settings. All of these settings can be provided in KAFKA_ prefixed environment variables. Instances of this model have properties that can be used to construct different types of kafka clients:

from faststream.broker import KafkaBroker

from safir.kafka import KafkaConnectionSettings

config = KafkaConnectionSettings()
kafka_broker = KafkaBroker(**config.faststream_broker_params)

When using this model directly, The validated property enforces at runtime that the correct settings were provided for the desired authentication method, and returns models to access those settings in a type-safe way:

from pathlib import Path

# ValidationError at runtime: ``client_key_path`` is not provided
config = KafkaConnectionSettings(
    bootstrap_servers="something:1234",
    security_protocol=KafkaSecurityProtocol.SSL,
    cluster_ca_path=Path("/some/cert.crt"),
    client_cert_path=Path("/some/other/cert.crt"),
)

config = KafkaConnectionSettings(
    bootstrap_servers="something:1234",
    security_protocol=KafkaSecurityProtocol.SSL,
    cluster_ca_path=Path("/some/path/ca.crt"),
    client_cert_path=Path("/some/path/user.crt"),
    client_key_path=Path("/some/path/user.key"),
)

blah = config.validated.sasl_username  # Static type error

Parameters:

• _case_sensitive (bool | None, default: None)

• _nested_model_default_partial_update (bool | None, default: None)

• _env_prefix (str | None, default: None)

• _env_prefix_target (Literal['variable', 'alias', 'all'] | None, default: None)

• _env_file (Path | str | Sequence[Path | str] | None, default: PosixPath('.'))

• _env_file_encoding (str | None, default: None)

• _env_file_depth (int | None, default: None)

• _env_ignore_empty (bool | None, default: None)

• _env_nested_delimiter (str | None, default: None)

• _env_nested_max_split (int | None, default: None)

• _env_parse_none_str (str | None, default: None)

• _env_parse_enums (bool | None, default: None)

• _cli_prog_name (str | None, default: None)

• _cli_parse_args (bool | list[str] | tuple[str, ...] | None, default: None)

• _cli_settings_source (CliSettingsSource[Any] | None, default: None)

• _cli_parse_none_str (str | None, default: None)

• _cli_hide_none_type (bool | None, default: None)

• _cli_avoid_json (bool | None, default: None)

• _cli_enforce_required (bool | None, default: None)

• _cli_use_class_docs_for_groups (bool | None, default: None)

• _cli_show_env_vars (bool | None, default: None)

• _cli_exit_on_error (bool | None, default: None)

• _cli_prefix (str | None, default: None)

• _cli_flag_prefix_char (str | None, default: None)

• _cli_implicit_flags (bool | Literal['dual', 'toggle'] | None, default: None)

• _cli_ignore_unknown_args (bool | None, default: None)

• _cli_kebab_case (bool | Literal['all', 'no_enums'] | None, default: None)

• _cli_shortcuts (Mapping[str, str | list[str]] | None, default: None)

• _secrets_dir (Path | str | Sequence[Path | str] | None, default: None)

• _build_sources (tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None, default: None)

• bootstrap_servers (str)

• security_protocol (SecurityProtocol)

• cluster_ca_path (Path | None, default: None)

• client_cert_path (Path | None, default: None)

• client_key_path (Path | None, default: None)

• sasl_mechanism (SaslMechanism | None, default: None)

• sasl_username (str | None, default: None)

• sasl_password (SecretStr | None, default: None)

Config:

• validate_by_name: bool = True

Fields:

• bootstrap_servers (str)

• client_cert_path (pathlib.Path | None)

• client_key_path (pathlib.Path | None)

• cluster_ca_path (pathlib.Path | None)

• sasl_mechanism (safir.kafka._kafka_config.SaslMechanism | None)

• sasl_password (pydantic.types.SecretStr | None)

• sasl_username (str | None)

• security_protocol (safir.kafka._kafka_config.SecurityProtocol)

Validators:

• validate_auth_settings » all fields

field bootstrap_servers: str [Required]

A comma-separated list of Kafka brokers to connect to. This should be a list of hostnames or IP addresses, each optionally followed by a port number, separated by commas.

Validated by:

• validate_auth_settings

field client_cert_path: Annotated[Path, PathType(path_type=file)] | None = None

The path to the PEM-formated client certificate file to use for authentication. This is only needed if the broker is configured to require SSL client authentication.

Validated by:

• validate_auth_settings

field client_key_path: Annotated[Path, PathType(path_type=file)] | None = None

The path to the PEM-formatted client key file to use for authentication. This is only needed if for the SSL securityprotocol.

Validated by:

• validate_auth_settings

field cluster_ca_path: Annotated[Path, PathType(path_type=file)] | None = None

The path to the PEM-formatted CA certificate file to use for verifying the broker’s certificate. This is only needed for SSL and SASL_SSL security protocols, andeven in those cases, only when the broker’s certificate is not signed by a CA trusted by the operating system.

Validated by:

• validate_auth_settings

field sasl_mechanism: SaslMechanism | None = None

The SASL mechanism to use for authentication. This is only needed for the SASL_SSL and SASL_PLAINTEXT securityprotocols.

Validated by:

• validate_auth_settings

field sasl_password: SecretStr | None = None

The password to use for SASL authentication. This is only needed for the SASL_SSL and SASL_PLAINTEXT securityprotocols.

Validated by:

• validate_auth_settings

field sasl_username: str | None = None

The username to use for SASL authentication. This is only needed for the SASL_SSL and SASL_PLAINTEXT securityprotocols.

Validated by:

• validate_auth_settings

field security_protocol: SecurityProtocol [Required]

The authentication and encryption mode for the connection.

Validated by:

• validate_auth_settings

classmethod settings_cached()

Return the cached default settings instance for this settings class.

The instance is constructed on first use and reused afterwards. Subclasses are cached separately, so caching a class does not affect its parents or children.

Return type:

Self

classmethod settings_clear_cache()

Remove the cached default settings instance for this settings class, if any.

Only the cache entry for this exact class is removed. Subclasses keep their own cached instances and must be cleared individually.

A settings_cached() call already in flight may still return the instance it was building, so clearing does not retroactively invalidate references other threads hold.

Return type:

None

classmethod settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings)

Define the sources and their order for loading the settings values.

Args:

settings_cls: The Settings class. init_settings: The InitSettingsSource instance. env_settings: The EnvSettingsSource instance. dotenv_settings: The DotEnvSettingsSource instance. file_secret_settings: The SecretsSettingsSource instance.

Returns:

A tuple containing the sources and their order for loading the settings values.

Parameters:

• settings_cls (type[BaseSettings])

• init_settings (PydanticBaseSettingsSource)

• env_settings (PydanticBaseSettingsSource)

• dotenv_settings (PydanticBaseSettingsSource)

• file_secret_settings (PydanticBaseSettingsSource)

Return type:

tuple[PydanticBaseSettingsSource, ...]

to_aiokafka_params()

Return type:

AIOKafkaParams

to_faststream_params()

Return type:

FastStreamBrokerParams

validator validate_auth_settings  »  all fields

Validate that the correct combination of parameters is specified.

Return type:

Self

property validated: SslSettings | SaslSslSettings | SaslPlaintextSettings | PlaintextSettings

Return a model with a subset of settings for a Kafka auth method.

This method will fail with a ValidationError if an invalid set of settings were provided.