From 78d0c122d370cadf1357f3619a12b2ea3f3928ac Mon Sep 17 00:00:00 2001 From: Gabriel Luiz Freitas Almeida Date: Tue, 5 Mar 2024 11:56:34 -0300 Subject: [PATCH] Update settings and task service --- .../langflow/services/settings/base.py | 37 ++++++++++++----- src/backend/langflow/services/task/service.py | 40 +++++++++++-------- 2 files changed, 51 insertions(+), 26 deletions(-) diff --git a/src/backend/langflow/services/settings/base.py b/src/backend/langflow/services/settings/base.py index 93f8b3085..3a1ec956e 100644 --- a/src/backend/langflow/services/settings/base.py +++ b/src/backend/langflow/services/settings/base.py @@ -58,13 +58,17 @@ class Settings(BaseSettings): STORE: Optional[bool] = True STORE_URL: Optional[str] = "https://api.langflow.store" - DOWNLOAD_WEBHOOK_URL: Optional[ - str - ] = "https://api.langflow.store/flows/trigger/ec611a61-8460-4438-b187-a4f65e5559d4" - LIKE_WEBHOOK_URL: Optional[str] = "https://api.langflow.store/flows/trigger/64275852-ec00-45c1-984e-3bff814732da" + DOWNLOAD_WEBHOOK_URL: Optional[str] = ( + "https://api.langflow.store/flows/trigger/ec611a61-8460-4438-b187-a4f65e5559d4" + ) + LIKE_WEBHOOK_URL: Optional[str] = ( + "https://api.langflow.store/flows/trigger/64275852-ec00-45c1-984e-3bff814732da" + ) STORAGE_TYPE: str = "local" + CELERY_ENABLED: bool = False + @validator("CONFIG_DIR", pre=True, allow_reuse=True) def set_langflow_dir(cls, value): if not value: @@ -91,7 +95,9 @@ class Settings(BaseSettings): @validator("DATABASE_URL", pre=True) def set_database_url(cls, value, values): if not value: - logger.debug("No database_url provided, trying LANGFLOW_DATABASE_URL env variable") + logger.debug( + "No database_url provided, trying LANGFLOW_DATABASE_URL env variable" + ) if langflow_database_url := os.getenv("LANGFLOW_DATABASE_URL"): value = langflow_database_url logger.debug("Using LANGFLOW_DATABASE_URL env variable.") @@ -101,7 +107,9 @@ class Settings(BaseSettings): # so we need to migrate to the new format # if there is a database in that location if not values["CONFIG_DIR"]: - raise ValueError("CONFIG_DIR not set, please set it or provide a DATABASE_URL") + raise ValueError( + "CONFIG_DIR not set, please set it or provide a DATABASE_URL" + ) new_path = f"{values['CONFIG_DIR']}/langflow.db" if Path("./langflow.db").exists(): @@ -125,15 +133,22 @@ class Settings(BaseSettings): if os.getenv("LANGFLOW_COMPONENTS_PATH"): logger.debug("Adding LANGFLOW_COMPONENTS_PATH to components_path") langflow_component_path = os.getenv("LANGFLOW_COMPONENTS_PATH") - if Path(langflow_component_path).exists() and langflow_component_path not in value: + if ( + Path(langflow_component_path).exists() + and langflow_component_path not in value + ): if isinstance(langflow_component_path, list): for path in langflow_component_path: if path not in value: value.append(path) - logger.debug(f"Extending {langflow_component_path} to components_path") + logger.debug( + f"Extending {langflow_component_path} to components_path" + ) elif langflow_component_path not in value: value.append(langflow_component_path) - logger.debug(f"Appending {langflow_component_path} to components_path") + logger.debug( + f"Appending {langflow_component_path} to components_path" + ) if not value: value = [BASE_COMPONENTS_PATH] @@ -145,7 +160,9 @@ class Settings(BaseSettings): logger.debug(f"Components path: {value}") return value - model_config = SettingsConfigDict(validate_assignment=True, extra="ignore", env_prefix="LANGFLOW_") + model_config = SettingsConfigDict( + validate_assignment=True, extra="ignore", env_prefix="LANGFLOW_" + ) # @model_validator() # @classmethod diff --git a/src/backend/langflow/services/task/service.py b/src/backend/langflow/services/task/service.py index 3f7a81f2c..e7f87d3f9 100644 --- a/src/backend/langflow/services/task/service.py +++ b/src/backend/langflow/services/task/service.py @@ -1,11 +1,14 @@ -from typing import Any, Callable, Coroutine, Union +from typing import TYPE_CHECKING, Any, Callable, Coroutine, Union + +from loguru import logger from langflow.services.base import Service from langflow.services.task.backends.anyio import AnyIOBackend from langflow.services.task.backends.base import TaskBackend from langflow.services.task.utils import get_celery_worker_status -from langflow.utils.logger import configure -from loguru import logger + +if TYPE_CHECKING: + from langflow.services.settings.service import SettingsService def check_celery_availability(): @@ -20,28 +23,31 @@ def check_celery_availability(): return status -try: - configure() - status = check_celery_availability() - - USE_CELERY = status.get("availability") is not None -except ImportError: - USE_CELERY = False - - class TaskService(Service): name = "task_service" - def __init__(self): - self.backend = self.get_backend() + def __init__(self, settings_service: "SettingsService"): + self.settings_service = settings_service + try: + if self.settings_service.settings.CELERY_ENABLED: + USE_CELERY = True + status = check_celery_availability() + + USE_CELERY = status.get("availability") is not None + else: + USE_CELERY = False + except ImportError: + USE_CELERY = False + self.use_celery = USE_CELERY + self.backend = self.get_backend() @property def backend_name(self) -> str: return self.backend.name def get_backend(self) -> TaskBackend: - if USE_CELERY: + if self.use_celery: from langflow.services.task.backends.celery import CeleryBackend logger.debug("Using Celery backend") @@ -68,7 +74,9 @@ class TaskService(Service): result = await result return task.id, result - async def launch_task(self, task_func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + async def launch_task( + self, task_func: Callable[..., Any], *args: Any, **kwargs: Any + ) -> Any: logger.debug(f"Launching task {task_func} with args {args} and kwargs {kwargs}") logger.debug(f"Using backend {self.backend}") task = self.backend.launch_task(task_func, *args, **kwargs)