69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
"""Match the runtime's legacy-first persistence configuration precedence."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
def postgres_connection_string(config) -> str | None:
|
|
if not shared_postgres_configured(config):
|
|
return None
|
|
checkpoint = getattr(config, "checkpointer", None)
|
|
if checkpoint is not None:
|
|
return getattr(checkpoint, "connection_string", None)
|
|
return getattr(config.database, "postgres_url", None)
|
|
|
|
|
|
def shared_postgres_configured(config) -> bool:
|
|
database = getattr(config, "database", None)
|
|
if getattr(database, "backend", None) not in {"postgres", "mysql"}:
|
|
return False
|
|
checkpoint = getattr(config, "checkpointer", None)
|
|
if checkpoint is None:
|
|
return database.backend == "postgres"
|
|
raw = os.getenv("DEER_FLOW_USE_PG_CHECKPOINTER")
|
|
enabled = raw.strip().lower() not in {"0", "false", "no", "off"} if raw else getattr(checkpoint, "use_postgres", True)
|
|
return checkpoint.type == "postgres" and enabled
|
|
|
|
|
|
def sqlite_checkpoint_path(config) -> str | None:
|
|
from deerflow.runtime.checkpointer.async_provider import _sqlite_conn_str_for_postgres_fallback, _use_postgres_enabled
|
|
from deerflow.runtime.store._sqlite_utils import resolve_sqlite_conn_str
|
|
|
|
checkpoint = getattr(config, "checkpointer", None)
|
|
if checkpoint is None:
|
|
return None
|
|
if checkpoint.type == "sqlite":
|
|
raw = getattr(checkpoint, "connection_string", None) or "store.db"
|
|
elif checkpoint.type == "postgres" and not _use_postgres_enabled(checkpoint):
|
|
raw = _sqlite_conn_str_for_postgres_fallback(checkpoint)
|
|
else:
|
|
return None
|
|
return resolve_sqlite_conn_str(raw)
|
|
|
|
|
|
def shared_storage_configured(config) -> bool:
|
|
if shared_postgres_configured(config):
|
|
return True
|
|
if getattr(getattr(config, "database", None), "backend", None) not in {"mysql", "postgres"}:
|
|
return False
|
|
path = sqlite_checkpoint_path(config)
|
|
return bool(path and path != ":memory:" and not path.startswith("file:"))
|
|
|
|
|
|
def validate_sqlite_mount(config, mountinfo: str | None = None) -> None:
|
|
"""SQLite databases and WAL sidecars must live on one shared local directory."""
|
|
path = sqlite_checkpoint_path(config)
|
|
if path is None:
|
|
return
|
|
database = Path(path).resolve()
|
|
content = mountinfo if mountinfo is not None else Path("/proc/self/mountinfo").read_text()
|
|
for line in content.splitlines():
|
|
left, right = line.split(" - ", 1)
|
|
mount = Path(left.split()[4].replace("\\040", " "))
|
|
# A single-file bind does not share the -wal/-shm files.
|
|
if mount != Path("/") and mount in database.parents:
|
|
if right.split()[0] in {"nfs", "nfs4", "cifs", "smb3", "9p"}:
|
|
raise RuntimeError("SQLite Worker requires a shared local directory, not a network filesystem")
|
|
return
|
|
raise RuntimeError("SQLite Worker requires the database directory to be persistently mounted in every container")
|