Merge pull request #34 from modern-python/build-config-from-dict-and-object

allow building configs from dict and object

Author: lesnik512

lite_bootstrap/instruments/base.py

@@ -1,5 +1,8 @@
 import abc
 import dataclasses
+import typing
+
+import typing_extensions
 
 
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
@@ -10,6 +13,21 @@ class BaseConfig:
     service_environment: str | None = None
     service_debug: bool = True
 
+    @classmethod
+    def from_dict(cls, data: dict[str, typing.Any]) -> typing_extensions.Self:
+        field_names = {f.name for f in dataclasses.fields(cls)}
+        return cls(**{k: v for k, v in data.items() if k in field_names})
+
+    @classmethod
+    def from_object(cls, obj: object) -> typing_extensions.Self:
+        prepared_data = {}
+        field_names = {f.name for f in dataclasses.fields(cls)}
+
+        for field in field_names:
+            if (value := getattr(obj, field, None)) is not None:
+                prepared_data[field] = value
+        return cls(**prepared_data)
+
 
 @dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
 class BaseInstrument(abc.ABC):

tests/test_config.py

@@ -0,0 +1,53 @@
+import dataclasses
+
+from opentelemetry.sdk.trace.export import ConsoleSpanExporter
+
+from lite_bootstrap import FastAPIConfig
+from lite_bootstrap.instruments.base import BaseConfig
+from tests.conftest import CustomInstrumentor
+
+
+def test_config_from_dict() -> None:
+    raw_config = {
+        "service_name": "microservice",
+        "service_version": "2.0.0",
+        "service_environment": "test",
+        "service_debug": False,
+        "cors_allowed_origins": ["http://test"],
+        "health_checks_path": "/custom-health/",
+        "logging_buffer_capacity": 0,
+        "opentelemetry_endpoint": "otl",
+        "opentelemetry_instrumentors": [CustomInstrumentor()],
+        "opentelemetry_span_exporter": ConsoleSpanExporter(),
+        "prometheus_metrics_path": "/custom-metrics/",
+        "sentry_dsn": "https://testdsn@localhost/1",
+        "swagger_offline_docs": True,
+        "extra_key": "extra_value",
+    }
+    config = FastAPIConfig.from_dict(raw_config)
+
+    for field in dataclasses.fields(FastAPIConfig):
+        if field.name in raw_config:
+            assert getattr(config, field.name) == raw_config[field.name]
+
+
+def test_config_from_object() -> None:
+    big_config = FastAPIConfig(
+        service_name="microservice",
+        service_version="2.0.0",
+        service_environment="test",
+        service_debug=False,
+        cors_allowed_origins=["http://test"],
+        health_checks_path="/custom-health/",
+        logging_buffer_capacity=0,
+        opentelemetry_endpoint="otl",
+        opentelemetry_instrumentors=[CustomInstrumentor()],
+        opentelemetry_span_exporter=ConsoleSpanExporter(),
+        prometheus_metrics_path="/custom-metrics/",
+        sentry_dsn="https://testdsn@localhost/1",
+        swagger_offline_docs=True,
+    )
+
+    short_config = BaseConfig.from_object(big_config)
+    for field in dataclasses.fields(BaseConfig):
+        assert getattr(short_config, field.name) == getattr(big_config, field.name)