routes: Extract shared auth, model validation, and error handling utilities

Consolidate duplicated boilerplate across all batch pipeline route handlers
into shared utility functions (check_auth_token, check_model_id,
execute_pipeline) and a shared RESPONSES dict in routes/utils.py.

This addresses Victor's feedback on PR livepeer#900 about reducing redundant
pipeline interface patterns and making common behavior properly defined
in one place rather than copy-pasted across every route.

https://claude.ai/code/session_01RvBGa2npztEMxwfAHH3Xve

Author: claude

runner/src/runner/routes/audio_to_text.py

@@ -1,8 +1,6 @@
 import logging
-import os
 from typing import Annotated, Dict, Tuple, Union
 
-import torch
 from fastapi import APIRouter, Depends, File, Form, UploadFile, status
 from fastapi.responses import JSONResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
@@ -12,9 +10,12 @@
 from runner.routes.utils import (
     HTTPError,
     TextResponse,
+    RESPONSES,
+    check_auth_token,
+    check_model_id,
+    execute_pipeline,
     file_exceeds_max_size,
     get_media_duration_ffmpeg,
-    handle_pipeline_exception,
     http_error,
     parse_key_from_metadata,
 )
@@ -37,21 +38,11 @@
     ),
 }
 
-RESPONSES = {
-    status.HTTP_200_OK: {
-        "content": {
-            "application/json": {
-                "schema": {
-                    "x-speakeasy-name-override": "data",
-                }
-            }
-        },
-    },
-    status.HTTP_400_BAD_REQUEST: {"model": HTTPError},
-    status.HTTP_401_UNAUTHORIZED: {"model": HTTPError},
+# Extend shared RESPONSES with additional status codes for this route.
+AUDIO_RESPONSES = {
+    **RESPONSES,
     status.HTTP_413_REQUEST_ENTITY_TOO_LARGE: {"model": HTTPError},
     status.HTTP_415_UNSUPPORTED_MEDIA_TYPE: {"model": HTTPError},
-    status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": HTTPError},
 }
 
 
@@ -76,7 +67,7 @@ def parse_return_timestamps(value: str) -> Union[bool, str]:
 @router.post(
     "/audio-to-text",
     response_model=TextResponse,
-    responses=RESPONSES,
+    responses=AUDIO_RESPONSES,
     description="Transcribe audio files to text.",
     operation_id="genAudioToText",
     summary="Audio To Text",
@@ -86,7 +77,7 @@ def parse_return_timestamps(value: str) -> Union[bool, str]:
 @router.post(
     "/audio-to-text/",
     response_model=TextResponse,
-    responses=RESPONSES,
+    responses=AUDIO_RESPONSES,
     include_in_schema=False,
 )
 def audio_to_text(
@@ -116,23 +107,12 @@ def audio_to_text(
     token: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
 ):
     return_timestamps = parse_return_timestamps(return_timestamps)
-    auth_token = os.environ.get("AUTH_TOKEN")
-    if auth_token:
-        if not token or token.credentials != auth_token:
-            return JSONResponse(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                headers={"WWW-Authenticate": "Bearer"},
-                content=http_error("Invalid bearer token."),
-            )
 
-    if model_id != "" and model_id != pipeline.model_id:
-        return JSONResponse(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            content=http_error(
-                f"pipeline configured with {pipeline.model_id} but called with "
-                f"{model_id}."
-            ),
-        )
+    if auth_error := check_auth_token(token):
+        return auth_error
+
+    if model_error := check_model_id(model_id, pipeline.model_id):
+        return model_error
 
     if file_exceeds_max_size(audio, 50 * 1024 * 1024):
         return JSONResponse(
@@ -154,17 +134,14 @@ def audio_to_text(
             content=http_error("Unable to calculate duration of file"),
         )
 
-    try:
-        return pipeline(
-            audio=audio, return_timestamps=return_timestamps, duration=duration
-        )
-    except Exception as e:
-        if isinstance(e, torch.cuda.OutOfMemoryError):
-            # TODO: Investigate why not all VRAM memory is cleared.
-            torch.cuda.empty_cache()
-        logger.error(f"AudioToText pipeline error: {e}")
-        return handle_pipeline_exception(
-            e,
-            default_error_message="Audio-to-text pipeline error.",
-            custom_error_config=PIPELINE_ERROR_CONFIG,
-        )
+    result, error = execute_pipeline(
+        pipeline,
+        default_error_message="Audio-to-text pipeline error.",
+        custom_error_config=PIPELINE_ERROR_CONFIG,
+        audio=audio,
+        return_timestamps=return_timestamps,
+        duration=duration,
+    )
+    if error:
+        return error
+    return result

runner/src/runner/routes/image_to_image.py

@@ -1,21 +1,19 @@
 import logging
-import os
 import random
 from typing import Annotated, Dict, Tuple, Union
 
-import torch
 from fastapi import APIRouter, Depends, File, Form, UploadFile, status
-from fastapi.responses import JSONResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from PIL import Image, ImageFile
 
 from runner.dependencies import get_pipeline
 from runner.pipelines.base import Pipeline
 from runner.routes.utils import (
-    HTTPError,
     ImageResponse,
-    handle_pipeline_exception,
-    http_error,
+    RESPONSES,
+    check_auth_token,
+    check_model_id,
+    execute_pipeline,
     image_to_data_url,
 )
 
@@ -35,21 +33,6 @@
     )
 }
 
-RESPONSES = {
-    status.HTTP_200_OK: {
-        "content": {
-            "application/json": {
-                "schema": {
-                    "x-speakeasy-name-override": "data",
-                }
-            }
-        },
-    },
-    status.HTTP_400_BAD_REQUEST: {"model": HTTPError},
-    status.HTTP_401_UNAUTHORIZED: {"model": HTTPError},
-    status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": HTTPError},
-}
-
 
 # TODO: Make model_id and other None properties optional once Go codegen tool supports
 # OAPI 3.1 https://github.com/deepmap/oapi-codegen/issues/373
@@ -153,23 +136,11 @@ async def image_to_image(
     pipeline: Pipeline = Depends(get_pipeline),
     token: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
 ):
-    auth_token = os.environ.get("AUTH_TOKEN")
-    if auth_token:
-        if not token or token.credentials != auth_token:
-            return JSONResponse(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                headers={"WWW-Authenticate": "Bearer"},
-                content=http_error("Invalid bearer token."),
-            )
+    if auth_error := check_auth_token(token):
+        return auth_error
 
-    if model_id != "" and model_id != pipeline.model_id:
-        return JSONResponse(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            content=http_error(
-                f"pipeline configured with {pipeline.model_id} but called with "
-                f"{model_id}."
-            ),
-        )
+    if model_error := check_model_id(model_id, pipeline.model_id):
+        return model_error
 
     seed = seed if seed is not None else random.randint(0, 2**32 - 1)
     seeds = [seed + i for i in range(num_images_per_prompt)]
@@ -181,30 +152,25 @@ async def image_to_image(
     images = []
     has_nsfw_concept = []
     for seed in seeds:
-        try:
-            imgs, nsfw_checks = pipeline(
-                prompt=prompt,
-                image=image,
-                strength=strength,
-                loras=loras,
-                guidance_scale=guidance_scale,
-                image_guidance_scale=image_guidance_scale,
-                negative_prompt=negative_prompt,
-                safety_check=safety_check,
-                seed=seed,
-                num_images_per_prompt=1,
-                num_inference_steps=num_inference_steps,
-            )
-        except Exception as e:
-            if isinstance(e, torch.cuda.OutOfMemoryError):
-                # TODO: Investigate why not all VRAM memory is cleared.
-                torch.cuda.empty_cache()
-            logger.error(f"ImageToImagePipeline pipeline error: {e}")
-            return handle_pipeline_exception(
-                e,
-                default_error_message="Image-to-image pipeline error.",
-                custom_error_config=PIPELINE_ERROR_CONFIG,
-            )
+        result, error = execute_pipeline(
+            pipeline,
+            default_error_message="Image-to-image pipeline error.",
+            custom_error_config=PIPELINE_ERROR_CONFIG,
+            prompt=prompt,
+            image=image,
+            strength=strength,
+            loras=loras,
+            guidance_scale=guidance_scale,
+            image_guidance_scale=image_guidance_scale,
+            negative_prompt=negative_prompt,
+            safety_check=safety_check,
+            seed=seed,
+            num_images_per_prompt=1,
+            num_inference_steps=num_inference_steps,
+        )
+        if error:
+            return error
+        imgs, nsfw_checks = result
         images.extend(imgs)
         has_nsfw_concept.extend(nsfw_checks)
 

runner/src/runner/routes/image_to_text.py

@@ -1,8 +1,6 @@
 import logging
-import os
 from typing import Annotated, Dict, Tuple, Union
 
-import torch
 from fastapi import APIRouter, Depends, File, Form, UploadFile, status
 from fastapi.responses import JSONResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
@@ -13,8 +11,11 @@
 from runner.routes.utils import (
     HTTPError,
     ImageToTextResponse,
+    RESPONSES,
+    check_auth_token,
+    check_model_id,
+    execute_pipeline,
     file_exceeds_max_size,
-    handle_pipeline_exception,
     http_error,
 )
 
@@ -31,27 +32,17 @@
     )
 }
 
-RESPONSES = {
-    status.HTTP_200_OK: {
-        "content": {
-            "application/json": {
-                "schema": {
-                    "x-speakeasy-name-override": "data",
-                }
-            }
-        },
-    },
-    status.HTTP_400_BAD_REQUEST: {"model": HTTPError},
-    status.HTTP_401_UNAUTHORIZED: {"model": HTTPError},
+# Extend shared RESPONSES with additional status codes for this route.
+IMAGE_TO_TEXT_RESPONSES = {
+    **RESPONSES,
     status.HTTP_413_REQUEST_ENTITY_TOO_LARGE: {"model": HTTPError},
-    status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": HTTPError},
 }
 
 
 @router.post(
     "/image-to-text",
     response_model=ImageToTextResponse,
-    responses=RESPONSES,
+    responses=IMAGE_TO_TEXT_RESPONSES,
     description="Transform image files to text.",
     operation_id="genImageToText",
     summary="Image To Text",
@@ -61,7 +52,7 @@
 @router.post(
     "/image-to-text/",
     response_model=ImageToTextResponse,
-    responses=RESPONSES,
+    responses=IMAGE_TO_TEXT_RESPONSES,
     include_in_schema=False,
 )
 async def image_to_text(
@@ -79,23 +70,11 @@ async def image_to_text(
     pipeline: Pipeline = Depends(get_pipeline),
     token: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
 ):
-    auth_token = os.environ.get("AUTH_TOKEN")
-    if auth_token:
-        if not token or token.credentials != auth_token:
-            return JSONResponse(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                headers={"WWW-Authenticate": "Bearer"},
-                content=http_error("Invalid bearer token"),
-            )
+    if auth_error := check_auth_token(token):
+        return auth_error
 
-    if model_id != "" and model_id != pipeline.model_id:
-        return JSONResponse(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            content=http_error(
-                f"pipeline configured with {pipeline.model_id} but called with "
-                f"{model_id}"
-            ),
-        )
+    if model_error := check_model_id(model_id, pipeline.model_id):
+        return model_error
 
     if file_exceeds_max_size(image, 50 * 1024 * 1024):
         return JSONResponse(
@@ -104,15 +83,13 @@ async def image_to_text(
         )
 
     image = Image.open(image.file).convert("RGB")
-    try:
-        return ImageToTextResponse(text=pipeline(prompt=prompt, image=image))
-    except Exception as e:
-        if isinstance(e, torch.cuda.OutOfMemoryError):
-            # TODO: Investigate why not all VRAM memory is cleared.
-            torch.cuda.empty_cache()
-        logger.error(f"ImageToTextPipeline error: {e}")
-        return handle_pipeline_exception(
-            e,
-            default_error_message="Image-to-text pipeline error.",
-            custom_error_config=PIPELINE_ERROR_CONFIG,
-        )
+    result, error = execute_pipeline(
+        pipeline,
+        default_error_message="Image-to-text pipeline error.",
+        custom_error_config=PIPELINE_ERROR_CONFIG,
+        prompt=prompt,
+        image=image,
+    )
+    if error:
+        return error
+    return ImageToTextResponse(text=result)