vipr_api.web.models package¶

Submodules¶

vipr_api.web.models.plugins module¶

Plugin Management Models for FastAPI endpoints.

class vipr_api.web.models.plugins.PluginBatchDeleteResponse(*, success: bool, total_operations: int, successful_operations: int, failed_operations: int, results: List[PluginBatchResult], config_updated: bool = False)¶

Bases: BaseModel

Response model for batch plugin delete endpoint.

config_updated: bool¶

failed_operations: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

results: List[PluginBatchResult]¶

success: bool¶

successful_operations: int¶

total_operations: int¶

class vipr_api.web.models.plugins.PluginBatchOperation(*, plugin_name: str, enabled: bool)¶

Bases: BaseModel

Single operation in batch toggle request.

enabled: bool¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plugin_name: str¶

class vipr_api.web.models.plugins.PluginBatchResult(*, plugin_name: str, success: bool, enabled: bool | None = None, error: str | None = None, files_deleted: ~typing.List[str] = <factory>)¶

Bases: BaseModel

Result for a single plugin operation in batch.

enabled: bool | None¶

error: str | None¶

files_deleted: List[str]¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plugin_name: str¶

success: bool¶

class vipr_api.web.models.plugins.PluginBatchToggleResponse(*, success: bool, total_operations: int, successful_operations: int, failed_operations: int, results: List[PluginBatchResult], requires_restart: bool)¶

Bases: BaseModel

Response model for batch plugin toggle endpoint.

failed_operations: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

requires_restart: bool¶

results: List[PluginBatchResult]¶

success: bool¶

successful_operations: int¶

total_operations: int¶

class vipr_api.web.models.plugins.PluginDeleteResponse(*, success: bool, message: str, plugin_name: str, files_deleted: ~typing.List[str] = <factory>, config_updated: bool = False)¶

Bases: BaseModel

Response model for plugin delete endpoint.

config_updated: bool¶

files_deleted: List[str]¶

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plugin_name: str¶

success: bool¶

class vipr_api.web.models.plugins.PluginListResponse(*, builtin: ~typing.List[~vipr_api.web.models.plugins.PluginStatusModel] = <factory>, runtime: ~typing.List[~vipr_api.web.models.plugins.PluginStatusModel] = <factory>, total_count: int = 0, enabled_count: int = 0, runtime_count: int = 0, builtin_count: int = 0)¶

Bases: BaseModel

Response model for plugin list endpoint.

builtin: List[PluginStatusModel]¶

builtin_count: int¶

enabled_count: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

runtime: List[PluginStatusModel]¶

runtime_count: int¶

total_count: int¶

class vipr_api.web.models.plugins.PluginReloadResponse(*, success: bool, message: str, restart_triggered: bool = False, restart_method: str | None = None)¶

Bases: BaseModel

Response model for plugin reload endpoint.

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

restart_method: str | None¶

restart_triggered: bool¶

success: bool¶

class vipr_api.web.models.plugins.PluginStatus(value)¶

Bases: str, Enum

Plugin status enumeration.

DISABLED = 'disabled'¶

ENABLED = 'enabled'¶

ERROR = 'error'¶

class vipr_api.web.models.plugins.PluginStatusModel(*, name: str, type: PluginType, status: PluginStatus, enabled: bool, loaded: bool, file_path: str | None = None, description: str | None = None, size_bytes: int | None = None, last_modified: datetime | None = None, validation_result: ValidationResult | None = None)¶

Bases: BaseModel

Plugin status information.

description: str | None¶

enabled: bool¶

file_path: str | None¶

last_modified: datetime | None¶

loaded: bool¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

size_bytes: int | None¶

status: PluginStatus¶

type: PluginType¶

validation_result: ValidationResult | None¶

class vipr_api.web.models.plugins.PluginToggleRequest(*, enabled: bool)¶

Bases: BaseModel

Model for plugin enable/disable request.

enabled: bool¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class vipr_api.web.models.plugins.PluginToggleResponse(*, success: bool, message: str, plugin_name: str, enabled: bool, requires_restart: bool)¶

