Server framework

This module is completely self contained and could be extracted to its own project.

Sub-modules

server.core.dependency_injector

server.core.service

Functions

def create_services(injectables: dict[str, object] = {}) ‑> dict[str, Service]

Resolve service dependencies and instantiate each service. This should only be called once.

Classes

class DependencyInjector

Does dependency injection.

Dependencies are resolved by parameter name. So if a class has init method

def __init__(self, hello, world):
    pass

the injector will look for two dependencies, called hello and world. These could either be an object registered with add_injectables, or an instance of another class passed to build_classes.

Injected arguments are only ever constructed once. So if two classes both depend on an object called hello, then they will both receive the same instance of the object called hello (whether that is an injectable, or another class in the class list).

Examples

Create a class that depends on some external injectable.

>>> class SomeClass(object):
...     def __init__(self, external):
...         self.external = external

Create a class that depends on the first class.

>>> class SomeOtherClass(object):
...     def __init__(self, some_class):
...         self.some_class = some_class

Do the dependency injection.

>>> injector = DependencyInjector()
>>> injector.add_injectables(external=object())
>>> classes = injector.build_classes({
...     "some_class": SomeClass,
...     "other": SomeOtherClass
... })

>>> assert isinstance(classes["some_class"], SomeClass)
>>> assert isinstance(classes["other"], SomeOtherClass)
>>> assert classes["other"].some_class is classes["some_class"]

Expand source code

class DependencyInjector(object):
    """
    Does dependency injection.

    Dependencies are resolved by parameter name. So if a class has init method
    ```
    def __init__(self, hello, world):
        pass
    ```
    the injector will look for two dependencies, called `hello` and `world`.
    These could either be an object registered with `add_injectables`, or an
    instance of another class passed to `build_classes`.

    Injected arguments are only ever constructed once. So if two classes both
    depend on an object called `hello`, then they will both receive the same
    instance of the object called `hello` (whether that is an injectable, or
    another class in the class list).

    # Examples
    Create a class that depends on some external injectable.
    >>> class SomeClass(object):
    ...     def __init__(self, external):
    ...         self.external = external

    Create a class that depends on the first class.
    >>> class SomeOtherClass(object):
    ...     def __init__(self, some_class):
    ...         self.some_class = some_class

    Do the dependency injection.
    >>> injector = DependencyInjector()
    >>> injector.add_injectables(external=object())
    >>> classes = injector.build_classes({
    ...     "some_class": SomeClass,
    ...     "other": SomeOtherClass
    ... })

    >>> assert isinstance(classes["some_class"], SomeClass)
    >>> assert isinstance(classes["other"], SomeOtherClass)
    >>> assert classes["other"].some_class is classes["some_class"]
    """

    def __init__(self) -> None:
        # Objects which are available to the constructors of injected objects
        self.injectables: dict[str, object] = {}

    def add_injectables(
        self, injectables: dict[str, object] = {}, **kwargs: object
    ) -> None:
        """
        Register additional objects that can be requested by injected classes.
        """
        self.injectables.update(injectables)
        self.injectables.update(kwargs)

    def build_classes(
        self, classes: dict[str, type] = {}, **kwargs: type
    ) -> dict[str, object]:
        """
        Resolve dependencies by name and instantiate each class.
        """
        # kwargs is temporary so we won't be messing with the caller's data
        kwargs.update(classes)
        classes = kwargs

        dep = self._make_dependency_graph(classes)
        # Can get away with a shallow copy because dep values are not modified
        # in-place.
        param_map = dep.copy()

        instances = self._build_classes_from_dependencies(
            dep, classes, param_map
        )
        self.add_injectables(**instances)
        return instances

    def _make_dependency_graph(self, classes: dict[str, type]) -> DependencyGraph:
        """
        Build dependency graph
        """
        graph: DependencyGraph = defaultdict(list)
        for name in self.injectables:
            graph[name] = []

        for obj_name, klass in classes.items():
            signature = inspect.signature(klass.__init__)
            # Strip off the `self` parameter
            params = list(signature.parameters.values())[1:]
            graph[obj_name] = [param.name for param in params]

        return graph

    def _build_classes_from_dependencies(
        self,
        dep: DependencyGraph,
        classes: dict[str, type],
        param_map: dict[str, list[str]]
    ) -> dict[str, object]:
        """
        Tries to build all classes in the dependency graph. Raises RuntimeError
        if some dependencies are not available or there was a cyclic dependency.
        """
        instances: dict[str, object] = {}
        resolved = ChainMap(instances, self.injectables)

        while True:
            if not dep:
                return instances

            # Find all services with no dependencies (leaves of our graph)
            leaves = [
                name for name, dependencies in dep.items() if not dependencies
            ]
            if not leaves:
                # Find which dependencies could not be resolved
                missing = {
                    d for dependencies in dep.values()
                    for d in dependencies if d not in dep
                }
                if missing:
                    raise RuntimeError(
                        f"Some dependencies could not be resolved: {missing}"
                    )
                else:
                    cycle = tuple(dep.keys())
                    raise RuntimeError(
                        f"Could not resolve cyclic dependency: {cycle}"
                    )

            # Build all objects with no dependencies
            for obj_name in leaves:
                if obj_name not in resolved:
                    klass = classes[obj_name]
                    param_names = param_map[obj_name]

                    # Build instances using the objects we've resolved so far
                    instances[obj_name] = klass(**{
                        param: resolved[param]
                        for param in param_names
                    })
                elif obj_name in classes:
                    instances[obj_name] = resolved[obj_name]

                del dep[obj_name]

            # Remove leaves from the dependency graph
            for name, dependencies in dep.items():
                dep[name] = [d for d in dependencies if d not in leaves]

