diff --git a/lite_bootstrap/instruments/base.py b/lite_bootstrap/instruments/base.py index c1b4d9e..04d2410 100644 --- a/lite_bootstrap/instruments/base.py +++ b/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): diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..c49871c --- /dev/null +++ b/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)