Bases: BaseModel

Response model for plugin toggle endpoint.

enabled: bool¶

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plugin_name: str¶

requires_restart: bool¶

success: bool¶

class vipr_api.web.models.plugins.PluginType(value)¶

Bases: str, Enum

Plugin type enumeration.

BUILTIN = 'builtin'¶

RUNTIME = 'runtime'¶

class vipr_api.web.models.plugins.PluginUploadResponse(*, success: bool, message: str, plugin_name: str, validation_result: ValidationResult)¶

Bases: BaseModel

Response model for plugin upload endpoint.

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plugin_name: str¶

success: bool¶

validation_result: ValidationResult¶

class vipr_api.web.models.plugins.ValidationResult(*, valid: bool, errors: ~typing.List[str] = <factory>, warnings: ~typing.List[str] = <factory>, has_load_function: bool = False, syntax_valid: bool = False, security_passed: bool = False)¶

Bases: BaseModel

Plugin validation result.

errors: List[str]¶

has_load_function: bool¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

security_passed: bool¶

syntax_valid: bool¶

valid: bool¶

warnings: List[str]¶

vipr_api.web.models.progress module¶

Progress tracking models for VIPR API.

class vipr_api.web.models.progress.CeleryExecutionMode(value)¶

Bases: str, Enum

Celery execution modes.

ASYNC = 'async'¶

EAGER = 'eager'¶

class vipr_api.web.models.progress.CeleryHealthStatus(value)¶

Bases: str, Enum

Celery health check status.

HEALTHY = 'healthy'¶

UNAVAILABLE = 'unavailable'¶

class vipr_api.web.models.progress.HealthCheckResponse(*, celery_mode: CeleryExecutionMode, can_execute: bool, redis_required: bool, redis_available: bool | None = None, status: CeleryHealthStatus, error: str | None = None)¶

Bases: BaseModel

Response model for health check endpoint.

can_execute: bool¶

celery_mode: CeleryExecutionMode¶

error: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

redis_available: bool | None¶

redis_required: bool¶

status: CeleryHealthStatus¶

class vipr_api.web.models.progress.TaskProgressResponse(*, task_id: str, status: CeleryState, message: str, progress: Dict[str, Any] | None = None, result: Dict[str, Any] | None = None, ui_data: Dict[str, Any] | None = None, error: str | None = None)¶

Bases: BaseModel

Response model for task progress endpoints.

error: str | None¶

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

progress: Dict[str, Any] | None¶

result: Dict[str, Any] | None¶

status: CeleryState¶

task_id: str¶

ui_data: Dict[str, Any] | None¶

vipr_api.web.models.reflectometry module¶

Reflectorch-specific models for the VIPR framework.

This module contains models specific to reflectometry data processing using reflectorch.

class vipr_api.web.models.reflectometry.AvailableModel(*, name: str, max_num_layers: int, param_ranges: ModelParameterRanges)¶

Bases: BaseModel

Information about an available reflectometry model.

max_num_layers: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

param_ranges: ModelParameterRanges¶

class vipr_api.web.models.reflectometry.BoundWidthRanges(*, thicknesses: list[float], roughnesses: list[float], slds: list[float], r_scale: list[float] | None = None, q_offset: list[float] | None = None, log10_background: list[float] | None = None)¶

Bases: BaseModel

Bound width ranges for subprior intervals.

log10_background: list[float] | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

q_offset: list[float] | None¶

r_scale: list[float] | None¶

roughnesses: list[float]¶

slds: list[float]¶

thicknesses: list[float]¶

class vipr_api.web.models.reflectometry.ModelParameterRanges(*, thicknesses: list[float], roughnesses: list[float], slds: list[float])¶

Bases: BaseModel

Parameter ranges for a reflectometry model.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

roughnesses: list[float]¶

slds: list[float]¶

thicknesses: list[float]¶

class vipr_api.web.models.reflectometry.ParameterBounds(*, thicknesses: list[float], roughnesses: list[float], slds: list[float], r_scale: list[float] | None = None, q_offset: list[float] | None = None, log10_background: list[float] | None = None)¶

