✨ feat(services): add support for service manager and service schema

- Added `__init__.py` file to the `services` directory to make it a package.
- Created `Service` class in `base.py` to serve as a base class for different services.
- Implemented `ServiceFactory` class in `factory.py` to create instances of services.
- Implemented `ServiceManager` class in `manager.py` to manage the creation and retrieval of services.
- Created `ServiceType` enum in `schema.py` to define the different types of services that can be registered with the service manager.
- Added `initialize_services` function in `manager.py` to initialize all the services needed.

The purpose of these changes is to provide a modular and extensible architecture for managing different services in the application. The `ServiceManager` allows for easy creation and retrieval of services, while the `ServiceType` enum provides a standardized way to refer to different types of services. The `Service` base class and `ServiceFactory` class provide a foundation for creating and managing specific services.

src/backend/langflow/services/__init__.py

@@ -0,0 +1,4 @@
+from .manager import service_manager
+from .schema import ServiceType
+
+__all__ = ["service_manager", "ServiceType"]

src/backend/langflow/services/base.py

@@ -0,0 +1,2 @@
+class Service:
+    name: str

src/backend/langflow/services/factory.py

@@ -0,0 +1,6 @@
+class ServiceFactory:
+    def __init__(self, service_class):
+        self.service_class = service_class
+
+    def create(self, *args, **kwargs):
+        raise NotImplementedError

src/backend/langflow/services/manager.py

@@ -0,0 +1,87 @@
+from langflow.services.schema import ServiceType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from langflow.services.factory import ServiceFactory
+
+
+class ServiceManager:
+    """
+    Manages the creation of different services.
+    """
+
+    def __init__(self):
+        self.services = {}
+        self.factories = {}
+
+    def register_factory(self, service_factory: "ServiceFactory"):
+        """
+        Registers a new factory.
+        """
+        self.factories[service_factory.service_class.name] = service_factory
+
+    def get(self, service_name: ServiceType):
+        """
+        Get (or create) a service by its name.
+        """
+        if service_name not in self.services:
+            self._create_service(service_name)
+
+        return self.services[service_name]
+
+    def _create_service(self, service_name: ServiceType):
+        """
+        Create a new service given its name.
+        """
+        self._validate_service_creation(service_name)
+
+        if service_name == ServiceType.SETTINGS_MANAGER:
+            self.services[service_name] = self.factories[service_name].create()
+        else:
+            settings_service = self.get(ServiceType.SETTINGS_MANAGER)
+            self.services[service_name] = self.factories[service_name].create(
+                settings_service
+            )
+
+    def _validate_service_creation(self, service_name: ServiceType):
+        """
+        Validate whether the service can be created.
+        """
+        if service_name not in self.factories:
+            raise ValueError(
+                f"No factory registered for the service class '{service_name.name}'"
+            )
+
+        if (
+            ServiceType.SETTINGS_MANAGER not in self.factories
+            and service_name != ServiceType.SETTINGS_MANAGER
+        ):
+            raise ValueError(
+                f"Cannot create service '{service_name.name}' before the settings service"
+            )
+
+    def update(self, service_name: ServiceType):
+        """
+        Update a service by its name.
+        """
+        if service_name in self.services:
+            self.services.pop(service_name, None)
+            self.get(service_name)
+
+
+service_manager = ServiceManager()
+
+
+def initialize_services():
+    """
+    Initialize all the services needed.
+    """
+    from langflow.services.database import factory as database_factory
+    from langflow.services.cache import factory as cache_factory
+    from langflow.services.chat import factory as chat_factory
+    from langflow.services.settings import factory as settings_factory
+
+    service_manager.register_factory(settings_factory.SettingsManagerFactory())
+    service_manager.register_factory(database_factory.DatabaseManagerFactory())
+    service_manager.register_factory(cache_factory.CacheManagerFactory())
+    service_manager.register_factory(chat_factory.ChatManagerFactory())

src/backend/langflow/services/schema.py

@@ -0,0 +1,13 @@
+from enum import Enum
+
+
+class ServiceType(str, Enum):
+    """
+    Enum for the different types of services that can be
+    registered with the service manager.
+    """
+
+    CACHE_MANAGER = "cache_manager"
+    SETTINGS_MANAGER = "settings_manager"
+    DATABASE_MANAGER = "database_manager"
+    CHAT_MANAGER = "chat_manager"