vipr_api.web.routers package

Subpackages

• vipr_api.web.routers.inference package
  • Submodules
  • vipr_api.web.routers.inference.export module
    • export_cli_yaml()
  • vipr_api.web.routers.inference.tasks module
    • cancel_task()
    • check_celery_backend()
    • get_task_progress()
    • health_check()
    • normalize_progress_info()
    • run_inference_async()
  • Module contents
• vipr_api.web.routers.panpe package
  • Submodules
  • vipr_api.web.routers.panpe.config module
    • get_available_panpe_models()
    • get_panpe_parameters()
  • Module contents
• vipr_api.web.routers.reflectorch package
  • Submodules
  • vipr_api.web.routers.reflectorch.config module
    • get_available_models()
    • get_parameters()
    • get_standard_config()
  • Module contents

Submodules

vipr_api.web.routers.config_security module

Config security validation utilities for VIPR API.

Provides validation for configuration data against registry and security patterns.

class vipr_api.web.routers.config_security.ConfigSecurityValidator

Bases: object

Validates VIPR configuration for security issues.

static validate_config_content_security(config_dict: dict[str, Any]) → list[str]

Validate config content for dangerous patterns.

Parameters:

config_dict – Configuration dictionary to validate

Returns:

List of validation error messages

static validate_handler_whitelist(config_dict: dict[str, Any]) → list[str]

Validate that all handlers are from allowed plugins.

Uses new config hierarchy: vipr.inference.{step}

Parameters:

config_dict – Configuration dictionary to validate

Returns:

List of validation error messages

static validate_parameter_bounds(config_dict: dict[str, Any]) → list[str]

Validate parameter values are within safe bounds.

Parameters:

config_dict – Configuration dictionary to validate

Returns:

List of validation warnings (not errors)

static validate_registry_compliance(config_dict: dict[str, Any]) → list[str]

Validate config against registry for dynamic hooks/filters.

Uses new config hierarchy: vipr.inference.hooks/filters

Parameters:

config_dict – VIPR configuration dictionary

Returns:

List of validation error messages

vipr_api.web.routers.config_security.validate_config_security(config_dict: dict[str, Any]) → dict[str, list[str]]

Comprehensive config security validation.

Parameters:

config_dict – Configuration dictionary to validate

Returns:

Dictionary with ‘errors’ and ‘warnings’ lists

vipr_api.web.routers.files module

Files router package.

vipr_api.web.routers.streaming module

Generic Streaming Router for VIPR Framework.

This router provides API endpoints for RabbitMQ streaming functionality that can be used across different domains/plugins, not just reflectorch.

async vipr_api.web.routers.streaming.get_consumer_stats(consumer_id: str) → ConsumerStatsResponse

Get statistics for a specific streaming consumer.

Parameters:

consumer_id – Unique identifier of the consumer

Returns:

ConsumerStatsResponse with detailed consumer statistics

Raises:

HTTPException – If consumer not found

async vipr_api.web.routers.streaming.get_consumer_tasks(consumer_id: str, limit: int | None = None, since: str | None = None) → ConsumerTasksResponse

Get task list for a specific streaming consumer.

Parameters:

• consumer_id – Unique identifier of the consumer

• limit – Maximum number of tasks to return (most recent first)

• since – ISO timestamp to filter tasks triggered after this time

Returns:

ConsumerTasksResponse containing task list and metadata

Raises:

HTTPException – If consumer not found

async vipr_api.web.routers.streaming.list_active_consumers() → dict[str, Any]

List all active streaming consumers.

Returns:

Dict containing list of active consumer IDs and summary stats

async vipr_api.web.routers.streaming.start_streaming_prediction(request: VIPRConfigWithStreaming) → StreamingResponse

Start streaming prediction consumer for real-time processing.

This endpoint starts a RabbitMQ consumer that listens for spectral data messages and triggers individual VIPR inference tasks for each subscan message.

Parameters:

request – StreamingStartRequest containing config and RabbitMQ settings

Returns:

StreamingResponse with consumer ID and status

Raises:

HTTPException – If consumer startup fails

async vipr_api.web.routers.streaming.stop_all_streaming_consumers() → dict[str, Any]

Stop all active streaming consumers.

Returns:

Dict containing final statistics for all stopped consumers

Raises:

HTTPException – If operation fails

async vipr_api.web.routers.streaming.stop_streaming_prediction(consumer_id: str) → dict[str, Any]

Stop a running streaming prediction consumer.

Parameters:

consumer_id – Unique identifier of the consumer to stop

Returns:

Dict containing final consumer statistics

Raises:

HTTPException – If consumer not found or stop fails

Module contents

API Router Module for VIPR Integration.

This package contains the various API routers for integrating VIPR functions into the FastAPI application.