jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

🔀 chore(utils.py): add type hints to imported modules for better code readability and maintainability

Browse source 
✨ feat(utils.py): add get_task_manager() function to retrieve the TaskManager instance from the service manager
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-09-01 17:22:01 -03:00
commit 6418d5e56c
1 changed files with 11 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/backend/langflow/services/utils.py
View file

@ -5,6 +5,10 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from langflow.services.database.manager import DatabaseManager from langflow.services.database.manager import DatabaseManager
from langflow.services.settings.manager import SettingsManager from langflow.services.settings.manager import SettingsManager
from langflow.services.cache.manager import BaseCacheManager
from langflow.services.session.manager import SessionManager
from langflow.services.task.manager import TaskManager
from sqlmodel import Session
def get_settings_manager() -> "SettingsManager": def get_settings_manager() -> "SettingsManager":
@ -15,14 +19,18 @@ def get_db_manager() -> "DatabaseManager":
return service_manager.get(ServiceType.DATABASE_MANAGER) return service_manager.get(ServiceType.DATABASE_MANAGER)
def get_session(): def get_session() -> "Session":
db_manager = service_manager.get(ServiceType.DATABASE_MANAGER) db_manager = service_manager.get(ServiceType.DATABASE_MANAGER)
yield from db_manager.get_session() yield from db_manager.get_session()
def get_cache_manager(): def get_cache_manager() -> "BaseCacheManager":
return service_manager.get(ServiceType.CACHE_MANAGER) return service_manager.get(ServiceType.CACHE_MANAGER)
def get_session_manager(): def get_session_manager() -> "SessionManager":
return service_manager.get(ServiceType.SESSION_MANAGER) return service_manager.get(ServiceType.SESSION_MANAGER)
def get_task_manager() -> "TaskManager":
return service_manager.get(ServiceType.TASK_MANAGER)