Remove the Hosted Video Inference API helpers

CHANGELOG.md

@@ -13,6 +13,17 @@ All notable changes to this project will be documented in this file.
   not visible; omitted, the server default (0.6) applies
   ([#571](https://github.com/roboflow/roboflow-python/pull/571)).
 
+### Removed
+
+- The Hosted Video Inference API helpers. The service behind them stopped working on
+  2026-07-29 and its endpoints now return 410 Gone. Removed: `predict_video`,
+  `poll_for_video_results` and `poll_until_video_results` on `InferenceModel` and
+  `TrainedModel`, `VideoInferenceModel`, `CLIPModel`, `GazeModel`,
+  `rfapi.get_video_job_status`, the `roboflow video infer` and `roboflow video status`
+  commands and the hidden `run_video_inference_api` alias. To run a model on stored
+  videos, use [Batch Processing](https://docs.roboflow.com/deployment/roboflow-cloud/batch-processing)
+  ([#574](https://github.com/roboflow/roboflow-python/pull/574)).
+
 ## 1.7.3
 
 ### Added

CLI-COMMANDS.md

@@ -542,9 +542,7 @@ roboflow video upload -p my-ar-project -f clip.mov --json
 # { "videoId": "aBcD1234", "status": "uploaded", "resolvedBatch": { ... } }
 ```
 
-Pass `--no-wait` to return as soon as the bytes are stored, then poll separately.
-`video upload-status` reports ingestion state — it is distinct from `video status`, which
-checks a legacy video *inference* job.
+Pass `--no-wait` to return as soon as the bytes are stored, then poll with `video upload-status`.
 
 ```bash
 # 2. Poll ingestion yourself.
@@ -577,16 +575,6 @@ Exit codes follow the CLI contract: `0` success, `1` error, `2` auth, `3` not fo
 `failed` ingestion state and a `--wait` timeout both exit nonzero; the timeout message names
 the video ID so you can re-check it with `video upload-status`.
 
-### Video inference
-
-Separate from native upload: this submits a legacy asynchronous inference job for a trained
-model version.
-
-```bash
-roboflow video infer -p my-project -v 3 -f video.mp4 --fps 10
-roboflow video status <job-id>
-```
-
 ### Shell completion
 
 The fastest path: let the CLI install completion for you. Auto-detects your shell from `$SHELL`.
@@ -676,7 +664,7 @@ Version numbers are always numeric — that's how `x/y` is disambiguated between
 | `asynctasks` | Inspect async background tasks (e.g. project forks) |
 | `trash` | List items in Trash |
 | `universe` | Search Roboflow Universe |
-| `video` | Native video upload/status/annotation, and video inference |
+| `video` | Native video upload, upload status and annotation |
 | `batch` | Batch processing jobs *(coming soon)* |
 | `completion` | Install or generate shell completion scripts (bash, zsh, fish) |
 

roboflow/__init__.py

@@ -17,11 +17,8 @@
 
 try:
     from roboflow.core.project import Project
-    from roboflow.models import CLIPModel, GazeModel  # noqa: F401
 except ImportError:
     Project = None  # type: ignore[assignment,misc]
-    CLIPModel = None  # type: ignore[assignment,misc]
-    GazeModel = None  # type: ignore[assignment,misc]
 
 
 def check_key(api_key, model, notebook, num_retries=0):

roboflow/adapters/rfapi.py

@@ -1968,19 +1968,6 @@ def get_labeling_stats(api_key, workspace_url, *, start_date=None, end_date=None
     return response.json()
 
 
-# ---------------------------------------------------------------------------
-# Phase 2: Video inference status
-# ---------------------------------------------------------------------------
-
-
-def get_video_job_status(api_key, job_id):
-    """GET /videoinfer?jobId={id} — check video inference job status."""
-    response = http.get(f"{API_URL}/videoinfer", params={"api_key": api_key, "job_id": job_id})
-    if response.status_code != 200:
-        raise RoboflowError(response.text)
-    return response.json()
-
-
 # ---------------------------------------------------------------------------
 # Batch Processing (Asset Library orchestration)
 # ---------------------------------------------------------------------------

roboflow/cli/handlers/_aliases.py

@@ -190,17 +190,3 @@ def get_workspace_info_alias(
         print("project", proj)  # noqa: T201
         ver = proj.version(args.version_number)
         print("version", ver)  # noqa: T201
-
-    @app.command("run_video_inference_api", hidden=True)
-    def run_video_inference_api_alias(
-        ctx: typer.Context,
-        project: Annotated[Optional[str], typer.Option("-p", help="Project ID")] = None,
-        version_number: Annotated[Optional[int], typer.Option("-v", help="Version")] = None,
-        video_file: Annotated[Optional[str], typer.Option("-f", help="Video file")] = None,
-        fps: Annotated[int, typer.Option("-fps", help="FPS")] = 5,
-    ) -> None:
-        """Run video inference (hidden legacy alias)."""
-        from roboflow.cli.handlers.video import _video_infer
-
-        args = ctx_to_args(ctx, project=project, version_number=version_number, video_file=video_file, fps=fps)
-        _video_infer(args)

roboflow/cli/handlers/video.py

@@ -1,4 +1,4 @@
-"""Video commands: native video Source ingestion/annotation and legacy video inference."""
+"""Video commands: native video Source ingestion and annotation."""
 
 from __future__ import annotations
 
@@ -10,34 +10,11 @@
 
 video_app = typer.Typer(
     cls=SortedGroup,
-    help="Native video upload/annotation and video inference operations",
+    help="Native video upload and annotation operations",
     no_args_is_help=True,
 )
 
 
-@video_app.command("infer")
-def infer(
-    ctx: typer.Context,
-    project: Annotated[str, typer.Option("-p", "--project", help="Project ID")],
-    version_number: Annotated[int, typer.Option("-v", "--version", help="Model version number")],
-    video_file: Annotated[str, typer.Option("-f", "--file", help="Path to video file")],
-    fps: Annotated[int, typer.Option("--fps", help="Frames per second")] = 5,
-) -> None:
-    """Run video inference."""
-    args = ctx_to_args(ctx, project=project, version_number=version_number, video_file=video_file, fps=fps)
-    _video_infer(args)
-
-
-@video_app.command("status")
-def status(
-    ctx: typer.Context,
-    job_id: Annotated[str, typer.Argument(help="Job ID to check")],
-) -> None:
-    """Check a legacy video inference job's status (not native upload ingestion)."""
-    args = ctx_to_args(ctx, job_id=job_id)
-    _video_status(args)
-
-
 @video_app.command("upload")
 def upload(
     ctx: typer.Context,
@@ -142,87 +119,6 @@ def annotate(
     _video_annotate(args)
 
 
-# ---------------------------------------------------------------------------
-# Business logic (unchanged from argparse version)
-# ---------------------------------------------------------------------------
-
-
-def _video_infer(args) -> None:  # noqa: ANN001
-    import roboflow
-    from roboflow.cli._output import output, output_error
-    from roboflow.config import load_roboflow_api_key
-
-    api_key = args.api_key or load_roboflow_api_key(None)
-    if not api_key:
-        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
-        return
-
-    try:
-        from roboflow.cli._output import suppress_sdk_output
-
-        with suppress_sdk_output():
-            rf = roboflow.Roboflow(api_key)
-            project = rf.workspace().project(args.project)
-            version = project.version(args.version_number)
-            model = getattr(version, "_model", None)
-            if model is None:
-                output_error(
-                    args,
-                    f"No model found for project '{args.project}' version {args.version_number}.",
-                    hint="Train or deploy a model for this version before running video inference.",
-                    exit_code=3,
-                )
-                return
-
-            job_id, _signed_url, _expire_time = model.predict_video(
-                args.video_file,
-                args.fps,
-                prediction_type="batch-video",
-            )
-    except Exception as exc:
-        output_error(args, str(exc))
-        return
-
-    data = {"job_id": job_id, "status": "submitted"}
-    output(args, data, text=f"Video inference submitted. Job ID: {job_id}")
-
-
-def _video_status(args) -> None:  # noqa: ANN001
-    from roboflow.adapters import rfapi
-    from roboflow.cli._output import output, output_error
-    from roboflow.config import load_roboflow_api_key
-
-    api_key = args.api_key or load_roboflow_api_key(None)
-    if not api_key:
-        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
-        return
-
-    try:
-        data = rfapi.get_video_job_status(api_key, args.job_id)
-    except rfapi.RoboflowError as exc:
-        msg = str(exc)
-        if "NOT FOUND" in msg.upper():
-            output_error(
-                args,
-                f"Video job '{args.job_id}' not found.",
-                hint="Check the job ID. You can get job IDs from 'roboflow video infer'.",
-                exit_code=3,
-            )
-        else:
-            output_error(args, msg, exit_code=3)
-        return
-
-    status = data.get("status", "unknown")
-    progress = data.get("progress", "")
-    text_lines = [
-        f"Job ID:   {args.job_id}",
-        f"Status:   {status}",
-    ]
-    if progress:
-        text_lines.append(f"Progress: {progress}")
-    output(args, data, text="\n".join(text_lines))
-
-
 # ---------------------------------------------------------------------------
 # Native video Source business logic
 # ---------------------------------------------------------------------------

roboflow/core/training.py

@@ -72,7 +72,6 @@ def __init__(self, api_key, workspace, project, model_id, model_type=None, metri
         # The second segment addresses the model on /ptFile: a model slug for
         # MMPV, a version number for SMPV.
         self._weights_id = model_id.split("/", 1)[1] if "/" in str(model_id) else model_id
-        self._video_model_cache = None
 
     def predict(self, image_path, hosted=False, confidence=40, overlap=30, format="json", **kwargs):
         """Run hosted inference on an image by this model's id.
@@ -92,72 +91,6 @@ def predict(self, image_path, hosted=False, confidence=40, overlap=30, format="j
         params.update(kwargs)
         return model.predict(image_path, prediction_type=prediction_type, **params)
 
-    def _video_model(self):
-        """Build (and cache) the legacy inference model used for video inference.
-
-        Video upload and result polling still flow through the legacy
-        ``/videoinfer`` endpoints, which the task-specific models implement.
-        Caching keeps ``predict_video`` and the poll methods on one underlying
-        object, so a job started here can be polled without re-passing its id.
-        """
-        if self._video_model_cache is not None:
-            return self._video_model_cache
-
-        from roboflow.models.classification import ClassificationModel
-        from roboflow.models.instance_segmentation import InstanceSegmentationModel
-        from roboflow.models.keypoint_detection import KeypointDetectionModel
-        from roboflow.models.object_detection import ObjectDetectionModel
-        from roboflow.models.semantic_segmentation import SemanticSegmentationModel
-
-        task = task_of_model_type(self.model_type or "")
-        legacy_class = {
-            TASK_CLS: ClassificationModel,
-            TASK_SEG: InstanceSegmentationModel,
-            TASK_SEM: SemanticSegmentationModel,
-            TASK_POSE: KeypointDetectionModel,
-        }.get(task, ObjectDetectionModel)
-
-        legacy_id = f"{self.workspace}/{self.project}/{self._weights_id}"
-        self._video_model_cache = legacy_class(self.__api_key, legacy_id)
-        return self._video_model_cache
-
-    def predict_video(self, video_path, fps=5, additional_models=None, prediction_type="batch-video"):
-        """Run hosted video inference for this model (DNA-era equivalent of the
-        legacy ``version.model.predict_video``).
-
-        Delegates to the task-appropriate legacy inference model built from this
-        model's id, so a ``TrainedModel`` can do everything the old
-        ``version.model`` could. Returns ``(job_id, signed_url, expires)``; poll
-        with :meth:`poll_until_video_results` on the same object.
-
-        NOTE: the legacy ``/videoinfer`` payload is keyed by ``<dataset>/<version>``.
-        For MMPV models addressed by ``<workspace>/<model-slug>`` this routes the
-        slug through as the version segment; verify against staging before relying
-        on it for slug-addressed models.
-        """
-        return self._video_model().predict_video(
-            video_path, fps=fps, additional_models=additional_models, prediction_type=prediction_type
-        )
-
-    def poll_for_video_results(self, job_id=None) -> dict:
-        """Check once for this model's video inference results (DNA-era equivalent
-        of the legacy ``version.model.poll_for_video_results``).
-
-        Returns ``{}`` while the job is still running. Defaults to the job started
-        by the most recent :meth:`predict_video` call on this object.
-        """
-        return self._video_model().poll_for_video_results(job_id)
-
-    def poll_until_video_results(self, job_id=None) -> dict:
-        """Block until this model's video inference job completes, returning the
-        results (DNA-era equivalent of the legacy
-        ``version.model.poll_until_video_results``).
-
-        Defaults to the job started by the most recent :meth:`predict_video` call
-        on this object.
-        """
-        return self._video_model().poll_until_video_results(job_id)
-
     def download(self, format="pt", location="."):
         """Download this model's PyTorch weights to ``location/weights.pt``."""
         weights_url = rfapi.get_model_weights_url(

roboflow/models/__init__.py

@@ -1,2 +0,0 @@
-from .clip import CLIPModel  # noqa: F401
-from .gaze import GazeModel  # noqa: F401