📘 Technical Design 021 — Application Entry Point and Bootstrap Orchestration¶

# Must manually compose scanning and bootstrap
scanner = DiPackageScanner(reflection)
plan_builder = BootstrapPlanBuilder()

# Manual scanning
plan = plan_builder.scan_packages(["myapp.services", "myapp.repositories"])

# Manual bootstrap
runtime = BootstrapRuntime()
container = DiContainer()
runtime.bootstrap(container, plan)

# Manual lifecycle
app = MyApplication()
app.run(container)

from dataclasses import dataclass
from typing import Callable, Sequence, TypeVar

from loryth.common.annotations import Annotation
from loryth.common.annotations import AnnotationContext

ApplicationT = TypeVar("ApplicationT", bound="ApplicationBase")


@dataclass(frozen=True)
class ApplicationConfiguration:
    """Configuration captured by @application decorator.

    :param scan_packages: Import paths to scan for annotated services.
    :type scan_packages: Sequence[str]
    :param exclude_packages: Packages to exclude during scanning.
    :type exclude_packages: Sequence[str]
    :param validate_dependencies: Whether dependency graphs are validated.
    :type validate_dependencies: bool
    :param eager_prepare: Whether singletons are eagerly prepared.
    :type eager_prepare: bool
    """

    scan_packages: Sequence[str]
    exclude_packages: Sequence[str] = ()
    validate_dependencies: bool = True
    eager_prepare: bool = True


class ApplicationError(Exception):
    """Raised when application orchestration fails."""


class ApplicationAnnotation(Annotation):
    """Annotation storing application configuration metadata."""

    def __init__(self, config: ApplicationConfiguration) -> None:
        """Initialize the application annotation.

        :param config: Application configuration extracted from the decorator.
        :type config: ApplicationConfiguration
        """
        super().__init__(name="application")
        self.config = config


class ApplicationBase:
    """Base class for DI-managed application entry points."""

    def run(self) -> None:
        """Start the application.

        Dependencies are provided via injection.

        :raises NotImplementedError: If a subclass does not implement the method.
        """
        raise NotImplementedError


def application(
    *,
    scan_packages: Sequence[str],
    exclude_packages: Sequence[str] = (),
    validate_dependencies: bool = True,
    eager_prepare: bool = True,
) -> Callable[[type[ApplicationT]], type[ApplicationT]]:
    """Decorator capturing application configuration metadata.

    :param scan_packages: Packages to include during discovery.
    :type scan_packages: Sequence[str]
    :param exclude_packages: Packages removed from discovery.
    :type exclude_packages: Sequence[str]
    :param validate_dependencies: Toggle for dependency validation.
    :type validate_dependencies: bool
    :param eager_prepare: Toggle for eager singleton preparation.
    :type eager_prepare: bool
    :return: Decorated application class.
    :rtype: type[ApplicationT]
    """

    def decorator(cls: type[ApplicationT]) -> type[ApplicationT]:
        """Attach application configuration metadata to the decorated class.

        :param cls: Application class being decorated.
        :type cls: type[ApplicationT]
        :return: Annotated application class.
        :rtype: type[ApplicationT]
        """
        config = ApplicationConfiguration(
            scan_packages=scan_packages,
            exclude_packages=exclude_packages,
            validate_dependencies=validate_dependencies,
            eager_prepare=eager_prepare,
        )
        AnnotationContext.annotate(cls, ApplicationAnnotation(config))
        return cls

    return decorator

from importlib import import_module

from loryth.common.annotations import find_annotation
from loryth.common.di.bootstrap import BootstrapPlanBuilder, BootstrapRuntime
from loryth.common.di.container import Container


