Add workspace image deletion (DATAMAN-162) (#442)

* adding search/search all functionality to workspace

* test

* config

* Delete images from workspace

* fix(pre_commit): 🎨 auto format pre-commit hooks

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: digaobarbosa

docs/index.md

@@ -126,6 +126,18 @@ Or from the CLI:
 roboflow search-export "class:person" -f coco -d my-project -l ./my-export
 ```
 
+### Delete Workspace Images
+
+Delete orphan images (not in any project) from your workspace:
+
+```python
+workspace = rf.workspace()
+
+# Delete orphan images by ID
+result = workspace.delete_images(["image_id_1", "image_id_2"])
+print(f"Deleted: {result['deletedSources']}, Skipped: {result['skippedSources']}")
+```
+
 ### Upload with Metadata
 
 Attach custom key-value metadata to images during upload:

pyproject.toml

@@ -88,6 +88,9 @@ convention = "google"
     "E402", # Module level import not at top of file
     "F401", # Imported but unused
 ]
+"tests/manual/*.py" = [
+    "INP001", # Manual scripts don't need __init__.py
+]
 
 [tool.ruff.lint.pyupgrade]
 # Preserve types, even if a file imports `from __future__ import annotations`.

roboflow/adapters/rfapi.py

@@ -199,6 +199,71 @@ def get_search_export(api_key: str, workspace_url: str, export_id: str, session:
     return response.json()
 
 
+def workspace_search(
+    api_key: str,
+    workspace_url: str,
+    query: str,
+    page_size: int = 50,
+    fields: Optional[List[str]] = None,
+    continuation_token: Optional[str] = None,
+) -> dict:
+    """Search across all images in a workspace using RoboQL syntax.
+
+    Args:
+        api_key: Roboflow API key.
+        workspace_url: Workspace slug/url.
+        query: RoboQL search query (e.g. ``"tag:review"``, ``"project:false"``).
+        page_size: Number of results per page (default 50).
+        fields: Fields to include in each result.
+        continuation_token: Token for fetching the next page.
+
+    Returns:
+        Parsed JSON response with ``results``, ``total``, and ``continuationToken``.
+
+    Raises:
+        RoboflowError: On non-200 response status codes.
+    """
+    url = f"{API_URL}/{workspace_url}/search/v1?api_key={api_key}"
+    payload: Dict[str, Union[str, int, List[str]]] = {
+        "query": query,
+        "pageSize": page_size,
+    }
+    if fields is not None:
+        payload["fields"] = fields
+    if continuation_token is not None:
+        payload["continuationToken"] = continuation_token
+
+    response = requests.post(url, json=payload)
+    if response.status_code != 200:
+        raise RoboflowError(response.text)
+    return response.json()
+
+
+def workspace_delete_images(
+    api_key: str,
+    workspace_url: str,
+    image_ids: List[str],
+) -> dict:
+    """Delete orphan images from a workspace.
+
+    Args:
+        api_key: Roboflow API key.
+        workspace_url: Workspace slug/url.
+        image_ids: List of image IDs to delete.
+
+    Returns:
+        Parsed JSON response with deletion counts.
+
+    Raises:
+        RoboflowError: On non-200 response status codes.
+    """
+    url = f"{API_URL}/{workspace_url}/images?api_key={api_key}"
+    response = requests.delete(url, json={"images": image_ids})
+    if response.status_code != 200:
+        raise RoboflowError(response.text)
+    return response.json()
+
+
 def upload_image(
     api_key,
     project_url,

roboflow/core/workspace.py

@@ -6,7 +6,7 @@
 import os
 import sys
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Generator, List, Optional
 
 import requests
 from PIL import Image
@@ -666,6 +666,112 @@ def _upload_zip(
         except Exception as e:
             print(f"An error occured when uploading the model: {e}")
 
+    def search(
+        self,
+        query: str,
+        page_size: int = 50,
+        fields: Optional[List[str]] = None,
+        continuation_token: Optional[str] = None,
+    ) -> dict:
+        """Search across all images in the workspace using RoboQL syntax.
+
+        Args:
+            query: RoboQL search query (e.g. ``"tag:review"``, ``"project:false"``
+                for orphan images, or free-text for semantic CLIP search).
+            page_size: Number of results per page (default 50).
+            fields: Fields to include in each result.
+                Defaults to ``["tags", "projects", "filename"]``.
+            continuation_token: Token returned by a previous call for fetching
+                the next page.
+
+        Returns:
+            Dict with ``results`` (list), ``total`` (int), and
+            ``continuationToken`` (str or None).
+
+        Example:
+            >>> ws = rf.workspace()
+            >>> page = ws.search("tag:review", page_size=10)
+            >>> print(page["total"])
+            >>> for img in page["results"]:
+            ...     print(img["filename"])
+        """
+        if fields is None:
+            fields = ["tags", "projects", "filename"]
+
+        return rfapi.workspace_search(
+            api_key=self.__api_key,
+            workspace_url=self.url,
+            query=query,
+            page_size=page_size,
+            fields=fields,
+            continuation_token=continuation_token,
+        )
+
+    def delete_images(self, image_ids: List[str]) -> dict:
+        """Delete orphan images from the workspace.
+
+        Only deletes images not associated with any project.
+        Images still in projects are skipped.
+
+        Args:
+            image_ids: List of image IDs to delete.
+
+        Returns:
+            Dict with ``deletedSources`` and ``skippedSources`` counts.
+
+        Example:
+            >>> ws = rf.workspace()
+            >>> result = ws.delete_images(["img_id_1", "img_id_2"])
+            >>> print(result["deletedSources"])
+        """
+        return rfapi.workspace_delete_images(
+            api_key=self.__api_key,
+            workspace_url=self.url,
+            image_ids=image_ids,
+        )
+
+    def search_all(
+        self,
+        query: str,
+        page_size: int = 50,
+        fields: Optional[List[str]] = None,
+    ) -> Generator[List[dict], None, None]:
+        """Paginated search across all images in the workspace.
+
+        Yields one page of results at a time, automatically following
+        ``continuationToken`` until all results have been returned.
+
+        Args:
+            query: RoboQL search query.
+            page_size: Number of results per page (default 50).
+            fields: Fields to include in each result.
+                Defaults to ``["tags", "projects", "filename"]``.
+
+        Yields:
+            A list of result dicts for each page.
+
+        Example:
+            >>> ws = rf.workspace()
+            >>> for page in ws.search_all("tag:review"):
+            ...     for img in page:
+            ...         print(img["filename"])
+        """
+        token = None
+        while True:
+            response = self.search(
+                query=query,
+                page_size=page_size,
+                fields=fields,
+                continuation_token=token,
+            )
+            results = response.get("results", [])
+            if not results:
+                break
+            yield results
+            token = response.get("continuationToken")
+            if not token:
+                break
+
     def search_export(
         self,
         query: str,

tests/manual/demo_workspace_search.py

@@ -0,0 +1,42 @@
+"""Manual demo for workspace-level search (DATAMAN-163).
+
+Usage:
+    python tests/manual/demo_workspace_search.py
+
+Uses staging credentials from CLAUDE.md.
+"""
+
+import os
+
+import roboflow
+
+thisdir = os.path.dirname(os.path.abspath(__file__))
+os.environ["ROBOFLOW_CONFIG_DIR"] = f"{thisdir}/data/.config"
+
+WORKSPACE = "model-evaluation-workspace"
+
+rf = roboflow.Roboflow()
+ws = rf.workspace(WORKSPACE)
+
+# --- Single page search ---
+print("=== Single page search ===")
+page = ws.search("project:false", page_size=5)
+print(f"Total results: {page['total']}")
+print(f"Results in this page: {len(page['results'])}")
+print(f"Continuation token: {page.get('continuationToken')}")
+for img in page["results"]:
+    print(f"  - {img.get('filename', 'N/A')}")
+
+# --- Paginated search_all ---
+print("\n=== Paginated search_all (page_size=3, max 2 pages) ===")
+count = 0
+for page_results in ws.search_all("*", page_size=3):
+    count += 1
+    print(f"Page {count}: {len(page_results)} results")
+    for img in page_results:
+        print(f"  - {img.get('filename', 'N/A')}")
+    if count >= 2:
+        print("(stopping after 2 pages for demo)")
+        break
+
+print("\nDone.")