"""Unified database backend configuration. Controls BOTH the LangGraph checkpointer and the DeerFlow application persistence layer (runs, threads metadata, users, etc.). The user configures one backend; the system handles physical separation details. SQLite mode: checkpointer and app share a single .db file ({sqlite_dir}/deerflow.db) with WAL journal mode enabled on every connection. WAL allows concurrent readers and a single writer without blocking, making a unified file safe for both workloads. Writers that contend for the lock wait via the default 5-second sqlite3 busy timeout rather than failing immediately. Postgres mode: both use the same database URL but maintain independent connection pools with different lifecycles. Gauss mode: application ORM data uses the official openGauss SQLAlchemy dialect over asyncpg. LangGraph's PostgreSQL checkpointer is intentionally not reused for GaussDB; configure the legacy ``checkpointer:`` section separately (normally sqlite). MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does not have a built-in MySQL backend in this project, so deployments that choose MySQL for business tables should configure the legacy ``checkpointer:`` section separately (usually sqlite) to preserve conversation checkpoints. Memory mode: checkpointer uses MemorySaver, app uses in-memory stores. No database is initialized. Sensitive values (postgres_url) should use $VAR syntax in config.yaml to reference environment variables from .env: database: backend: postgres postgres_url: $DATABASE_URL The $VAR resolution is handled by AppConfig.resolve_env_variables() before this config is instantiated -- DatabaseConfig itself does not need to do any environment variable processing. """ from __future__ import annotations import os from typing import Literal from pydantic import BaseModel, Field class DatabaseConfig(BaseModel): backend: Literal["memory", "sqlite", "postgres", "gauss", "mysql"] = Field( default="memory", description=( "Storage backend for application data. 'memory' for development " "(no persistence across restarts), 'sqlite' for single-node " "deployment, 'postgres' for production multi-node deployment, " "'gauss' for openGauss/GaussDB business tables, and 'mysql' for " "MySQL-backed business tables." ), ) sqlite_dir: str = Field( default=".deer-flow/data", description=("Directory for the SQLite database file. Both checkpointer and application data share {sqlite_dir}/deerflow.db."), ) postgres_url: str = Field( default="", description=( "PostgreSQL connection URL, shared by checkpointer and app. " "Use $DATABASE_URL in config.yaml to reference .env. " "Example: postgresql://user:pass@host:5432/deerflow " "(the +asyncpg driver suffix is added automatically where needed)." ), ) gauss_url: str = Field( default="", description=( "openGauss/GaussDB connection URL for the application ORM. " "Use $GAUSS_DATABASE_URL in config.yaml to reference .env. " "Example: opengauss+asyncpg://user:pass@host:26000/deerflow. " "Plain opengauss:// or PostgreSQL-style URLs are normalized to the " "official asynchronous openGauss SQLAlchemy dialect." ), ) mysql_url: str = Field( default="", description=( "MySQL connection URL for the application ORM. " "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. " "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. " "If mysql:// is provided, the +asyncmy driver suffix is added automatically." ), ) echo_sql: bool = Field( default=False, description="Echo all SQL statements to log (debug only).", ) pool_size: int = Field( default=5, description="Connection pool size for the app ORM engine (postgres/gauss/mysql only).", ) max_overflow: int = Field( default=10, description=( "Extra connections allowed beyond pool_size under load (postgres/gauss/mysql only). " "The real connection ceiling is pool_size + max_overflow. SQLAlchemy default is 10." ), ) pool_timeout: int = Field( default=30, description=( "Seconds a caller waits for a free pooled connection before erroring (postgres/gauss/mysql only). " "SQLAlchemy default is 30." ), ) pool_recycle: int = Field( default=1800, description=( "Recycle pooled connections after this many seconds (postgres/gauss/mysql only). " "Prevents stale-connection errors when a firewall, pgbouncer, or MySQL " "wait_timeout silently drops idle connections. pool_pre_ping catches most " "cases, but pool_recycle is the belt-and-suspenders safety net. " "Multi-worker sizing: each worker opens up to (pool_size + max_overflow) " "connections; total across workers must stay below the DB's max_connections." ), ) # -- Derived helpers (not user-configured) -- @property def _resolved_sqlite_dir(self) -> str: """Resolve sqlite_dir to an absolute path (relative to CWD).""" from pathlib import Path return str(Path(self.sqlite_dir).resolve()) @property def sqlite_path(self) -> str: """Unified SQLite file path shared by checkpointer and app.""" return os.path.join(self._resolved_sqlite_dir, "deerflow.db") # Backward-compatible aliases @property def checkpointer_sqlite_path(self) -> str: """SQLite file path for the LangGraph checkpointer (alias for sqlite_path).""" return self.sqlite_path @property def app_sqlite_path(self) -> str: """SQLite file path for application ORM data (alias for sqlite_path).""" return self.sqlite_path @property def app_sqlalchemy_url(self) -> str: """SQLAlchemy async URL for the application ORM engine.""" if self.backend == "sqlite": return f"sqlite+aiosqlite:///{self.sqlite_path}" if self.backend == "postgres": url = self.postgres_url if url.startswith("postgresql://"): url = url.replace("postgresql://", "postgresql+asyncpg://", 1) return url if self.backend == "gauss": url = self.gauss_url.strip() if not url: raise ValueError("database.gauss_url is required for backend='gauss'") # GaussDB speaks the PostgreSQL wire protocol, but its server # version string and DDL capabilities are not identical to # PostgreSQL. Always select the official openGauss dialect instead # of silently sending the URL through SQLAlchemy's generic PG # dialect (the latter commonly connects and then fails while # SQLAlchemy initializes/creates the schema). for prefix in ( "opengauss://", "gaussdb://", "postgres://", "postgresql://", "postgresql+asyncpg://", ): if url.startswith(prefix): return "opengauss+asyncpg://" + url[len(prefix) :] if url.startswith("opengauss+asyncpg://"): return url raise ValueError( "database.gauss_url must use opengauss://, " "opengauss+asyncpg://, or a PostgreSQL-style URL" ) if self.backend == "mysql": url = self.mysql_url if not url: raise ValueError("database.mysql_url is required for backend='mysql'") if url.startswith("mysql://"): url = url.replace("mysql://", "mysql+asyncmy://", 1) return url raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")