Closed insani7y closed 1 month ago
Perhaps, it's a good idea to change abstractions a bit to something like this:
from contextlib import ExitStack, contextmanager
from functools import partial
from typing import (
TYPE_CHECKING,
Any,
Callable,
ContextManager,
Generator,
Protocol,
TypedDict,
TypeVar,
)
if TYPE_CHECKING:
def merge_dict_configs(a: Any, b: Any) -> Any: ...
def merge_pydantic_configs(a: Any, b: Any) -> Any: ...
class Litestar: ...
class OpenTelemetryInstrumentationMiddleware: ...
class LitestarOpentelemetryConfig: ...
InstrumentConfig = TypeVar("InstrumentConfig")
InstrumentResult = TypeVar("InstrumentResult")
class Instrument(ContextManager, Protocol[InstrumentConfig, InstrumentResult]):
def __enter__(self, config: InstrumentConfig) -> InstrumentResult: ...
class OpenTelemetryConfig: ...
class OpenTelemetryInstrumentResult(TypedDict):
tracer_provider: Any
exclude_urls: list[str]
@contextmanager
def instrument_opentelemetry(
config: OpenTelemetryConfig,
) -> Generator[OpenTelemetryInstrumentResult, None, None]:
yield {"tracer_provider": ..., "exclude_urls": []}
ApplicationT = TypeVar("ApplicationT")
ApplicationConfigT = TypeVar("ApplicationConfigT")
class Configurator(Protocol[ApplicationT, ApplicationConfigT]):
def make_application_from_config(
self, config: ApplicationConfigT
) -> ApplicationT: ...
def configure_opentelemetry(
self, result: OpenTelemetryInstrumentResult
) -> None: ...
def configure_teardown(self, teardown: Callable[[], None]) -> None: ...
class Settings: ...
def bootstrap(
configurator: Configurator[ApplicationT, ApplicationConfigT],
settings: Settings,
*,
application_config: ApplicationConfigT | None = None,
opentelemetry_config: OpenTelemetryConfig | None = None,
) -> ApplicationT:
exit_stack = ExitStack()
full_application_config = application_config
for instrument_config, instrument, framework_adapter in [
(
opentelemetry_config,
instrument_opentelemetry,
configurator.configure_opentelemetry,
)
]:
config_piece = framework_adapter(
instrument(merge_pydantic_configs(settings, instrument_config))
)
full_application_config = merge_dict_configs(
full_application_config, exit_stack.enter_context(config=config_piece)
)
full_application_config = merge_dict_configs(
full_application_config, configurator.configure_teardown(exit_stack.close)
)
return configurator.make_application_from_config(full_application_config)
class LitestarConfigurator(Configurator[Litestar, dict[str, Any]]):
def make_application_from_config(self, config: dict[str, Any]) -> Any:
return Litestar(**config)
def configure_opentelemetry(self, result: OpenTelemetryInstrumentResult) -> None:
return {
"middleware": OpenTelemetryInstrumentationMiddleware(
LitestarOpentelemetryConfig(
tracer_provider=result["tracer_provider"],
exclude=result["exclude_urls"],
),
),
}
def configure_teardown(self, teardown: Callable[[], None]) -> None:
return {"on_shutdown": [teardown]}
bootstrap_litestar = partial(bootstrap, configurator=LitestarConfigurator())
def main() -> None:
application = bootstrap_litestar(Settings())
Tests are working! But there is much to be done still
Once you merge this PR into your default branch, you're all set! Codecov will compare coverage reports and display results in all future pull requests.
:information_source: You can also turn on project coverage checks and project coverage reporting on Pull Request comment
Thanks for integrating Codecov - We've got you covered :open_umbrella:
First at all, thank you for your work. Package are really getting better. I like current structure and package usage.
However, I have some suggestions and questions, and I am hopeful that they will only make better the package.
bootstrap_before
and bootstrap_after
. Names of this methods dont tell me about ist usages. It provides some config data to app, but i can know this only if i read usage from docs, code or docstrings of this methods. I dont know how to name it better by now, but if any thoughts come to me - i will write them down.LGTM!
Tests are broken yet.