Add ssl context support (#19)

* Add ssl context support

* Update version

* Enrich example with ssl context

Author: 1owkeyme

examples/kafka_proto_event_producing.py

@@ -1,4 +1,5 @@
 import asyncio
+import ssl
 
 import pydantic
 
@@ -30,16 +31,32 @@ def proto(self) -> UserJoinedECSTProtobuf:
         )
 
 
+def create_kafka_producer(ssl_context: ssl.SSLContext | None = None) -> kafka_adapters.KafkaProducer:
+    dsn = "localhost:9092"
+    value_serializer = protobuf.protobuf_value_serializer
+    if ssl_context is None:
+        return kafka_adapters.kafka_producer_factory(
+            security_protocol="PLAINTEXT",
+            sasl_mechanism="PLAIN",
+            dsn=dsn,
+            value_serializer=value_serializer,
+        )
+    return kafka_adapters.kafka_producer_factory(
+        security_protocol="SASL_SSL",
+        sasl_mechanism="SCRAM-SHA-256",
+        ssl_context=ssl_context,
+        dsn=dsn,
+        value_serializer=value_serializer,
+    )
+
+
 async def main():
     event = UserJoinedECST(
         event_name="user_joined_ecst",
         topic="user_joined_proto",
         payload=UserJoinedECSTPayload(user_id="123", meeting_id="456"),
     )
-    kafka_producer = kafka_adapters.kafka_producer_factory(
-        dsn="localhost:9092",
-        value_serializer=protobuf.protobuf_value_serializer,
-    )
+    kafka_producer = create_kafka_producer(ssl_context=None)
     broker = kafka.KafkaMessageBroker(
         producer=kafka_producer,
     )

pyproject.toml

@@ -29,7 +29,7 @@ maintainers = [{name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru"}]
 name = "python-cqrs"
 readme = "README.md"
 requires-python = ">=3.10"
-version = "3.0.1"
+version = "3.1.1"
 
 [project.optional-dependencies]
 dev = [

src/cqrs/adapters/kafka.py

@@ -1,6 +1,7 @@
 import asyncio
 import functools
 import logging
+import ssl
 import typing
 
 import aiokafka
@@ -89,6 +90,7 @@ def kafka_producer_factory(
     dsn: typing.Text,
     security_protocol: SecurityProtocol = "PLAINTEXT",
     sasl_mechanism: SaslMechanism = "PLAIN",
+    ssl_context: ssl.SSLContext | None = None,
     retry_count: int = 3,
     retry_delay: int = 1,
     user: typing.Text | None = None,
@@ -105,6 +107,7 @@ def kafka_producer_factory(
         sasl_mechanism=sasl_mechanism,
         sasl_plain_username=user,
         sasl_plain_password=password,
+        ssl_context=ssl_context,
         loop=loop,
     )
     return KafkaProducer(