Methods

def add_injectables(self, injectables: dict[str, object] = {}, **kwargs: object) ‑> None

Register additional objects that can be requested by injected classes.

def build_classes(self, classes: dict[str, type] = {}, **kwargs: type) ‑> dict[str, object]

Resolve dependencies by name and instantiate each class.

class Service

All services should inherit from this class.

Services are singleton objects which manage some server task.

Expand source code

class Service():
    """
    All services should inherit from this class.

    Services are singleton objects which manage some server task.
    """
    def __init_subclass__(cls, name: Optional[str] = None, **kwargs: Any):
        """
        For tracking which services have been defined.
        """
        super().__init_subclass__(**kwargs)
        arg_name = name or snake_case(cls.__name__)
        service_registry[arg_name] = cls

    async def initialize(self) -> None:
        """
        Called once while the server is starting.
        """
        pass  # pragma: no cover

    async def graceful_shutdown(self) -> None:
        """
        Called once after the graceful shutdown period is initiated.

        This signals that the service should stop accepting new events but
        continue to wait for existing ones to complete normally. The hook
        funciton `shutdown` will be called after the grace period has ended to
        fully shutdown the service.
        """
        pass  # pragma: no cover

    async def shutdown(self) -> None:
        """
        Called once after the server received the shutdown signal.
        """
        pass  # pragma: no cover

    def on_connection_lost(self, conn) -> None:
        """
        Called every time a connection ends.
        """
        pass  # pragma: no cover

Subclasses

• BroadcastService
• ConfigurationService
• GameService
• GeoIpService
• LadderService
• ViolationService
• MessageQueueService
• OAuthService
• PartyService
• PlayerService
• RatingService
• AchievementService
• EventService
• GameStatsService

Methods

async def graceful_shutdown(self) ‑> None

Called once after the graceful shutdown period is initiated.

This signals that the service should stop accepting new events but continue to wait for existing ones to complete normally. The hook funciton shutdown will be called after the grace period has ended to fully shutdown the service.

async def initialize(self) ‑> None

Called once while the server is starting.

def on_connection_lost(self, conn) ‑> None

Called every time a connection ends.

async def shutdown(self) ‑> None

Called once after the server received the shutdown signal.