class ApplicationRuntime:
    """Coordinates bootstrap runtime and application execution."""

    def __init__(self, bootstrap_runtime: BootstrapRuntime | None = None) -> None:
        """Create an application runtime coordinator.

        :param bootstrap_runtime: Optional runtime used to execute bootstrap stages.
        :type bootstrap_runtime: BootstrapRuntime | None
        """
        self._bootstrap_runtime = bootstrap_runtime or BootstrapRuntime()

    @classmethod
    def launch(cls, path: str, *, bootstrap_runtime: BootstrapRuntime | None = None) -> None:
        """Convenience helper that creates a runtime and executes the application.

        :param path: Dotted module path in ``module:Class`` format.
        :type path: str
        :param bootstrap_runtime: Optional runtime used for bootstrap orchestration.
        :type bootstrap_runtime: BootstrapRuntime | None
        :raises ApplicationError: If the application cannot be resolved or started.
        """
        runtime = cls(bootstrap_runtime=bootstrap_runtime)
        runtime.run(path)

    def run(self, path: str) -> None:
        """Bootstrap, resolve, and execute an application entry point.

        :param path: Dotted module path in ``module:Class`` format.
        :type path: str
        :raises ApplicationError: If resolution or execution fails.
        """
        application_type = self._resolve_application(path)
        config = self._read_configuration(application_type)
        container = self._bootstrap_container(config)

        application = container.resolve(application_type)
        application.run()

    def _resolve_application(self, path: str) -> type[ApplicationBase]:
        """Load the application class from a module path.

        :param path: Dotted module path in ``module:Class`` format.
        :type path: str
        :raises ApplicationError: If the module or class cannot be located.
        :return: Application class that inherits from ``ApplicationBase``.
        :rtype: type[ApplicationBase]
        """
        try:
            module_path, class_name = path.split(":")
            module = import_module(module_path)
            application = getattr(module, class_name)
        except (ValueError, AttributeError, ModuleNotFoundError) as exc:
            raise ApplicationError(f"Cannot resolve application entry point '{path}': {exc}") from exc

        if not issubclass(application, ApplicationBase):
            raise ApplicationError(f"Application '{path}' must inherit from ApplicationBase")

        return application

    def _read_configuration(self, application_type: type[ApplicationBase]) -> ApplicationConfiguration:
        """Read configuration metadata attached by @application.

        :param application_type: Application class being resolved.
        :type application_type: type[ApplicationBase]
        :raises ApplicationError: If the application annotation is missing.
        :return: Configuration declared via the ``@application`` decorator.
        :rtype: ApplicationConfiguration
        """
        annotation = find_annotation(application_type, ApplicationAnnotation)
        if annotation is None:
            raise ApplicationError(
                f"Application '{application_type.__module__}:{application_type.__name__}' is missing @application metadata",
            )
        return annotation.config

    def _bootstrap_container(self, config: ApplicationConfiguration) -> Container:
        """Execute bootstrap plan based on decorator metadata.

        :param config: Application configuration extracted from annotations.
        :type config: ApplicationConfiguration
        :return: Container prepared with all discovered services.
        :rtype: Container
        """
        plan_builder = BootstrapPlanBuilder.create()
        plan = plan_builder.scan_packages(
            config.scan_packages,
            exclude=config.exclude_packages,
        ).build()

        container = Container()
        self._bootstrap_runtime.bootstrap(
            container,
            plan,
            validate_dependencies=config.validate_dependencies,
            eager_prepare=config.eager_prepare,
        )

        return container

@application(scan_packages=["orders.services"])
class OrderApplication(ApplicationBase):
    """Launch order-processing services."""

    def __init__(
        self,
        http_server: HttpServer,
        scheduler: JobScheduler,
        consumer: MessageConsumer,
    ) -> None:
        """Store injected services required by the application.

        :param http_server: HTTP server responsible for serving APIs.
        :type http_server: HttpServer
        :param scheduler: Scheduler that manages background jobs.
        :type scheduler: JobScheduler
        :param consumer: Message consumer handling inbound events.
        :type consumer: MessageConsumer
        """
        self._http_server = http_server
        self._scheduler = scheduler
        self._consumer = consumer

    def run(self) -> None:
        """Start all long-running services."""
        self._consumer.start()       # RabbitMQ consumer owns its thread
        self._scheduler.start()      # Scheduler manages its own worker pool
        self._http_server.start()    # Web server blocks the main thread

