refactor(config): update di to not use pydantic settings directly

app/__init__.py

@@ -17,7 +17,7 @@
 from app.infra.telemetry.otel.traces import initialize_traces
 from app.infra.telemetry.prometheus import setup_prometheus_client
 import app.api.sms.routes
-from .settings import get_config
+from .settings import get_settings
 
 logging.root.setLevel(logging.INFO)
 console_formatter = uvicorn.logging.ColourizedFormatter(
@@ -27,7 +27,7 @@
 for handler in root.handlers:
     handler.setFormatter(console_formatter)
 
-_settings = get_config()
+_settings = get_settings()
 
 # initialize sentry
 initialize_sentry()

app/adapters/broker/producers/sms_received_producer.py

@@ -18,10 +18,10 @@ class SmsReceivedProducer(Producer):
 
     def __init__(self, topic: str, event_stream: AsyncEventStream):
         """
-        Creates an instance of an sms received producer with a topic to send events to and a KafkaProducer client to
+        Creates an instance of an SMS received producer with a topic to send events to and a KafkaProducer client to
         use to send events.
         Args:
-            topic (str): Topic to send message to.
+            topic (str): Topic to send messages to.
             event_stream (AsyncEventStream): Kafka Producer client to use
         """
         self.topic = topic

app/config/di/container.py

@@ -1,5 +1,5 @@
 from dependency_injector import containers, providers
-from .gateways_container import GatewaysContainer
+from .infra_container import InfraContainer
 from .event_stream_container import EventStreamContainer
 from .services_container import ServicesContainer
 from .repository_container import RepositoryContainer
@@ -11,12 +11,12 @@ class ApplicationContainer(containers.DeclarativeContainer):
     Application container wiring all dependencies together
     """
 
-    kafka = providers.Container(EventStreamContainer)
+    event_stream = providers.Container(EventStreamContainer)
 
-    gateways = providers.Container(GatewaysContainer)
+    infra = providers.Container(InfraContainer)
 
-    services = providers.Container(ServicesContainer, gateways=gateways, kafka_container=kafka)
+    services = providers.Container(ServicesContainer, infra=infra, event_stream=event_stream)
 
-    repository = providers.Container(RepositoryContainer, gateways=gateways)
+    repository = providers.Container(RepositoryContainer, gateways=infra)
 
     domain = providers.Container(DomainContainer, services=services, repository=repository)

app/config/di/event_stream_container.py

@@ -1,14 +1,10 @@
 """
 Kafka DI container
 """
-from typing import cast
 
 from dependency_injector import containers, providers
 import sanctumlabs.messageschema.events.notifications.sms.v1.events_pb2 as events
-from eventmsg_adaptor.config.kafka import KafkaConfig, KafkaSchemaRegistryConfig, KafkaSecurityProtocolConfig
-from eventmsg_adaptor.event_streams import AsyncEventStream
-from eventmsg_adaptor import factory
-from eventmsg_adaptor.config import Config, AdapterConfigs
+from eventmsg_adaptor.config.kafka import KafkaSchemaRegistryConfig
 
 from app.infra.broker.kafka.config import KafkaProducerConfig, KafkaConsumerConfig
 from app.infra.broker.kafka.producers.simple_producer import KafkaSimpleProducer
@@ -17,31 +13,16 @@
 from app.infra.broker.kafka.serializers.protobuf_serializer import KafkaProtobufSerializer
 from app.infra.broker.kafka.deserializers.protobuf_deserializer import KafkaProtobufDeserializer
 from app.infra.broker.kafka.registry import KafkaRegistry
-from app.settings import KafkaSettings
 
 
 class EventStreamContainer(containers.DeclarativeContainer):
     """
     Dependency Injector container for event adapter
     """
-    kafka_config = providers.Configuration(pydantic_settings=[KafkaSettings()])
+    kafka_config = providers.Configuration()
     # TODO: load from env
     # kafka_config.from_pydantic(KafkaSettings())
 
-    config = Config(
-        service_name="ujumbe",
-        default_adapter="kafka",
-        adapters=AdapterConfigs(
-            kafka=KafkaConfig(
-                # bootstrap_servers=[kafka_config.kafka_bootstrap_servers()],
-                # schema_registy=KafkaSchemaRegistryConfig(
-                #     schema_registry_url=kafka_config.kafka_schema_registry()
-                # )
-            )
-        )
-    )
-    kafka_event_stream = cast(AsyncEventStream, factory(config, adapter_name="aiokafka"))
-
     schema_registry = providers.Singleton(
         KafkaRegistry,
         params=KafkaSchemaRegistryConfig(url=kafka_config.kafka_schema_registry())

app/config/di/gateways_container.py → app/config/di/infra_container.py

@@ -4,16 +4,16 @@
 from app.settings import DatabaseSettings, TwilioSmsClientSettings
 
 
-class GatewaysContainer(containers.DeclarativeContainer):
+class InfraContainer(containers.DeclarativeContainer):
     """
     Dependency Injector container for Gateway services or 3rd party services used in the application
     """
 
-    db_config = providers.Configuration(pydantic_settings=[DatabaseSettings()])
+    db_config = providers.Configuration()
     # TODO: load settings from env directly
     # db_config.from_pydantic(settings=DatabaseSettings())
 
-    twilio_sms_config = providers.Configuration(pydantic_settings=[TwilioSmsClientSettings()])
+    twilio_sms_config = providers.Configuration()
     # TODO: load settings from env directly
     # twilio_sms_config.from_pydantic(TwilioSmsClientSettings())
 

app/config/di/services_container.py

@@ -11,7 +11,6 @@
 )
 from app.adapters.broker.producers.sms_sent_producer import SmsSentProducer
 from app.adapters.sms_svc.sms_service import UjumbeSmsService
-from app.settings import KafkaSettings
 
 
 class ServicesContainer(containers.DeclarativeContainer):
@@ -21,22 +20,22 @@ class ServicesContainer(containers.DeclarativeContainer):
     see https://github.com/ets-labs/python-dependency-injector for more details
     """
 
-    gateways = providers.DependenciesContainer()
-    event_stream_container = providers.DependenciesContainer()
+    infra = providers.DependenciesContainer()
+    event_stream = providers.DependenciesContainer()
 
-    kafka_config = providers.Configuration(pydantic_settings=[KafkaSettings()])
+    kafka_config = providers.Configuration()
     # TODO: load from env
     # kafka_config.from_pydantic(KafkaSettings())
 
     # Sms Received
     sms_received_consumer = providers.Factory(
         SmsReceivedConsumer,
-        kafka_consumer=event_stream_container.sms_received_protobuf_consumer,
+        kafka_consumer=event_stream.sms_received_protobuf_consumer,
     )
 
     sms_received_producer = providers.Factory(
         SmsReceivedProducer,
-        event_stream=event_stream_container.kafka_event_stream,
+        event_stream=event_stream.kafka_event_stream,
         topic=kafka_config.sms_received_topic(),
     )
 
@@ -45,34 +44,34 @@ class ServicesContainer(containers.DeclarativeContainer):
     # consumer
     sms_callback_received_consumer = providers.Factory(
         SmsCallbackReceivedProducer,
-        kafka_consumer=event_stream_container.sms_callback_received_protobuf_consumer,
+        kafka_consumer=event_stream.sms_callback_received_protobuf_consumer,
     )
 
     # producer
     sms_callback_received_producer = providers.Factory(
         SmsCallbackReceivedConsumer,
-        kafka_producer=event_stream_container.sms_callback_received_protobuf_producer,
+        kafka_producer=event_stream.sms_callback_received_protobuf_producer,
         topic=kafka_config.sms_callback_received_topic(),
     )
 
     # Sms Submitted
     sms_submitted_consumer = providers.Factory(
         SmsSubmittedConsumer,
-        kafka_consumer=event_stream_container.sms_submitted_protobuf_consumer,
+        kafka_consumer=event_stream.sms_submitted_protobuf_consumer,
     )
 
     sms_submitted_producer = providers.Factory(
         SmsSubmittedProducer,
-        kafka_producer=event_stream_container.sms_submitted_protobuf_producer,
+        kafka_producer=event_stream.sms_submitted_protobuf_producer,
         topic=kafka_config.sms_submitted_topic(),
     )
 
     # Sms Sent producer
 
     sms_sent_producer = providers.Factory(
         SmsSentProducer,
-        kafka_producer=event_stream_container.sms_sent_protobuf_producer,
+        kafka_producer=event_stream.sms_sent_protobuf_producer,
         topic=kafka_config.sms_sent_topic(),
     )
 
-    sms_service = providers.Factory(UjumbeSmsService, sms_client=gateways.sms_client)
+    sms_service = providers.Factory(UjumbeSmsService, sms_client=infra.sms_client)

app/config/event_adapter.py

@@ -0,0 +1,26 @@
+from typing import cast
+
+from eventmsg_adaptor.config.kafka import KafkaConfig, KafkaSchemaRegistryConfig, KafkaSecurityProtocolConfig
+from eventmsg_adaptor.event_streams import AsyncEventStream
+from eventmsg_adaptor import factory
+from eventmsg_adaptor.config import Config, AdapterConfigs
+
+from app.settings import get_kafka_settings, get_settings
+
+_settings = get_settings()
+_kafka_settings = get_kafka_settings()
+
+event_adapter_config = Config(
+    service_name=_settings.server_name,
+    default_adapter="kafka",
+    adapters=AdapterConfigs(
+        kafka=KafkaConfig(
+            bootstrap_servers=[_kafka_settings.kafka_bootstrap_servers],
+            schema_registry=KafkaSchemaRegistryConfig(
+                schema_registry_url=_kafka_settings.kafka_schema_registry
+            )
+        )
+    )
+)
+
+kafka_event_stream = cast(AsyncEventStream, factory(event_adapter_config, adapter_name="aiokafka"))

app/infra/telemetry/otel/metrics.py

@@ -2,10 +2,10 @@
 from opentelemetry.exporter.prometheus import PrometheusMetricReader
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry import metrics
-from app.settings import get_config
+from app.settings import get_settings
 from app.infra.logger import log as logger
 
-_settings = get_config()
+_settings = get_settings()
 
 
 def initialize_metrics():

app/infra/telemetry/otel/traces.py

@@ -2,10 +2,10 @@
 from opentelemetry.propagate import set_global_textmap
 from opentelemetry.sdk.trace import TracerProvider
 from sentry_sdk.integrations.opentelemetry import SentryPropagator, SentrySpanProcessor
-from app.settings import get_config
+from app.settings import get_settings
 from app.infra.logger import log as logger
 
-_settings = get_config()
+_settings = get_settings()
 
 
 def initialize_traces():

app/infra/telemetry/prometheus/__init__.py

@@ -1,8 +1,8 @@
 from prometheus_client import make_asgi_app
-from app.settings import get_config
+from app.settings import get_settings
 from app.infra.logger import log as logger
 
-_settings = get_config()
+_settings = get_settings()
 
 
 def setup_prometheus_client():

app/infra/telemetry/sentry/__init__.py

@@ -2,10 +2,10 @@
 from sentry_sdk.integrations.fastapi import FastApiIntegration
 from sentry_sdk.integrations.starlette import StarletteIntegration
 from sentry_sdk.integrations.loguru import LoguruIntegration
-from app.settings import get_config
+from app.settings import get_settings
 from app.infra.logger import log as logger
 
-_settings = get_config()
+_settings = get_settings()
 
 
 def _traces_sampler(sampling_context):

app/settings.py

@@ -153,8 +153,19 @@ class AppSettings(BaseSettings):
 
 
 @lru_cache()
-def get_config():
+def get_settings() -> AppSettings:
     """
     This is wrapped with lru_cache to ensure we don't continuously read from .env file on restarts
     """
+    load_dotenv()
     return AppSettings()
+
+
+@lru_cache()
+def get_kafka_settings() -> KafkaSettings:
+    """Returns Kafka settings
+    Returns:
+        KafkaSettings: Kafka settings
+    """
+    load_dotenv()
+    return KafkaSettings()

pyproject.toml

@@ -38,6 +38,7 @@ pydantic = "^2.4.2"
 botocore = "^1.31.69"
 boto3 = "^1.28.69"
 # TODO: remove botocore and boto3 once eventmsg-adaptor library has adapter bug fixed
+injector = "^0.21.0"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.1.0"

tests/__init__.py

@@ -2,7 +2,7 @@
 from fastapi.testclient import TestClient
 from httpx import AsyncClient
 from app import app
-from app.settings import AppSettings, get_config, SentrySettings
+from app.settings import AppSettings, get_settings, SentrySettings
 
 base_url = "http://ujumbe-test-server"
 
@@ -13,7 +13,7 @@ class BaseTestCase(unittest.TestCase):
     """
 
     def setUp(self):
-        app.dependency_overrides[get_config] = self._get_settings_override()
+        app.dependency_overrides[get_settings] = self._get_settings_override()
         self.app = app
         self.test_client = TestClient(app=app, base_url=base_url)
         self.async_client = AsyncClient(app=app, base_url=base_url)