Bases: BaseModel

Reflectometry-specific parameter bounds for model configuration.

log10_background: list[float] | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

q_offset: list[float] | None¶

r_scale: list[float] | None¶

roughnesses: list[float]¶

slds: list[float]¶

thicknesses: list[float]¶

class vipr_api.web.models.reflectometry.ReflectometryModelParameters(*, max_layers: int, parameter_bounds: ParameterBounds, bound_width_ranges: BoundWidthRanges, requires_q_resolution: bool)¶

Bases: BaseModel

Reflectometry-specific model parameters.

bound_width_ranges: BoundWidthRanges¶

max_layers: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameter_bounds: ParameterBounds¶

requires_q_resolution: bool¶

vipr_api.web.models.streaming module¶

Pydantic models for generic streaming functionality.

This module contains data models for RabbitMQ streaming endpoints that can be used across different domains/plugins, not just reflectorch.

class vipr_api.web.models.streaming.ConsumerStatsInfo(*, total_tasks_triggered: int, is_consuming: bool, last_message_time: str | None = None)¶

Bases: BaseModel

Consumer statistics information for tasks response.

is_consuming: bool¶

last_message_time: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

total_tasks_triggered: int¶

class vipr_api.web.models.streaming.ConsumerStatsResponse(*, consumer_id: str, is_consuming: bool, messages_received: int, tasks_triggered: int, processing_errors: int, start_time: str | None = None, last_message_time: str | None = None, uptime_seconds: float | None = None, configuration: dict[str, str])¶

Bases: BaseModel

Consumer statistics response.

configuration: dict[str, str]¶

consumer_id: str¶

is_consuming: bool¶

last_message_time: str | None¶

messages_received: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

processing_errors: int¶

start_time: str | None¶

tasks_triggered: int¶

uptime_seconds: float | None¶

class vipr_api.web.models.streaming.ConsumerTasksResponse(*, consumer_id: str, total_tasks_available: int, tasks: list[StreamingTask], consumer_stats: ConsumerStatsInfo)¶

Bases: BaseModel

Response containing consumer tasks and metadata.

consumer_id: str¶

consumer_stats: ConsumerStatsInfo¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tasks: list[StreamingTask]¶

total_tasks_available: int¶

class vipr_api.web.models.streaming.RabbitMQConfig(*, rabbitmq_url: str = 'amqp://localhost:5672/', exchange_name: str = 'vipr.experiments', routing_pattern: str = 'reflectometry.growth.#')¶

Bases: BaseModel

Configuration for RabbitMQ connection.

exchange_name: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rabbitmq_url: str¶

routing_pattern: str¶

class vipr_api.web.models.streaming.StreamingConfig(*, rabbitmq_config: RabbitMQConfig)¶

Bases: BaseModel

Streaming configuration wrapper.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rabbitmq_config: RabbitMQConfig¶

class vipr_api.web.models.streaming.StreamingResponse(*, consumer_id: str, status: str, message: str)¶

Bases: BaseModel

Response from streaming operations.

consumer_id: str¶

message: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: str¶

class vipr_api.web.models.streaming.StreamingTask(*, task_id: str, triggered_at: str, growth_step: int, scan_id: str, csv_file_path: str, routing_key: str)¶

Bases: BaseModel

Individual streaming task information.

csv_file_path: str¶

growth_step: int¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

routing_key: str¶

scan_id: str¶

task_id: str¶

triggered_at: str¶

class vipr_api.web.models.streaming.VIPRConfigWithStreaming(*, config_name: str | None = None, vipr: VIPRInference, streaming: StreamingConfig | None = None)¶

Bases: BaseModel

Complete VIPR configuration including streaming capabilities.

config_name: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

streaming: StreamingConfig | None¶

vipr: VIPRInference¶

vipr_api.web.models package¶

Submodules¶

vipr_api.web.models.plugins module¶

vipr_api.web.models.progress module¶

vipr_api.web.models.reflectometry module¶

vipr_api.web.models.streaming module¶

Module contents¶