feat: add scroll and batch delete methods to Qdrant client
Add bulk operations needed for maintenance: - scroll_all_points(): paginated iteration over all points - delete_by_ids(): batch delete points by ID list 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -551,6 +551,97 @@ class QdrantClientWrapper:
|
||||
logger.error(f"Search failed: {e}", exc_info=True)
|
||||
return []
|
||||
|
||||
async def scroll_all_points(
|
||||
self,
|
||||
collection_name: str,
|
||||
batch_size: int = 100,
|
||||
with_payload: bool = True,
|
||||
with_vectors: bool = False,
|
||||
filter_conditions: Optional[Dict[str, Any]] = None
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Scroll through all points in a collection.
|
||||
|
||||
Args:
|
||||
collection_name: Collection name
|
||||
batch_size: Number of points per batch
|
||||
with_payload: Include payload in results
|
||||
with_vectors: Include vectors in results
|
||||
filter_conditions: Optional filter conditions
|
||||
|
||||
Returns:
|
||||
List of all points with id and payload
|
||||
"""
|
||||
all_points = []
|
||||
offset = None
|
||||
|
||||
# Build filter if provided
|
||||
scroll_filter = None
|
||||
if filter_conditions:
|
||||
conditions = []
|
||||
for key, value in filter_conditions.items():
|
||||
conditions.append(
|
||||
FieldCondition(key=key, match=MatchValue(value=value))
|
||||
)
|
||||
scroll_filter = Filter(must=conditions)
|
||||
|
||||
try:
|
||||
while True:
|
||||
points, next_offset = self.client.scroll(
|
||||
collection_name=collection_name,
|
||||
scroll_filter=scroll_filter,
|
||||
limit=batch_size,
|
||||
offset=offset,
|
||||
with_payload=with_payload,
|
||||
with_vectors=with_vectors
|
||||
)
|
||||
|
||||
for point in points:
|
||||
all_points.append({
|
||||
"id": str(point.id),
|
||||
"payload": dict(point.payload) if point.payload else {}
|
||||
})
|
||||
|
||||
if next_offset is None:
|
||||
break
|
||||
offset = next_offset
|
||||
|
||||
return all_points
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to scroll collection {collection_name}: {e}", exc_info=True)
|
||||
return []
|
||||
|
||||
async def delete_by_ids(
|
||||
self,
|
||||
collection_name: str,
|
||||
point_ids: List[str]
|
||||
) -> int:
|
||||
"""
|
||||
Delete points by their IDs.
|
||||
|
||||
Args:
|
||||
collection_name: Collection name
|
||||
point_ids: List of point IDs to delete
|
||||
|
||||
Returns:
|
||||
Number of points deleted
|
||||
"""
|
||||
if not point_ids:
|
||||
return 0
|
||||
|
||||
try:
|
||||
self.client.delete(
|
||||
collection_name=collection_name,
|
||||
points_selector=point_ids
|
||||
)
|
||||
logger.info(f"Deleted {len(point_ids)} points from {collection_name}")
|
||||
return len(point_ids)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to delete points by IDs: {e}", exc_info=True)
|
||||
return 0
|
||||
|
||||
async def list_collections(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
List all collections with stats.
|
||||
|
||||
Reference in New Issue
Block a user