API Plugin - Usage Guide

This document provides practical examples and best practices for using the API Plugin.

Table of Contents

Usage Examples

@api Decorator Usage

Real example from vipr-core/vipr/plugins/api/controllers.py (UIController):

from vipr.plugins.api.decorator import api
from vipr.plugins.api.result_storage import ResultStorage
from cement import Controller, ex

@api("/ui/result", "GET",
     response=UIResultResponse,
     tags=["ui"],
     operation_id="getUIResult",
     summary="Get Stored UI Result",
     description="Retrieve a stored UI result by UUID")
@ex(
    help='Retrieve stored UI result by ID',
    arguments=[
        (['--id'], {'help': 'Result ID (UUID)', 'required': True}),
    ]
)
def get_result(self):
    """Retrieve stored UI result by UUID."""
    result_id = self.app.pargs.id
    
    # Get result from storage
    storage = ResultStorage()
    result = storage.get_result(result_id)
    
    if result is not None:
        return {
            "result_id": result_id,
            "status": "completed",
            "data": result,
            "retrieved_at": self._get_current_timestamp()
        }
    
    raise Exception(f"UI result {result_id} not found")

Key Features Demonstrated:

  • Dual Access: Available via CLI (vipr ui get-result --id <uuid>) and HTTP (GET /ui/result?id=<uuid>)

  • Response Model: Type-safe UIResultResponse for validation

  • Direct Retrieval: Immediate result access from storage

  • OpenAPI Integration: Automatic schema generation with tags and descriptions

Note: For long-running operations like inference, use manual FastAPI routers with Celery instead of @api decorator. See README.md for details.

DataCollector - Tables

# Simple row-by-row construction
app.datacollector.table(
    'parameters',        # id: unique identifier for programmatic access
    'Model Parameters'   # title: human-readable display name
)\
    .add_row(name='SLD', value=2.5, unit='10⁻⁶ Ų⁻²')\
    .add_row(name='thickness', value=100, unit='Å')\
    .add_row(name='roughness', value=5, unit='Å')

# Column-based construction
app.datacollector.table(
    'data',         # id: unique identifier
    'Input Data'    # title: display name
)\
    .add_column('q', [0.01, 0.02, 0.03])\
    .add_column('R', [0.1, 0.05, 0.02])

DataCollector - Diagrams

Diagrams use a two-step process:

  1. Store data: Use set_data(name, values) to store individual data arrays

  2. Define series: Use add_series(x, y, error_bars, label) to define which data arrays to plot together

Each series represents one curve/line in the diagram and establishes the relationship between X and Y data.

# Create diagram with data and series
app.datacollector.diagram('reflectivity', 'Reflectivity Curve')\
    .set_data('q', q_values)\
    .set_data('R_predicted', R_pred)\
    .set_data('R_predicted_error', R_pred_err)\
    .set_data('R_experimental', R_exp)\
    .set_data('R_experimental_error', R_exp_err)\
    .add_series('q', 'R_predicted', error_bars='R_predicted_error', label='Predicted')\
    .add_series('q', 'R_experimental', error_bars='R_experimental_error', label='Experimental')\
    .set_metadata(
        x_label='Q (Ų⁻¹)',
        y_label='Reflectivity',
        y_scale='log',
        x_scale='linear'
    )

Note on CSV extraction: When diagrams have series definitions, each series is saved as a separate CSV file with aligned X-Y columns and optional error bars. This ensures proper data alignment for each series.

DataCollector - Images

# From matplotlib figure
import matplotlib.pyplot as plt

fig, ax = plt.subplots()
ax.plot(x, y)
ax.set_xlabel('X')
ax.set_ylabel('Y')

app.datacollector.image('plot', 'My Plot')\
    .set_from_matplotlib(fig, format='png', dpi=150)

# From file
app.datacollector.image('logo', 'Logo')\
    .set_from_file('/path/to/image.png')

# From bytes
with open('image.png', 'rb') as f:
    image_bytes = f.read()

app.datacollector.image('custom', 'Custom Image')\
    .set_from_bytes(image_bytes, format='png')

DataCollector - Logging

Always use Cement logging - it’s automatically forwarded to DataCollector:

self.app.log.info("Processing started")
self.app.log.warning("Warning message")
self.app.log.error("Error occurred")

Why? The VIPRLogHandler automatically forwards all Cement logging to DataCollector for frontend timeline display, providing both console output and UI integration without code duplication.

Batch Processing

# Batch processing example
batch_data = [spectrum1, spectrum2, spectrum3]

# Set batch-level metadata
app.datacollector.set_batch_metadata(
    total_items=len(batch_data),
    processing_mode='batch'
)

