68 lines
2.2 KiB
YAML
68 lines
2.2 KiB
YAML
---
|
|
apiVersion: v1
|
|
kind: ConfigMap
|
|
metadata:
|
|
name: kafka-config
|
|
namespace: message-hub
|
|
data:
|
|
kafka.py: |
|
|
from ssl import SSLContext
|
|
|
|
import ssl
|
|
|
|
from aiokafka.helpers import create_ssl_context
|
|
from faststream.kafka import KafkaBroker
|
|
from faststream.security import BaseSecurity, SASLPlaintext, SASLScram512
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class KafkaSettings(BaseSettings):
|
|
HOST: str = 'localhost'
|
|
PORT: int = 9092
|
|
USERNAME: str | None = None
|
|
PASSWORD: str | None = None
|
|
SECURITY_PROTOCOL: str = 'PLAINTEXT'
|
|
SASL_MECHANISM: str | None = None
|
|
SSL_CAFILE: str | None = None
|
|
|
|
model_config = SettingsConfigDict(env_prefix='KAFKA_', env_file='.env', extra='ignore')
|
|
|
|
@property
|
|
def bootstrap_servers(self) -> list[str]:
|
|
return [f'{self.HOST}:{str(self.PORT)}']
|
|
|
|
def get_ssl_context(self) -> SSLContext:
|
|
context = create_ssl_context(cafile=self.SSL_CAFILE)
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
return context
|
|
|
|
def get_security(self) -> BaseSecurity | None:
|
|
use_ssl = self.SECURITY_PROTOCOL in ('SSL', 'SASL_SSL')
|
|
ssl_context = self.get_ssl_context() if use_ssl else None
|
|
if self.SASL_MECHANISM == 'PLAINTEXT':
|
|
return SASLPlaintext(
|
|
username=self.USERNAME or '',
|
|
password=self.PASSWORD or '',
|
|
ssl_context=ssl_context,
|
|
use_ssl=use_ssl,
|
|
)
|
|
if self.SASL_MECHANISM == 'SCRAM-SHA-512':
|
|
return SASLScram512(
|
|
username=self.USERNAME or '',
|
|
password=self.PASSWORD or '',
|
|
ssl_context=ssl_context,
|
|
use_ssl=use_ssl,
|
|
)
|
|
if use_ssl:
|
|
return BaseSecurity(ssl_context=ssl_context)
|
|
return None
|
|
|
|
@property
|
|
def broker(self) -> KafkaBroker:
|
|
return KafkaBroker(
|
|
bootstrap_servers=self.bootstrap_servers,
|
|
security=self.get_security(),
|
|
logger=None,
|
|
)
|