import inspect
from onegov.server import errors
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
from .application import Application
from .config import ApplicationConfig
[docs]
class CachedApplication:
""" Wraps an application class with a configuration, returning a new
instance the first time `get()` is called and the same instance very
time after that.
"""
[docs]
instance: 'Application | None'
def __init__(
self,
application_class: type['Application'],
namespace: str,
configuration: dict[str, Any] | None = None
):
[docs]
self.application_class = application_class
[docs]
self.configuration = configuration or {}
[docs]
self.namespace = namespace
self.instance = None
[docs]
def get(self) -> 'Application':
if self.instance is None:
self.instance = self.application_class()
self.instance.namespace = self.namespace
self.instance.configure_application(**self.configuration)
return self.instance
[docs]
class ApplicationCollection:
""" Keeps a list of applications and their roots.
The applications are registered lazily and only instantiated/configured
once the `get()` is called.
"""
[docs]
applications: dict[str, CachedApplication]
def __init__(
self,
applications: 'Iterable[ApplicationConfig] | None' = None
):
self.applications = {}
for a in applications or ():
self.register(
a.root, a.application_class, a.namespace, a.configuration)
[docs]
def register(
self,
root: str,
application_class: type['Application'],
namespace: str,
configuration: dict[str, Any] | None = None
) -> None:
""" Registers the given path for the given application_class and
configuration.
"""
if root in self.applications:
raise errors.ApplicationConflictError(
"tried to register '{}' twice".format(root))
self.applications[root] = CachedApplication(
application_class, namespace, configuration
)
[docs]
def get(self, root: str) -> 'Application | None':
""" Returns the applicaton for the given path, creating a new instance
if none exists already.
"""
application = self.applications.get(root)
if application is None:
return None
else:
return application.get()
[docs]
def morepath_applications(self) -> 'Iterator[CachedApplication]':
""" Iterates through the applications that depend on morepath. """
for app in self.applications.values():
for base in inspect.getmro(app.application_class):
if base.__module__.startswith('morepath.'):
yield app
break