Kafka BasicAuth

onetl.connection.db_connection.kafka.kafka_basic_auth.KafkaBasicAuth

Bases: KafkaAuth, GenericOptions

Connect to Kafka using sasl.mechanism="PLAIN".

For more details see Kafka Documentation.

Added in 0.9.0

Examples

Auth in Kafka with user and password:

from onetl.connection import Kafka

auth = Kafka.BasicAuth(
    user="some_user",
    password="abc",
)

Source code in onetl/connection/db_connection/kafka/kafka_basic_auth.py

class KafkaBasicAuth(KafkaAuth, GenericOptions):
    """
    Connect to Kafka using `sasl.mechanism="PLAIN"`.

    For more details see [Kafka Documentation](https://kafka.apache.org/documentation/#security_sasl_plain).

    !!! success "Added in 0.9.0"

    Examples
    --------

    Auth in Kafka with user and password:

    ```python
    from onetl.connection import Kafka

    auth = Kafka.BasicAuth(
        user="some_user",
        password="abc",
    )
    ```
    """

    user: str = Field(alias="username")
    password: SecretStr

    def get_jaas_conf(self) -> str:
        return (
            "org.apache.kafka.common.security.plain.PlainLoginModule required "
            f'username="{self.user}" '
            f'password="{self.password.get_secret_value()}";'
        )

    def get_options(self, kafka: Kafka) -> dict:
        return {
            "sasl.mechanism": "PLAIN",
            "sasl.jaas.config": self.get_jaas_conf(),
        }

    def cleanup(self, kafka: Kafka) -> None:
        # nothing to cleanup
        pass

parse(options) classmethod

If a parameter inherited from the ReadOptions class was passed, then it will be returned unchanged. If a Dict object was passed it will be converted to ReadOptions.

Otherwise, an exception will be raised

Source code in onetl/impl/generic_options.py

@classmethod
def parse(
    cls,
    options: GenericOptions | dict | None,
) -> Self:
    """
    If a parameter inherited from the ReadOptions class was passed, then it will be returned unchanged.
    If a Dict object was passed it will be converted to ReadOptions.

    Otherwise, an exception will be raised
    """

    if not options:
        return cls()

    if isinstance(options, dict):
        return cls.parse_obj(options)

    if not isinstance(options, cls):
        msg = f"{options.__class__.__name__} is not a {cls.__name__} instance"
        raise TypeError(msg)

    return options