# Process each item with dedicated collector (keeps data isolated per item/spectrum)
for i, spectrum in enumerate(batch_data):
    # Create item-specific collector - each item gets its own tables/diagrams/images
    # This prevents data mixing: item 0 has its own 'results' table, item 1 has its own, etc.
    item_dc = app.datacollector.create_item_collector(i)
    
    # Add item-specific data
    item_dc.table('results', f'Results {i}')\
        .add_row(sld=spectrum.sld, thickness=spectrum.thickness)
    
    item_dc.diagram('reflectivity', f'Reflectivity {i}')\
        .set_data('q', spectrum.q)\
        .set_data('R', spectrum.R)\
        .add_series('q', 'R', label=f'Spectrum {i}')

# Add global logs
app.datacollector.add_global_log(
    f"Batch processing complete: {len(batch_data)} items",
    LogLevel.INFO
)

Hook Integration

The API Plugin automatically stores results via the INFERENCE_COMPLETE_HOOK:

# In __init__.py
def store_ui_data_directly(app, inference=None):
    """Store UI data after inference completion."""
    # Collect buffered logs
    if hasattr(app.log, '_log_buffer'):
        app.datacollector.data.global_logs.extend(app.log._log_buffer)
        app.log._log_buffer.clear()
    
    # Get result_id from config
    result_id = app.config.get('vipr', 'result_id')
    
    if result_id:
        # Save all collected data
        app.datacollector.save_result(result_id)

app.hook.register('INFERENCE_COMPLETE_HOOK', store_ui_data_directly, weight=200)

Weight 200 ensures storage occurs after all other hooks complete.

CLI Commands

# Retrieve stored result
vipr ui get-result --id <uuid>

# Check if result exists
vipr ui get-result-status --id <uuid>

# Get storage information
vipr ui get-storage-info

# Cleanup old results (default: 24 hours)
vipr ui cleanup-storage --max-age-hours 24

Best Practices

1. Logging

Always use Cement logging - it’s automatically forwarded to DataCollector:

self.app.log.info("Processing started")
self.app.log.warning("Warning message")
self.app.log.error("Error occurred")

2. Batch Processing

Always use item collectors for batch data:

for i, item in enumerate(batch):
    item_dc = app.datacollector.create_item_collector(i)
    item_dc.table('results').add_row(value=item.value)

Integration Examples

Backend Integration

Example from vipr-api/services/vipr/api/web/routers/inference/tasks.py:

@router.get("/progress/{task_id}")
async def get_task_progress(task_id: str):
    """Get task progress (status and metadata only)."""
    task_result = AsyncResult(task_id, app=celery_app)
    
    if task_result.state == CeleryState.SUCCESS:
        result_data = task_result.result  # Metadata only (result_id, status, etc.)
        
        return TaskProgressResponse(
            task_id=task_id,
            status=CeleryState.SUCCESS,
            result=result_data  # No UI data - keeps progress polling fast
        )

Note: Progress endpoint returns only task metadata. UI visualization data (tables, diagrams, images) must be fetched separately via /ui/result/{id} to keep progress polling lightweight and fast.

Frontend Integration

Example from vipr-frontend/stores/inferenceStore.ts:

import type { UIData } from '@vipr/api/index'

export const useInferenceStore = defineStore('inference', () => {
  const displayData = ref<UIData | null>(null);
  
  async function runInference() {
    // 1. Start async inference
    const response = await $inferenceApi.runInferenceAsyncApiInferenceRunPost(config);
    const taskId = response.data?.task_id;
    
    // 2. Poll progress
    const progressTracking = useProgressTracking(taskId);
    await progressTracking.startPolling();
    
    // 3. Fetch final results using task_id as unified identifier
    const resultResponse = await $uiApi.getUIResult(taskId);
    
    if (resultResponse?.data?.data) {
      displayData.value = resultResponse.data.data; // Type: UIData
      // Access: displayData.value.items[0].tables
      //         displayData.value.items[0].diagrams
      //         displayData.value.global_logs
    }
  }
  
  return { displayData, runInference };
});

FastAPI Router Generation

The build_router() function in vipr-framework automatically generates FastAPI routers from @api decorated methods:

# In vipr-framework/services/vipr/main.py
from vipr.plugins.api.fastapi import build_router

# Generate router from all @api decorated methods
api_router = build_router(app)
fastapi_app.include_router(api_router)

Flow:

  1. Scan all loaded controllers for @api decorated methods

  2. Extract metadata (path, method, request/response models)

  3. Generate FastAPI endpoint with proper validation

  4. Register with OpenAPI schema

See Also