🔧 fix(manager.py): import List from typing module to fix type hinting error

 feat(manager.py): add support for registering factories with dependencies to handle service creation
🔧 fix(manager.py): fix service creation logic to handle dependencies and create services in the correct order
 feat(manager.py): add support for initializing session manager with dependencies
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-08-21 16:37:56 -03:00
commit 181620e538

View file

@ -1,5 +1,5 @@
from langflow.services.schema import ServiceType
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, List
if TYPE_CHECKING:
from langflow.services.factory import ServiceFactory
@ -13,13 +13,19 @@ class ServiceManager:
def __init__(self):
self.services = {}
self.factories = {}
self.dependencies = {}
def register_factory(self, service_factory: "ServiceFactory"):
def register_factory(
self, service_factory: "ServiceFactory", dependencies: List[ServiceType] = None
):
"""
Registers a new factory.
Registers a new factory with dependencies.
"""
if service_factory.service_class.name not in self.factories:
self.factories[service_factory.service_class.name] = service_factory
if dependencies is None:
dependencies = []
service_name = service_factory.service_class.name
self.factories[service_name] = service_factory
self.dependencies[service_name] = dependencies
def get(self, service_name: ServiceType):
"""
@ -32,17 +38,25 @@ class ServiceManager:
def _create_service(self, service_name: ServiceType):
"""
Create a new service given its name.
Create a new service given its name, handling dependencies.
"""
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
)
# Create dependencies first
for dependency in self.dependencies.get(service_name, []):
if dependency not in self.services:
self._create_service(dependency)
# Collect the dependent services
dependent_services = {
dep.value: self.services[dep]
for dep in self.dependencies.get(service_name, [])
}
# Create the actual service
self.services[service_name] = self.factories[service_name].create(
**dependent_services
)
def _validate_service_creation(self, service_name: ServiceType):
"""
@ -53,14 +67,6 @@ class ServiceManager:
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.
@ -81,11 +87,26 @@ def initialize_services():
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
from langflow.services.session import factory as session_manager_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(
database_factory.DatabaseManagerFactory(),
dependencies=[ServiceType.SETTINGS_MANAGER],
)
service_manager.register_factory(
cache_factory.CacheManagerFactory(), dependencies=[ServiceType.SETTINGS_MANAGER]
)
service_manager.register_factory(chat_factory.ChatManagerFactory())
service_manager.register_factory(
session_manager_factory.SessionManagerFactory(),
dependencies=[ServiceType.CACHE_MANAGER],
)
# Test cache connection
service_manager.get(ServiceType.CACHE_MANAGER)
# Test database connection
service_manager.get(ServiceType.DATABASE_MANAGER)
def initialize_settings_manager():
@ -95,3 +116,22 @@ def initialize_settings_manager():
from langflow.services.settings import factory as settings_factory
service_manager.register_factory(settings_factory.SettingsManagerFactory())
def initialize_session_manager():
"""
Initialize the session manager.
"""
from langflow.services.session import factory as session_manager_factory
from langflow.services.cache import factory as cache_factory
initialize_settings_manager()
service_manager.register_factory(
cache_factory.CacheManagerFactory(), dependencies=[ServiceType.SETTINGS_MANAGER]
)
service_manager.register_factory(
session_manager_factory.SessionManagerFactory(),
dependencies=[ServiceType.CACHE_MANAGER],
)