Backends

The main idea of a backend is to supply the necessary configuration to create a connection with the backend.

kstreams currently has support for Kafka and InMemory backend.

kstreams.backends.kafka.Kafka

The Kafka backend validates the given attributes.

It uses pydantic internally.

Attributes:

Name	Type	Description
bootstrap_servers	List[str]	kafka list of hostname:port
security_protocol	SecurityProtocol	Protocol used to communicate with brokers
ssl_context	Optional[SSLContext]	a python std ssl.SSLContext instance, you can generate it with create_ssl_context or create_ssl_context_from_mem
sasl_mechanism	SaslMechanism	Authentication mechanism when security_protocol is configured for SASL_PLAINTEXT or SASL_SSL
sasl_plain_username	Optional[str]	username for sasl PLAIN authentication
sasl_plain_password	Optional[str]	password for sasl PLAIN authentication
sasl_oauth_token_provider	Optional[str]	smth

Raises:

Type	Description
ValidationError	a pydantic.ValidationError exception

PLAINTEXT

Example

from kstreams.backends.kafka import Kafka
from kstreams import create_engine, Stream

backend = Kafka(bootstrap_servers=["localhost:9092"])
stream_engine = create_engine(title="my-stream-engine", backend=backend)

SSL

Example

Create SSL context

import ssl

from kstreams.backends.kafka import Kafka
from kstreams import create_engine, utils, Stream


def get_ssl_context() -> ssl.SSLContext:
    return utils.create_ssl_context(
        cafile="certificate-authority-file-path",
        capath="points-to-directory-with-several-ca-certificates",
        cadata="same-as-cafile-but-ASCII-or-bytes-format",
        certfile="client-certificate-file-name",
        keyfile="client-private-key-file-name",
        password="password-to-load-certificate-chain",
    )

backend = Kafka(
    bootstrap_servers=["localhost:9094"],
    security_protocol="SSL",
    ssl_context=get_ssl_context(),
)

stream_engine = create_engine(title="my-stream-engine", backend=backend)

Note

Check create ssl context util

Example

Create SSL context from memory

import ssl

from kstreams.backends.kafka import Kafka
from kstreams import create_engine, utils, Stream


def get_ssl_context() -> ssl.SSLContext:
    return utils.create_ssl_context_from_mem(
        cadata="ca-certificates-as-unicode",
        certdata="client-certificate-as-unicode",
        keydata="client-private-key-as-unicode",
        password="optional-password-to-load-certificate-chain",
    )

backend = Kafka(
    bootstrap_servers=["localhost:9094"],
    security_protocol="SSL",
    ssl_context=get_ssl_context(),
)

stream_engine = create_engine(title="my-stream-engine", backend=backend)

Note

Check create ssl context from memerory util

kstreams.backends.memory.InMemory

InMemory backend for KStreams. This backend is useful for testing and development purposes, as it does not require any external dependencies. Do not use this backend in production, as it does not provide any durability guarantees.

Example

from kstreams.backends.memory import InMemory
from kstreams import create_engine


backend = InMemory()
stream_engine = create_engine(title="my-stream-engine", backend=backend)

Note

Full example of how to use the InMemory backend can be found here