--- title: snapshot_service | CodeWeaver Docs description: API reference for codeweaver.engine.services.snapshot_service url: "https://docs.knitli.com/api/engine/services/snapshot_service" type: static generatedAt: "2026-04-17T17:21:08.600Z" --- # snapshot_service [Open in ChatGPT](https://chatgpt.com/?q=Read%20https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F.%20I%20want%20to%20ask%20questions%20about%20it.)[Open in Claude](https://claude.ai/new?q=Read%20https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F.%20I%20want%20to%20ask%20questions%20about%20it.)[View in Markdown](/codeweaver/api/engine/services/snapshot_service.md) [Share on LinkedIn](https://www.linkedin.com/sharing/share-offsite/?url=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F)[Share on X](https://x.com/intent/tweet?url=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F&text=snapshot_service)[Share on Threads](https://threads.net/intent/post?url=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F&text=snapshot_service)[Share on Bluesky](https://bsky.app/intent/compose?text=snapshot_service%20https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F)[Share on Facebook](https://www.facebook.com/sharer/sharer.php?u=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F)[Share on Reddit](https://reddit.com/submit?url=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F&title=snapshot_service)[Share on Hacker News](https://news.ycombinator.com/submitlink?u=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F&t=snapshot_service)[Share on Email](mailto:?subject=snapshot_service&body=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F)[Share on WhatsApp](https://wa.me/?text=snapshot_service%20https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F)[Share on Telegram](https://t.me/share/url?url=https%3A%2F%2Fdocs.knitli.com%2Fcodeweaver%2Fapi%2Fengine%2Fservices%2Fsnapshot_service%2F&text=snapshot_service) # `codeweaver.engine.services.snapshot_service` [Section titled “codeweaver.engine.services.snapshot_service”](#codeweaverengineservicessnapshot_service) Snapshot backup service for Qdrant vector store disaster recovery. This service creates and manages periodic snapshots of the vector store for point-in-time recovery scenarios. It integrates with the failover maintenance loop and handles snapshot retention, cleanup, and storage management. ## Class: `CleanupStatsDict` [Section titled “Class: CleanupStatsDict”](#class-cleanupstatsdict) Dictionary for snapshot cleanup statistics. ## Class: `QdrantSnapshotBackupService` [Section titled “Class: QdrantSnapshotBackupService”](#class-qdrantsnapshotbackupservice) Service for creating and managing Qdrant vector store snapshots. This service provides periodic snapshot creation for disaster recovery, with automatic retention management and support for local/cloud storage. Features: - Periodic snapshot creation - Automatic retention management (keep N most recent snapshots) - Local and cloud storage support - Graceful error handling - Integration with failover maintenance loop ### Method: `cleanup_old_snapshots` [Section titled “Method: cleanup_old_snapshots”](#method-cleanup_old_snapshots) ** ``` cleanup_old_snapshots() ``` Delete old snapshots beyond retention count. Keeps the N most recent snapshots (configured by retention_count). Returns: Dictionary with cleanup statistics: - total: Total snapshots found - kept: Number of snapshots retained - deleted: Number of snapshots deleted - failed: Number of deletion failures Example: >>> stats = await service.cleanup_old_snapshots() >>> print(f”Deleted {stats[‘deleted’]} old snapshots”) ### Method: `create_snapshot` [Section titled “Method: create_snapshot”](#method-create_snapshot) ** ``` create_snapshot() ``` Create a new snapshot of the collection. Args: wait: Whether to wait for snapshot creation to complete Returns: Snapshot name if successful, None otherwise Example: >>> service = QdrantSnapshotBackupService(vector_store) >>> snapshot_name = await service.create_snapshot() >>> print(f”Created snapshot: {snapshot_name}“) ### Method: `delete_snapshot` [Section titled “Method: delete_snapshot”](#method-delete_snapshot) ** ``` delete_snapshot() ``` Delete a specific snapshot. Args: snapshot_name: Name of the snapshot to delete Returns: True if successfully deleted, False otherwise ### Method: `get_latest_snapshot` [Section titled “Method: get_latest_snapshot”](#method-get_latest_snapshot) ** ``` get_latest_snapshot() ``` Get metadata for the most recent snapshot. Returns: Snapshot metadata dictionary, or None if no snapshots exist ### Method: `list_snapshots` [Section titled “Method: list_snapshots”](#method-list_snapshots) ** ``` list_snapshots() ``` List all available snapshots for the collection. Returns: List of snapshot metadata dictionaries Example: >>> snapshots = await service.list_snapshots() >>> for snapshot in snapshots: … print(f”{snapshot[‘name’]}: {snapshot[‘size’]} bytes”) ### Method: `restore_snapshot` [Section titled “Method: restore_snapshot”](#method-restore_snapshot) ** ``` restore_snapshot() ``` Restore the collection from a snapshot. WARNING: This operation replaces the entire collection with the snapshot state. Args: snapshot_name: Name of the snapshot to restore wait: Whether to wait for restore operation to complete Returns: True if restore initiated successfully, False otherwise Example: >>> success = await service.restore_snapshot(“snapshot_20250127_120000”) >>> if success: … print(“Restore completed successfully”) ### Method: `snapshot_and_cleanup` [Section titled “Method: snapshot_and_cleanup”](#method-snapshot_and_cleanup) ** ``` snapshot_and_cleanup() ``` Create a new snapshot and clean up old ones. This is the main method for periodic snapshot maintenance. It combines snapshot creation and retention management. Args: wait: Whether to wait for snapshot creation to complete Returns: SnapshotCleanupResults: Dictionary with operation results: - snapshot_created: Whether new snapshot was created successfully - snapshot_name: Name of new snapshot (None if failed) - cleanup_stats: Statistics from cleanup operation Example: >>> result = await service.snapshot_and_cleanup() >>> if result[“snapshot_created”]: … print(f”Created: {result[‘snapshot_name’]}”) … print(f”Cleaned up: {result[‘cleanup_stats’][‘deleted’]} old snapshots”) ## Class: `SnapshotMetaDict` [Section titled “Class: SnapshotMetaDict”](#class-snapshotmetadict) Metadata dictionary for snapshot operations results.