file_manager module

class file_manager.BackupInfo(backup_id, user_id, created_at, file_count, total_size, backup_type, status, file_path, repo, path, metadata)

Bases: object

מידע על גיבוי

פרמטרים:

• backup_id (str)

• user_id (int)

• created_at (datetime)

• file_count (int)

• total_size (int)

• backup_type (str)

• status (str)

• file_path (str)

• repo (str | None)

• path (str | None)

• metadata (Dict[str, Any] | None)

__init__(backup_id, user_id, created_at, file_count, total_size, backup_type, status, file_path, repo, path, metadata)

פרמטרים:

• backup_id (str)

• user_id (int)

• created_at (datetime)

• file_count (int)

• total_size (int)

• backup_type (str)

• status (str)

• file_path (str)

• repo (str | None)

• path (str | None)

• metadata (Dict[str, Any] | None)

class file_manager.BackupManager

Bases: object

מנהל גיבויים

__init__()

cleanup_expired_backups(retention_days=None, *, max_per_user=None, budget_seconds=None)

ניקוי גיבויים ישנים ממערכת הקבצים ומ‑GridFS באופן מבוקר.

פרמטרים: - retention_days: ימים לשמירת גיבוי לפני מחיקה (ברירת מחדל: BACKUPS_RETENTION_DAYS או 30) - max_per_user: כמות מקסימלית של גיבויים לשמירה לכל משתמש (ברירת מחדל: BACKUPS_MAX_PER_USER או None) - budget_seconds: תקציב זמן לניקוי כדי לא לחסום את ה־worker (ברירת מחדל: BACKUPS_CLEANUP_BUDGET_SECONDS או 3)

החזרה: dict עם counters לסריקה/מחיקות ושגיאות.

Return type:

dict

פרמטרים:

• retention_days (int | None)

• max_per_user (int | None)

• budget_seconds (float | None)

save_backup_bytes(data, metadata)

שומר ZIP של גיבוי בהתאם למצב האחסון ומחזיר backup_id או None במקרה כשל.

אם storage==mongo: שומר ל-GridFS עם המטאדטה. אם storage==fs: שומר לקובץ תחת backup_dir.

Return type:

Optional[str]

פרמטרים:

• data (bytes)

• metadata (Dict[str, Any])

save_backup_file(file_path)

שומר קובץ ZIP קיים לאחסון היעד (Mongo/FS) ומחזיר backup_id אם הצליח.

Return type:

Optional[str]

פרמטרים:

file_path (str)

list_backups(user_id)

מחזירה רשימת קבצי ZIP ששייכים למשתמש המבקש בלבד.

כל פריט חייב להיות מסווג כשייך ל-user_id דרך אחד מהבאים: - metadata.json בתוך ה-ZIP עם שדה user_id תואם - דפוס מזהה בשם: backup_<user_id>_*

ZIPים ללא שיוך ברור למשתמש לא ייכללו כדי למנוע זליגת מידע.

Return type:

List[BackupInfo]

פרמטרים:

user_id (int)

restore_from_backup(user_id, backup_path, overwrite=True, purge=False, extra_tags=None)

משחזר קבצים מ-ZIP למסד הנתונים.

• purge=True: מסמן את כל הקבצים הקיימים של המשתמש כלא פעילים לפני השחזור

• overwrite=True: שמירה תמיד כגרסה חדשה עבור אותו שם (כברירת מחדל)

החזרה: dict עם restored_files ו-errors

Return type:

Dict[str, Any]

פרמטרים:

• user_id (int)

• backup_path (str)

• overwrite (bool)

• purge (bool)

• extra_tags (List[str] | None)

delete_backups(user_id, backup_ids)

מוחק מספר גיבויי ZIP לפי backup_id ממערכת הקבצים ומ-GridFS (אם בשימוש).

החזרה: {”deleted“: int, ”errors“: [str, …]}

Return type:

Dict[str, Any]

פרמטרים:

• user_id (int)

• backup_ids (List[str])

delete_backup(backup_id, user_id)

מחיקת גיבוי

Return type:

bool

פרמטרים:

• backup_id (str)

• user_id (int)