class SafeApplication(ApplicationBase):
    """Fails fast when startup prerequisites are missing."""

    def __init__(self, health_check: StartupHealthCheck) -> None:
        """Store the injected health check dependency.

        :param health_check: Service that validates environment readiness.
        :type health_check: StartupHealthCheck
        """
        self._health_check = health_check

    def run(self) -> None:
        """Verify readiness and emit clear diagnostics.

        :raises ApplicationError: If the application fails to start.
        """
        if not self._health_check.is_ready():
            raise ApplicationError("Startup health check failed")

        try:
            self._start_components()
        except Exception as exc:  # noqa: BLE001 - surfaced via ApplicationError
            raise ApplicationError(f"Application startup failed: {exc}") from exc

    def _start_components(self) -> None:
        """Private helper invoked during run().

        :raises ApplicationError: If a component fails during startup.
        """
        ...

from typing import Protocol

from loryth.common.di.application import ApplicationBase, ApplicationRuntime, application
from loryth.common.di.injection import autowired


class InventoryApi(Protocol):
    """Protocol for the inventory HTTP API."""

    def start(self) -> None: ...


class LoggerProtocol(Protocol):
    """Protocol for structured logging."""

    def info(self, message: str) -> None: ...


@application(scan_packages=["myapp.services"])
class InventoryApplication(ApplicationBase):
    """Launches the inventory API with injected dependencies."""

    def __init__(self, api: InventoryApi) -> None:
        self._api = api
        self._logger: Annotated[LoggerProtocol, Autowired("inventory")]

    def run(self) -> None:
        """Start the API service."""
        self._logger.info("Starting inventory service")
        self._api.start()


if __name__ == "__main__":
    ApplicationRuntime.launch("myapp.entry:InventoryApplication")

from loryth.common.di.application import ApplicationBase, ApplicationRuntime, application


@application(scan_packages=["payments.services"])
class PaymentApplication(ApplicationBase):
    """Coordinates background services that manage their own threads."""

    def __init__(
        self,
        consumer: PaymentConsumer,
        scheduler: ReconciliationScheduler,
        api: PaymentsApi,
    ) -> None:
        self._consumer = consumer
        self._scheduler = scheduler
        self._api = api

    def run(self) -> None:
        """Launch long-running components."""
        self._consumer.start()       # RabbitMQ consumer owns its thread
        self._scheduler.start()      # Scheduler manages its worker pool
        self._api.serve_forever()    # Blocking HTTP server call


if __name__ == "__main__":
    ApplicationRuntime.launch("payments.app:PaymentApplication")

import argparse

from loryth.common.di.application import ApplicationBase, ApplicationRuntime, application


@application(scan_packages=["reports.services"])
class ReportCli(ApplicationBase):
    """Generates reports from the command line."""

    def __init__(self, generator: ReportGenerator) -> None:
        self._generator = generator

    def run(self) -> None:
        """Parse CLI arguments and execute the command."""
        parser = argparse.ArgumentParser()
        parser.add_argument("--output", required=True)
        options = parser.parse_args()

        self._generator.generate(output_path=options.output)


if __name__ == "__main__":
    ApplicationRuntime.launch("reports.cli:ReportCli")

tests/unit/di/application/
├── test_application_decorator.py        # @application metadata capture
├── test_application_runtime.py          # launch() error handling and flow control
└── test_application_configuration.py    # Validation of ApplicationConfiguration defaults

tests/integration/di/
├── test_application_launch.py      # End-to-end launch via runtime helper
├── test_application_injection.py   # Constructor + autowired injection flows
└── test_runtime_container_bridge.py   # Coordination with bootstrap/container components

tests/regression/di/
├── test_manual_bootstrap_compat.py  # Manual bootstrap remains functional
└── test_entrypoint_migration.py    # Migration from manual to runtime entry points