Update settings and task service

This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-03-05 11:56:34 -03:00
commit 78d0c122d3
2 changed files with 51 additions and 26 deletions

View file

@ -58,13 +58,17 @@ class Settings(BaseSettings):
STORE: Optional[bool] = True STORE: Optional[bool] = True
STORE_URL: Optional[str] = "https://api.langflow.store" STORE_URL: Optional[str] = "https://api.langflow.store"
DOWNLOAD_WEBHOOK_URL: Optional[ DOWNLOAD_WEBHOOK_URL: Optional[str] = (
str "https://api.langflow.store/flows/trigger/ec611a61-8460-4438-b187-a4f65e5559d4"
] = "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" LIKE_WEBHOOK_URL: Optional[str] = (
"https://api.langflow.store/flows/trigger/64275852-ec00-45c1-984e-3bff814732da"
)
STORAGE_TYPE: str = "local" STORAGE_TYPE: str = "local"
CELERY_ENABLED: bool = False
@validator("CONFIG_DIR", pre=True, allow_reuse=True) @validator("CONFIG_DIR", pre=True, allow_reuse=True)
def set_langflow_dir(cls, value): def set_langflow_dir(cls, value):
if not value: if not value:
@ -91,7 +95,9 @@ class Settings(BaseSettings):
@validator("DATABASE_URL", pre=True) @validator("DATABASE_URL", pre=True)
def set_database_url(cls, value, values): def set_database_url(cls, value, values):
if not value: 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"): if langflow_database_url := os.getenv("LANGFLOW_DATABASE_URL"):
value = langflow_database_url value = langflow_database_url
logger.debug("Using LANGFLOW_DATABASE_URL env variable.") logger.debug("Using LANGFLOW_DATABASE_URL env variable.")
@ -101,7 +107,9 @@ class Settings(BaseSettings):
# so we need to migrate to the new format # so we need to migrate to the new format
# if there is a database in that location # if there is a database in that location
if not values["CONFIG_DIR"]: 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" new_path = f"{values['CONFIG_DIR']}/langflow.db"
if Path("./langflow.db").exists(): if Path("./langflow.db").exists():
@ -125,15 +133,22 @@ class Settings(BaseSettings):
if os.getenv("LANGFLOW_COMPONENTS_PATH"): if os.getenv("LANGFLOW_COMPONENTS_PATH"):
logger.debug("Adding LANGFLOW_COMPONENTS_PATH to components_path") logger.debug("Adding LANGFLOW_COMPONENTS_PATH to components_path")
langflow_component_path = os.getenv("LANGFLOW_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): if isinstance(langflow_component_path, list):
for path in langflow_component_path: for path in langflow_component_path:
if path not in value: if path not in value:
value.append(path) 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: elif langflow_component_path not in value:
value.append(langflow_component_path) 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: if not value:
value = [BASE_COMPONENTS_PATH] value = [BASE_COMPONENTS_PATH]
@ -145,7 +160,9 @@ class Settings(BaseSettings):
logger.debug(f"Components path: {value}") logger.debug(f"Components path: {value}")
return 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() # @model_validator()
# @classmethod # @classmethod

View file

@ -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.base import Service
from langflow.services.task.backends.anyio import AnyIOBackend from langflow.services.task.backends.anyio import AnyIOBackend
from langflow.services.task.backends.base import TaskBackend from langflow.services.task.backends.base import TaskBackend
from langflow.services.task.utils import get_celery_worker_status 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(): def check_celery_availability():
@ -20,28 +23,31 @@ def check_celery_availability():
return status return status
try:
configure()
status = check_celery_availability()
USE_CELERY = status.get("availability") is not None
except ImportError:
USE_CELERY = False
class TaskService(Service): class TaskService(Service):
name = "task_service" name = "task_service"
def __init__(self): def __init__(self, settings_service: "SettingsService"):
self.backend = self.get_backend() 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.use_celery = USE_CELERY
self.backend = self.get_backend()
@property @property
def backend_name(self) -> str: def backend_name(self) -> str:
return self.backend.name return self.backend.name
def get_backend(self) -> TaskBackend: def get_backend(self) -> TaskBackend:
if USE_CELERY: if self.use_celery:
from langflow.services.task.backends.celery import CeleryBackend from langflow.services.task.backends.celery import CeleryBackend
logger.debug("Using Celery backend") logger.debug("Using Celery backend")
@ -68,7 +74,9 @@ class TaskService(Service):
result = await result result = await result
return task.id, 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"Launching task {task_func} with args {args} and kwargs {kwargs}")
logger.debug(f"Using backend {self.backend}") logger.debug(f"Using backend {self.backend}")
task = self.backend.launch_task(task_func, *args, **kwargs) task = self.backend.launch_task(task_func, *args, **kwargs)