opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-09-08 11:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): type storage statistics with StorageStatisticsDict TypedDict (#34609)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
YBoy 2026-04-07 02:59:32 +02:00 committed by GitHub
parent 4b500f988d
commit 8d37116fec
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 33 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
api/extensions/storage/clickzetta_volume/file_lifecycle.py
View File

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