You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
130 lines
4.0 KiB
130 lines
4.0 KiB
from __future__ import annotations
|
|
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field, ValidationError as PydanticValidationError, field_validator
|
|
|
|
from app.core.exceptions import ValidationError
|
|
|
|
BackendType = Literal["local_directory", "s3", "webdav"]
|
|
|
|
|
|
class BackendConfigError(ValidationError):
|
|
code = "backend_config_invalid"
|
|
message = "Backend configuration is invalid."
|
|
|
|
|
|
class LocalDirectoryConfig(BaseModel):
|
|
base_path: str = Field(min_length=1)
|
|
|
|
@field_validator("base_path")
|
|
@classmethod
|
|
def clean_base_path(cls, value: str) -> str:
|
|
cleaned = value.strip()
|
|
if not cleaned:
|
|
raise ValueError("base_path must not be empty")
|
|
return cleaned
|
|
|
|
|
|
class S3CompatibleConfig(BaseModel):
|
|
bucket: str = Field(min_length=1)
|
|
region: str | None = None
|
|
endpoint_url: str | None = None
|
|
access_key_id: str | None = None
|
|
secret_access_key: str | None = None
|
|
prefix: str = ""
|
|
force_path_style: bool = False
|
|
verify_ssl: bool = True
|
|
storage_class: str | None = None
|
|
server_side_encryption: str | None = None
|
|
|
|
@field_validator("bucket")
|
|
@classmethod
|
|
def clean_bucket(cls, value: str) -> str:
|
|
cleaned = value.strip()
|
|
if not cleaned:
|
|
raise ValueError("bucket must not be empty")
|
|
return cleaned
|
|
|
|
@field_validator("prefix")
|
|
@classmethod
|
|
def clean_prefix(cls, value: str) -> str:
|
|
return value.strip().strip("/")
|
|
|
|
|
|
class WebDAVConfig(BaseModel):
|
|
endpoint_url: str = Field(min_length=1)
|
|
username: str = Field(min_length=1)
|
|
password: str | None = None
|
|
root_path: str = Field(default="iron", min_length=1)
|
|
verify_ssl: bool = True
|
|
|
|
@field_validator("endpoint_url", "username")
|
|
@classmethod
|
|
def clean_required_text(cls, value: str) -> str:
|
|
cleaned = value.strip()
|
|
if not cleaned:
|
|
raise ValueError("value must not be empty")
|
|
return cleaned
|
|
|
|
@field_validator("endpoint_url")
|
|
@classmethod
|
|
def clean_endpoint_url(cls, value: str) -> str:
|
|
return value.strip().rstrip("/")
|
|
|
|
@field_validator("root_path")
|
|
@classmethod
|
|
def clean_root_path(cls, value: str) -> str:
|
|
cleaned = "/".join(part for part in value.strip().strip("/").split("/") if part)
|
|
if not cleaned:
|
|
raise ValueError("root_path must not be empty")
|
|
return cleaned
|
|
|
|
|
|
CONFIG_MODELS: dict[str, type[BaseModel]] = {
|
|
"local_directory": LocalDirectoryConfig,
|
|
"s3": S3CompatibleConfig,
|
|
"webdav": WebDAVConfig,
|
|
}
|
|
|
|
SECRET_CONFIG_KEYS = {
|
|
"access_key_id",
|
|
"secret_access_key",
|
|
"password",
|
|
}
|
|
|
|
|
|
def supported_backend_types() -> set[str]:
|
|
return set(CONFIG_MODELS)
|
|
|
|
|
|
def validate_backend_config(backend_type: str, config: dict[str, Any]) -> dict[str, Any]:
|
|
model = CONFIG_MODELS.get(backend_type)
|
|
if model is None:
|
|
raise BackendConfigError(f"Unsupported backend type: {backend_type}.")
|
|
|
|
try:
|
|
return model.model_validate(config).model_dump()
|
|
except PydanticValidationError as exc:
|
|
first_error = exc.errors()[0] if exc.errors() else {}
|
|
location = ".".join(str(part) for part in first_error.get("loc", []))
|
|
message = first_error.get("msg", "Invalid backend configuration.")
|
|
prefix = f"{location}: " if location else ""
|
|
raise BackendConfigError(prefix + message) from exc
|
|
|
|
|
|
def split_backend_config(backend_type: str, config: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
normalized = validate_backend_config(backend_type, config)
|
|
public_config: dict[str, Any] = {}
|
|
secret_config: dict[str, Any] = {}
|
|
for key, value in normalized.items():
|
|
if key in SECRET_CONFIG_KEYS:
|
|
if value:
|
|
secret_config[key] = value
|
|
else:
|
|
public_config[key] = value
|
|
return public_config, secret_config
|
|
|
|
|
|
def merge_backend_config(public_config: dict[str, Any], secret_config: dict[str, Any]) -> dict[str, Any]:
|
|
return {**public_config, **secret_config}
|
|
|