feat(ServiceFactory): dynamic service registration (#579)

craft_application/application.py

@@ -407,7 +407,7 @@ def project(self) -> models.Project:
 
     def is_managed(self) -> bool:
         """Shortcut to tell whether we're running in managed mode."""
-        return self.services.ProviderClass.is_managed()
+        return self.services.get_class("provider").is_managed()
 
     def run_managed(self, platform: str | None, build_for: str | None) -> None:
         """Run the application in a managed instance."""

craft_application/services/fetch.py

@@ -83,7 +83,7 @@ def setup(self) -> None:
         """Start the fetch-service process with proper arguments."""
         super().setup()
 
-        if not self._services.ProviderClass.is_managed():
+        if not self._services.get_class("provider").is_managed():
             # Early fail if the fetch-service is not installed.
             fetch.verify_installed()
 

craft_application/services/service_factory.py

@@ -1,4 +1,4 @@
-# Copyright 2023 Canonical Ltd.
+# Copyright 2023-2024 Canonical Ltd.
 #
 # This program is free software: you can redistribute it and/or modify it
 # under the terms of the GNU Lesser General Public License version 3, as
@@ -15,14 +15,41 @@
 from __future__ import annotations
 
 import dataclasses
+import importlib
+import re
 import warnings
-from typing import TYPE_CHECKING, Any
+from typing import (
+    TYPE_CHECKING,
+    Annotated,
+    Any,
+    ClassVar,
+    Literal,
+    TypeVar,
+    cast,
+    overload,
+)
+
+import annotated_types
 
 from craft_application import models, services
 
 if TYPE_CHECKING:
     from craft_application.application import AppMetadata
 
+_DEFAULT_SERVICES = {
+    "config": "ConfigService",
+    "fetch": "FetchService",
+    "init": "InitService",
+    "lifecycle": "LifecycleService",
+    "provider": "ProviderService",
+    "remote_build": "RemoteBuildService",
+    "request": "RequestService",
+}
+_CAMEL_TO_PYTHON_CASE_REGEX = re.compile(r"(?<!^)(?=[A-Z])")
+
+T = TypeVar("T")
+_ClassName = Annotated[str, annotated_types.Predicate(lambda x: x.endswith("Class"))]
+
 
 @dataclasses.dataclass
 class ServiceFactory:
@@ -35,22 +62,24 @@ class ServiceFactory:
     and possibly have its existing service classes overridden.
     """
 
-    app: AppMetadata
-
-    PackageClass: type[services.PackageService]
-    LifecycleClass: type[services.LifecycleService] = services.LifecycleService
-    ProviderClass: type[services.ProviderService] = services.ProviderService
-    RemoteBuildClass: type[services.RemoteBuildService] = services.RemoteBuildService
-    RequestClass: type[services.RequestService] = services.RequestService
-    ConfigClass: type[services.ConfigService] = services.ConfigService
-    FetchClass: type[services.FetchService] = services.FetchService
-    InitClass: type[services.InitService] = services.InitService
+    _service_classes: ClassVar[
+        dict[str, tuple[str, str] | type[services.AppService]]
+    ] = {}
 
+    # These exist so that child ServiceFactory classes can use them.
+    app: AppMetadata
+    PackageClass: type[services.PackageService] = None  # type: ignore[assignment]
+    LifecycleClass: type[services.LifecycleService] = None  # type: ignore[assignment]
+    ProviderClass: type[services.ProviderService] = None  # type: ignore[assignment]
+    RemoteBuildClass: type[services.RemoteBuildService] = None  # type: ignore[assignment]
+    RequestClass: type[services.RequestService] = None  # type: ignore[assignment]
+    ConfigClass: type[services.ConfigService] = None  # type: ignore[assignment]
+    FetchClass: type[services.FetchService] = None  # type: ignore[assignment]
+    InitClass: type[services.InitService] = None  # type: ignore[assignment]
     project: models.Project | None = None
 
     if TYPE_CHECKING:
         # Cheeky hack that lets static type checkers report the correct types.
-        # Any apps that add their own services should do this too.
         package: services.PackageService = None  # type: ignore[assignment]
         lifecycle: services.LifecycleService = None  # type: ignore[assignment]
         provider: services.ProviderService = None  # type: ignore[assignment]
@@ -62,6 +91,68 @@ class ServiceFactory:
 
     def __post_init__(self) -> None:
         self._service_kwargs: dict[str, dict[str, Any]] = {}
+        self._services: dict[str, services.AppService] = {}
+
+        factory_dict = dataclasses.asdict(self)
+        for cls_name, value in factory_dict.items():
+            if cls_name.endswith("Class"):
+                if value is not None:
+                    identifier = _CAMEL_TO_PYTHON_CASE_REGEX.sub(
+                        "_", cls_name[:-5]
+                    ).lower()
+                    warnings.warn(
+                        f'Registering services on service factory instantiation is deprecated. Use ServiceFactory.register("{identifier}", {value.__name__}) instead.',
+                        category=DeprecationWarning,
+                        stacklevel=3,
+                    )
+                    self.register(identifier, value)
+                setattr(self, cls_name, self.get_class(cls_name))
+
+        if "package" not in self._service_classes:
+            raise TypeError(
+                "A PackageService must be registered before creating the ServiceFactory."
+            )
+
+    @classmethod
+    def register(
+        cls,
+        name: str,
+        service_class: type[services.AppService] | str,
+        *,
+        module: str | None = None,
+    ) -> None:
+        """Register a service class with a given name.
+
+        :param name: the name to call the service class.
+        :param service_class: either a service class or a string that names the service
+            class.
+        :param module: If service_class is a string, the module from which to import
+            the service class.
+        """
+        if isinstance(service_class, str):
+            if module is None:
+                raise KeyError("Must set module if service_class is set by name.")
+            cls._service_classes[name] = (module, service_class)
+        else:
+            if module is not None:
+                raise KeyError(
+                    "Must not set module if service_class is passed by value."
+                )
+            cls._service_classes[name] = service_class
+
+        # For backwards compatibility with class attribute service types.
+        service_cls_name = "".join(word.title() for word in name.split("_")) + "Class"
+        setattr(cls, service_cls_name, cls.get_class(name))
+
+    @classmethod
+    def reset(cls) -> None:
+        """Reset the registered services."""
+        cls._service_classes.clear()
+        for name, class_name in _DEFAULT_SERVICES.items():
+            module_name = name.replace("_", "")
+            cls.register(
+                name, class_name, module=f"craft_application.services.{module_name}"
+            )
 
     def set_kwargs(
         self,
@@ -94,25 +185,101 @@ def update_kwargs(
         """
         self._service_kwargs.setdefault(service, {}).update(kwargs)
 
-    def __getattr__(self, service: str) -> services.AppService:
-        """Instantiate a service class.
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["config", "ConfigService", "ConfigClass"]
+    ) -> type[services.ConfigService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["fetch", "FetchService", "FetchClass"]
+    ) -> type[services.FetchService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["init", "InitService", "InitClass"]
+    ) -> type[services.InitService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["lifecycle", "LifecycleService", "LifecycleClass"]
+    ) -> type[services.LifecycleService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["package", "PackageService", "PackageClass"]
+    ) -> type[services.PackageService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["provider", "ProviderService", "ProviderClass"]
+    ) -> type[services.ProviderService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["remote_build", "RemoteBuildService", "RemoteBuildClass"]
+    ) -> type[services.RemoteBuildService]: ...
+    @overload
+    @classmethod
+    def get_class(
+        cls, name: Literal["request", "RequestService", "RequestClass"]
+    ) -> type[services.RequestService]: ...
+    @overload
+    @classmethod
+    def get_class(cls, name: str) -> type[services.AppService]: ...
+    @classmethod
+    def get_class(cls, name: str) -> type[services.AppService]:
+        """Get the class for a service by its name."""
+        if name.endswith("Class"):
+            service_cls_name = name
+            service = _CAMEL_TO_PYTHON_CASE_REGEX.sub("_", name[:-5]).lower()
+        elif name.endswith("Service"):
+            service = _CAMEL_TO_PYTHON_CASE_REGEX.sub("_", name[:-7]).lower()
+            service_cls_name = name[:-7] + "Class"
+        else:
+            service_cls_name = "".join(word.title() for word in name.split("_"))
+            service_cls_name += "Class"
+            service = name
+        if service not in cls._service_classes:
+            raise AttributeError(f"Not a registered service: {service}")
+        service_info = cls._service_classes[service]
+        if isinstance(service_info, tuple):
+            module_name, class_name = service_info
+            module = importlib.import_module(module_name)
+            return cast(type[services.AppService], getattr(module, class_name))
+        return service_info
 
-        This allows us to lazy-load only the necessary services whilst still
-        treating them as attributes of our factory in a dynamic manner.
-        For a service (e.g. ``package``, the PackageService instance) that has not
-        been instantiated, this method finds the corresponding class, instantiates
-        it with defaults and any values set using ``set_kwargs``, and stores the
-        instantiated service as an instance attribute, allowing the same service
-        instance to be reused for the entire run of the application.
+    @overload
+    def get(self, service: Literal["config"]) -> services.ConfigService: ...
+    @overload
+    def get(self, service: Literal["fetch"]) -> services.FetchService: ...
+    @overload
+    def get(self, service: Literal["init"]) -> services.InitService: ...
+    @overload
+    def get(self, service: Literal["package"]) -> services.PackageService: ...
+    @overload
+    def get(self, service: Literal["lifecycle"]) -> services.LifecycleService: ...
+    @overload
+    def get(self, service: Literal["provider"]) -> services.ProviderService: ...
+    @overload
+    def get(self, service: Literal["remote_build"]) -> services.RemoteBuildService: ...
+    @overload
+    def get(self, service: Literal["request"]) -> services.RequestService: ...
+    @overload
+    def get(self, service: str) -> services.AppService: ...
+    def get(self, service: str) -> services.AppService:
+        """Get a service by name.
+
+        :param service: the name of the service (e.g. "config")
+        :returns: An instantiated and set up service class.
+
+        Also caches the service so as to provide a single service instance per
+        ServiceFactory.
         """
-        service_cls_name = "".join(word.title() for word in service.split("_"))
-        service_cls_name += "Class"
-        classes = dataclasses.asdict(self)
-        if service_cls_name not in classes:
-            raise AttributeError(service)
-        cls = getattr(self, service_cls_name)
-        if not issubclass(cls, services.AppService):
-            raise TypeError(f"{cls.__name__} is not a service class")
+        if service in self._services:
+            return self._services[service]
+        cls = self.get_class(service)
         kwargs = self._service_kwargs.get(service, {})
         if issubclass(cls, services.ProjectService):
             if not self.project:
@@ -121,7 +288,25 @@ def __getattr__(self, service: str) -> services.AppService:
                 )
             kwargs.setdefault("project", self.project)
 
-        instance: services.AppService = cls(app=self.app, services=self, **kwargs)
+        instance = cls(app=self.app, services=self, **kwargs)
         instance.setup()
-        setattr(self, service, instance)
+        self._services[service] = instance
         return instance
+
+    def __getattr__(self, name: str) -> services.AppService | type[services.AppService]:
+        """Instantiate a service class.
+
+        This allows us to lazy-load only the necessary services whilst still
+        treating them as attributes of our factory in a dynamic manner.
+        For a service (e.g. ``package``, the PackageService instance) that has not
+        been instantiated, this method finds the corresponding class, instantiates
+        it with defaults and any values set using ``set_kwargs``, and stores the
+        instantiated service as an instance attribute, allowing the same service
+        instance to be reused for the entire run of the application.
+        """
+        result = self.get_class(name) if name.endswith("Class") else self.get(name)
+        setattr(self, name, result)
+        return result
+
+
+ServiceFactory.reset()  # Set up default services on import.

pyproject.toml

@@ -3,6 +3,7 @@ name = "craft-application"
 description = "A framework for *craft applications."
 dynamic = ["version", "readme"]
 dependencies = [
+    "annotated-types>=0.6.0",
     "craft-archives>=2.0.0",
     "craft-cli>=2.10.1",
     "craft-grammar>=2.0.0",

tests/conftest.py

@@ -30,6 +30,7 @@
 import pydantic
 import pytest
 from craft_application import application, git, launchpad, models, services, util
+from craft_application.services import service_factory
 from craft_cli import EmitterMode, emit
 from craft_providers import bases
 from jinja2 import FileSystemLoader
@@ -46,6 +47,12 @@ def _create_fake_build_plan(num_infos: int = 1) -> list[models.BuildInfo]:
     return [models.BuildInfo("foo", arch, arch, base)] * num_infos
 
 
+@pytest.fixture(autouse=True)
+def reset_services():
+    yield
+    service_factory.ServiceFactory.reset()
+
+
 @pytest.fixture
 def features(request) -> dict[str, bool]:
     """Fixture that controls the enabled features.
@@ -310,19 +317,21 @@ def _get_loader(self, template_dir: pathlib.Path) -> jinja2.BaseLoader:
 
 @pytest.fixture
 def fake_services(
+    tmp_path,
     app_metadata,
     fake_project,
     fake_lifecycle_service_class,
     fake_package_service_class,
     fake_init_service_class,
 ):
-    return services.ServiceFactory(
-        app_metadata,
-        project=fake_project,
-        PackageClass=fake_package_service_class,
-        LifecycleClass=fake_lifecycle_service_class,
-        InitClass=fake_init_service_class,
+    services.ServiceFactory.register("package", fake_package_service_class)
+    services.ServiceFactory.register("lifecycle", fake_lifecycle_service_class)
+    services.ServiceFactory.register("init", fake_init_service_class)
+    factory = services.ServiceFactory(app_metadata, project=fake_project)
+    factory.update_kwargs(
+        "lifecycle", work_dir=tmp_path, cache_dir=tmp_path / "cache", build_plan=[]
     )
+    return factory
 
 
 class FakeApplication(application.Application):