diff --git a/src/backend/langflow/services/settings/base.py b/src/backend/langflow/services/settings/base.py index fe1fe496f..cb004888a 100644 --- a/src/backend/langflow/services/settings/base.py +++ b/src/backend/langflow/services/settings/base.py @@ -1,6 +1,7 @@ import contextlib import json import os +from shutil import copy2 import secrets from typing import Optional, List from pathlib import Path @@ -30,6 +31,9 @@ class Settings(BaseSettings): OUTPUT_PARSERS: dict = {} CUSTOM_COMPONENTS: dict = {} + # Define the default LANGFLOW_DIR + CONFIG_DIR: Optional[str] = None + DEV: bool = False DATABASE_URL: Optional[str] = None CACHE: str = "InMemoryCache" @@ -53,8 +57,31 @@ class Settings(BaseSettings): # > The application does not request login and logs in automatically as a super user. AUTO_LOGIN: bool = True + @validator("CONFIG_DIR", pre=True, allow_reuse=True) + def set_langflow_dir(cls, value): + if not value: + import appdirs + + # Define the app name and author + app_name = "langflow" + app_author = "logspace" + + # Get the cache directory for the application + cache_dir = appdirs.user_cache_dir(app_name, app_author) + + # Create a .langflow directory inside the cache directory + value = Path(cache_dir) + value.mkdir(parents=True, exist_ok=True) + + if isinstance(value, str): + value = Path(value) + if not value.exists(): + value.mkdir(parents=True, exist_ok=True) + + return str(value) + @validator("DATABASE_URL", pre=True) - def set_database_url(cls, value): + def set_database_url(cls, value, values): if not value: logger.debug( "No database_url provided, trying LANGFLOW_DATABASE_URL env variable" @@ -64,7 +91,28 @@ class Settings(BaseSettings): logger.debug("Using LANGFLOW_DATABASE_URL env variable.") else: logger.debug("No DATABASE_URL env variable, using sqlite database") - value = "sqlite:///./langflow.db" + # Originally, we used sqlite:///./langflow.db + # 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" + ) + + new_path = f"{values['CONFIG_DIR']}/langflow.db" + if Path("./langflow.db").exists(): + if Path(new_path).exists(): + logger.debug(f"Database already exists at {new_path}, using it") + else: + try: + logger.debug("Copying existing database to new location") + copy2("./langflow.db", new_path) + logger.debug(f"Copied existing database to {new_path}") + except Exception: + logger.error("Failed to copy database, using default path") + new_path = "./langflow.db" + + value = f"sqlite:///{new_path}" return value