from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional, cast
from pydantic import BaseModel, Field
from weaviate.backup.backup_location import _BackupLocationConfig
STORAGE_NAMES = {
"filesystem",
"s3",
"gcs",
"azure",
}
[docs]
class BackupCompressionLevel(str, Enum):
"""Which compression level should be used to compress the backup."""
DEFAULT = "DefaultCompression"
BEST_SPEED = "BestSpeed"
BEST_COMPRESSION = "BestCompression"
ZSTD_BEST_SPEED = "ZstdBestSpeed"
ZSTD_DEFAULT = "ZstdDefaultCompression"
ZSTD_BEST_COMPRESSION = "ZstdBestCompression"
NO_COMPRESSION = "NoCompression"
[docs]
class BackupStorage(str, Enum):
"""Which backend should be used to write the backup to."""
FILESYSTEM = "filesystem"
S3 = "s3"
GCS = "gcs"
AZURE = "azure"
[docs]
class BackupStatus(str, Enum):
"""The status of a backup."""
STARTED = "STARTED"
TRANSFERRING = "TRANSFERRING"
TRANSFERRED = "TRANSFERRED"
CANCELLING = "CANCELLING"
FINALIZING = "FINALIZING"
SUCCESS = "SUCCESS"
FAILED = "FAILED"
CANCELED = "CANCELED"
[docs]
class _BackupConfigBase(BaseModel):
CPUPercentage: Optional[int] = Field(default=None, alias="cpu_percentage")
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret = cast(dict, self.model_dump(exclude_none=True))
for key, val in ret.items():
if isinstance(val, _BackupLocationConfig):
ret[key] = val._to_dict()
return ret
[docs]
class BackupConfigCreate(_BackupConfigBase):
"""Options to configure the backup when creating a backup."""
ChunkSize: Optional[int] = Field(
default=None,
alias="chunk_size",
description="DEPRECATED: This parameter no longer has any effect.",
exclude=True,
)
CompressionLevel: Optional[BackupCompressionLevel] = Field(
default=None, alias="compression_level"
)
[docs]
class BackupConfigRestore(_BackupConfigBase):
"""Options to configure the backup when restoring a backup."""
[docs]
class BackupStatusReturn(BaseModel):
"""Return type of the backup status methods."""
error: Optional[str] = Field(default=None)
status: BackupStatus
path: str
backup_id: str = Field(alias="id")
size: float = Field(default=0)
[docs]
class BackupReturn(BackupStatusReturn):
"""Return type of the backup creation and restore methods."""
collections: List[str] = Field(default_factory=list, alias="classes")
[docs]
class BackupListReturn(BaseModel):
"""Return type of the backup list method."""
collections: List[str] = Field(default_factory=list, alias="classes")
status: BackupStatus
backup_id: str = Field(alias="id")
started_at: Optional[datetime] = Field(alias="startedAt", default=None)
completed_at: Optional[datetime] = Field(alias="completedAt", default=None)
size: float = Field(default=0)