strepsiades
71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
from typing import Any, Dict, Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
class PersistConfig(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", validate_assignment=True)
|
|
|
|
backend: str = "sqlite"
|
|
db_path: Optional[str] = None
|
|
url: Optional[str] = None
|
|
backend_options: Dict[str, Any] = Field(default_factory=dict)
|
|
enable: bool = True
|
|
wipe_on_start: bool = False
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def _validate_target(cls, data: Any) -> Any:
|
|
if not isinstance(data, dict):
|
|
return data
|
|
|
|
normalized = dict(data)
|
|
normalized["backend"] = (str(normalized.get("backend") or "sqlite").strip().lower() or "sqlite")
|
|
normalized["db_path"] = cls._normalize_optional_text(normalized.get("db_path"))
|
|
normalized["url"] = cls._normalize_optional_text(normalized.get("url"))
|
|
|
|
if not normalized["db_path"] and not normalized["url"]:
|
|
raise ValueError("persist requires either db_path or url")
|
|
if normalized["backend"] != "sqlite" and not normalized["url"]:
|
|
raise ValueError(f"persist.url is required for backend={normalized['backend']}")
|
|
return normalized
|
|
|
|
@staticmethod
|
|
def _normalize_optional_text(value: Optional[str]) -> Optional[str]:
|
|
if value is None:
|
|
return None
|
|
normalized = str(value).strip()
|
|
return normalized or None
|
|
|
|
def resolved_url(self) -> str:
|
|
if self.url:
|
|
return self.url
|
|
if self.backend == "sqlite" and self.db_path:
|
|
return f"sqlite+aiosqlite:///{self.db_path}"
|
|
raise ValueError(f"persist.url is required for backend={self.backend}")
|
|
|
|
def resolved_db_path(self) -> Optional[str]:
|
|
if self.db_path:
|
|
return self.db_path
|
|
if not self.url:
|
|
return None
|
|
|
|
from sqlalchemy.engine import make_url
|
|
|
|
parsed = make_url(self.url)
|
|
if parsed.get_backend_name() != "sqlite":
|
|
return None
|
|
database = parsed.database
|
|
if not database:
|
|
return None
|
|
return str(database)
|
|
|
|
def wipe_file_path(self) -> Optional[str]:
|
|
db_path = self.resolved_db_path()
|
|
if not db_path or db_path == ":memory:":
|
|
return None
|
|
return db_path
|
|
|
|
def display_target(self) -> str:
|
|
return self.url or self.db_path or "<unset>"
|