deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py

"""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.

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", "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, '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)."
        ),
    )
    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/mysql only).",
    )
    max_overflow: int = Field(
        default=10,
        description=(
            "Extra connections allowed beyond pool_size under load (postgres/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/mysql only). "
            "SQLAlchemy default is 30."
        ),
    )

    # -- 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 == "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}")