fix: optional ssl for kafka and optional tg user fields

This commit is contained in:
2025-04-10 11:19:46 +04:00
parent f083c19cdc
commit b5c8e35910
6 changed files with 16 additions and 20 deletions

View File

@@ -1,20 +1,19 @@
from aiokafka import AIOKafkaConsumer
from aiokafka.errors import KafkaConnectionError
from backend.config import KAFKA_URL, KAFKA_CONSUMER_TOPIC
from backend.config import KAFKA_URL, KAFKA_CONSUMER_TOPIC, KAFKA_ENABLE_SSL
from backend.session import session_maker
from external.kafka.context import context
from external.kafka.services.consumer_service import ConsumerService
async def consume_messages():
consumer = AIOKafkaConsumer(
KAFKA_CONSUMER_TOPIC,
bootstrap_servers=KAFKA_URL,
group_id='crm',
security_protocol='SSL',
ssl_context=context,
security_protocol='SSL' if KAFKA_ENABLE_SSL else 'PLAINTEXT',
ssl_context=context if KAFKA_ENABLE_SSL else None,
)
try:
await consumer.start()