222 lines
9.2 KiB
Python
222 lines
9.2 KiB
Python
"""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, field_validator, model_validator
|
|
|
|
|
|
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). "
|
|
"When omitted, PostgreSQL defaults to 300s; other backends default to 1800s. "
|
|
"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."
|
|
),
|
|
)
|
|
command_timeout: float | None = Field(
|
|
default=30,
|
|
gt=0,
|
|
allow_inf_nan=False,
|
|
description=(
|
|
"asyncpg command timeout in seconds for the PostgreSQL/openGauss app ORM. "
|
|
"Independent of pool_timeout (connection checkout) and the psycopg checkpoint pool. "
|
|
"Null disables the command deadline. Not passed to MySQL or SQLite drivers."
|
|
),
|
|
)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def _postgres_pool_default(cls, values):
|
|
# Keep MySQL's existing 1800s default and all explicitly set values.
|
|
if isinstance(values, dict) and values.get("backend") == "postgres" and "pool_recycle" not in values:
|
|
return {**values, "pool_recycle": 300}
|
|
return values
|
|
|
|
@field_validator("command_timeout", mode="before")
|
|
@classmethod
|
|
def _reject_boolean_timeout(cls, value):
|
|
if isinstance(value, bool):
|
|
raise ValueError("command_timeout must be a positive finite number or null")
|
|
return value
|
|
|
|
# -- 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}")
|