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 <https://kafka.apache.org/documentation/#security_sasl_plain>_.

.. versionadded:: 0.9.0

Examples

Auth in Kafka with user and password:

.. code:: python

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>`_.

    .. versionadded:: 0.9.0

    Examples
    --------

    Auth in Kafka with user and password:

    .. code:: 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:  # noqa: WPS473
        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: type[T],
    options: GenericOptions | dict | None,
) -> T:
    """
    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):
        raise TypeError(
            f"{options.__class__.__name__} is not a {cls.__name__} instance",
        )

    return options