edraft
ddc62dfb9a
Some checks failed
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 19s
Build on push / query (push) Successful in 19s
Build on push / dependency (push) Successful in 17s
Build on push / application (push) Successful in 15s
Build on push / database (push) Successful in 18s
Build on push / mail (push) Successful in 19s
Build on push / translation (push) Successful in 23s
Build on push / auth (push) Successful in 16s
Build on push / api (push) Failing after 14s
66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
import asyncio
|
|
from typing import Type, Optional, TypeVar, Generic
|
|
|
|
from cpl.application.abc.application_abc import ApplicationABC
|
|
from cpl.application.abc.application_extension_abc import ApplicationExtensionABC
|
|
from cpl.application.abc.startup_abc import StartupABC
|
|
from cpl.application.abc.startup_extension_abc import StartupExtensionABC
|
|
from cpl.application.host import Host
|
|
from cpl.dependency.service_collection import ServiceCollection
|
|
|
|
TApp = TypeVar("TApp", bound=ApplicationABC)
|
|
|
|
|
|
class ApplicationBuilder(Generic[TApp]):
|
|
|
|
def __init__(self, app: Type[ApplicationABC]):
|
|
assert app is not None, "app must not be None"
|
|
assert issubclass(app, ApplicationABC), "app must be an subclass of ApplicationABC or its subclass"
|
|
|
|
self._app = app if app is not None else ApplicationABC
|
|
|
|
self._services = ServiceCollection()
|
|
|
|
self._startup: Optional[StartupABC] = None
|
|
self._app_extensions: list[Type[ApplicationExtensionABC]] = []
|
|
self._startup_extensions: list[Type[StartupExtensionABC]] = []
|
|
|
|
self._async_loop = asyncio.get_event_loop()
|
|
|
|
@property
|
|
def services(self) -> ServiceCollection:
|
|
return self._services
|
|
|
|
@property
|
|
def service_provider(self):
|
|
return self._services.build()
|
|
|
|
def with_startup(self, startup: Type[StartupABC]) -> "ApplicationBuilder":
|
|
self._startup = startup
|
|
return self
|
|
|
|
def with_extension(
|
|
self,
|
|
extension: Type[ApplicationExtensionABC | StartupExtensionABC],
|
|
) -> "ApplicationBuilder":
|
|
if (issubclass(extension, ApplicationExtensionABC)) and extension not in self._app_extensions:
|
|
self._app_extensions.append(extension)
|
|
elif (issubclass(extension, StartupExtensionABC)) and extension not in self._startup_extensions:
|
|
self._startup_extensions.append(extension)
|
|
|
|
return self
|
|
|
|
def build(self) -> TApp:
|
|
for extension in self._startup_extensions:
|
|
Host.run(extension.configure_configuration)
|
|
Host.run(extension.configure_services, self._services)
|
|
|
|
if self._startup is not None:
|
|
Host.run(self._startup.configure_configuration)
|
|
Host.run(self._startup.configure_services, self._services)
|
|
|
|
for extension in self._app_extensions:
|
|
Host.run(extension.run, self.service_provider)
|
|
|
|
return self._app(self.service_provider)
|