diff --git a/api/extensions/storage/clickzetta_volume/file_lifecycle.py b/api/extensions/storage/clickzetta_volume/file_lifecycle.py index 483bd6bbf6..86b1bba544 100644 --- a/api/extensions/storage/clickzetta_volume/file_lifecycle.py +++ b/api/extensions/storage/clickzetta_volume/file_lifecycle.py @@ -13,7 +13,7 @@ import operator from dataclasses import asdict, dataclass from datetime import datetime from enum import StrEnum, auto -from typing import Any +from typing import Any, TypedDict from pydantic import TypeAdapter @@ -22,6 +22,17 @@ logger = logging.getLogger(__name__) _metadata_adapter: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any]) +class StorageStatisticsDict(TypedDict): + total_files: int + active_files: int + archived_files: int + deleted_files: int + total_size: int + versions_count: int + oldest_file: str | None + newest_file: str | None + + class FileStatus(StrEnum): """File status enumeration""" @@ -384,7 +395,7 @@ class FileLifecycleManager: logger.exception("Failed to cleanup old versions") return 0 - def get_storage_statistics(self) -> dict[str, Any]: + def get_storage_statistics(self) -> StorageStatisticsDict: """Get storage statistics Returns: @@ -393,16 +404,16 @@ class FileLifecycleManager: try: metadata_dict = self._load_metadata() - stats: dict[str, Any] = { - "total_files": len(metadata_dict), - "active_files": 0, - "archived_files": 0, - "deleted_files": 0, - "total_size": 0, - "versions_count": 0, - "oldest_file": None, - "newest_file": None, - } + stats = StorageStatisticsDict( + total_files=len(metadata_dict), + active_files=0, + archived_files=0, + deleted_files=0, + total_size=0, + versions_count=0, + oldest_file=None, + newest_file=None, + ) oldest_date = None newest_date = None @@ -437,7 +448,16 @@ class FileLifecycleManager: except Exception: logger.exception("Failed to get storage statistics") - return {} + return StorageStatisticsDict( + total_files=0, + active_files=0, + archived_files=0, + deleted_files=0, + total_size=0, + versions_count=0, + oldest_file=None, + newest_file=None, + ) def _create_version_backup(self, filename: str, metadata: dict): """Create version backup"""