Skip to main content

REST Connect Filter

The REST Connect filter is an input filter for OpenFilter that provides HTTP REST API endpoints to receive data from external sources. It exposes FastAPI endpoints that can accept JSON payloads and binary data, then forwards the received data through the OpenFilter pipeline as Frame objects.

Overview​

The REST Connect filter is designed to handle HTTP-based data ingestion scenarios where you need to:

  • Accept data from web applications and external systems
  • Provide RESTful API endpoints for data ingestion
  • Support various HTTP methods (GET, POST, PUT, PATCH, DELETE)
  • Load local files from a specified resource path
  • Convert HTTP requests into OpenFilter Frame objects
  • Handle JSON and binary data payloads

Key Features​

  • REST API Endpoints: FastAPI-based HTTP server with multiple endpoints
  • JSON Payload Support: Direct JSON data ingestion
  • Local File Loading: Serve files from local resource path
  • Path Parameters: Dynamic endpoint routing with path parameters
  • HTTP Methods: Support for GET, POST, PUT, PATCH, DELETE
  • Data Conversion: Automatic conversion to Frame objects
  • Resource Management: Local file serving capabilities

Configuration​

Basic Configuration​

from openfilter.filter_runtime.filter import Filter
from openfilter.filter_runtime.filters.rest import Rest

# Simple REST API server
Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
    )),
])

Advanced Configuration​

# REST API with local resources
Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
        resource_path='/path/to/local/files',
        methods=['GET', 'POST', 'PUT'],
        endpoints=['/data', '/files/{filename}'],
    )),
])

Environment Variables​

You can configure via environment variables:

export FILTER_OUTPUTS="tcp://*:5550"
export FILTER_HOST="0.0.0.0"
export FILTER_PORT="8080"
export FILTER_RESOURCE_PATH="/path/to/files"
export FILTER_METHODS="GET,POST,PUT"
export FILTER_ENDPOINTS="/data,/files/{filename}"

HTTP Endpoints​

Default Endpoints​

The REST Connect filter provides several built-in endpoints:

1. Root Endpoint (/)​

  • Method: GET
  • Purpose: Health check and basic information
  • Response: JSON with server status

2. Data Endpoint (/data)​

  • Method: POST, PUT, PATCH
  • Purpose: JSON data ingestion
  • Content-Type: application/json
  • Body: JSON payload

3. Dynamic File Endpoint (/files/{filename})​

  • Method: GET, POST, PUT, PATCH, DELETE
  • Purpose: File operations with path parameters
  • Path Parameters: filename - the file to operate on

Custom Endpoints​

You can define custom endpoints using the endpoints configuration:

endpoints=[
    '/api/v1/data',
    '/api/v1/files/{filename}',
    '/custom/endpoint/{id}',
]

HTTP Methods​

Supported Methods​

  • GET: Retrieve data and files
  • POST: Send data
  • PUT: Update/replace data
  • PATCH: Partial data updates
  • DELETE: Remove data

Method Configuration​

methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE']

Data Handling​

JSON Payload Processing​

The REST Connect filter converts JSON payloads into Frame objects:

# Incoming JSON
{
    "detections": [
        {"class": "person", "confidence": 0.95, "bbox": [100, 100, 200, 200]}
    ],
    "timestamp": "2024-01-15T10:30:00Z",
    "camera_id": "cam1"
}

# Converted to Frame object
frame = Frame()
frame.data = {
    "detections": [...],
    "timestamp": "...",
    "camera_id": "cam1"
}

Local File Serving​

Resource Path Configuration​

The resource_path parameter specifies where local files are served from:

resource_path='/path/to/local/files'

File Access​

Files can be accessed via the /files/{filename} endpoint:

GET /files/image1.jpg
GET /files/data/sensor_readings.json
GET /files/videos/sample.mp4

File Operations​

Different HTTP methods provide different file operations:

  • GET: Download/read file
  • POST: Create file
  • PUT: Replace file content
  • PATCH: Update file metadata
  • DELETE: Remove file

Usage Examples​

Example 1: Basic REST API Server​

Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
    )),
    (ObjectDetection, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
    )),
    (ImageOut, dict(
        sources='tcp://localhost:5552',
        outputs='file:///output/detected_{frame_number}.jpg',
    )),
])

Behavior: Accepts HTTP requests, processes data through object detection, and saves results.

Example 2: Data Processing Pipeline​

Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
        resource_path='/data',
        endpoints=['/data', '/files/{filename}'],
    )),
    (DataProcessor, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
    )),
    (OutputFilter, dict(
        sources='tcp://localhost:5552',
        outputs='file:///processed/data.json',
    )),
])

Behavior: Accepts data requests, processes them, and saves processed results.

Example 3: JSON Data Ingestion​

Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
        methods=['POST', 'PUT'],
        endpoints=['/data', '/api/v1/sensors'],
    )),
    (DataProcessor, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
    )),
    (Recorder, dict(
        sources='tcp://localhost:5552',
        outputs='file:///logs/sensor_data.jsonl',
        rules=['+'],
    )),
])

Behavior: Accepts JSON sensor data, processes it, and logs results.

Example 4: Multi-Endpoint API​

Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
        resource_path='/resources',
        endpoints=[
            '/api/v1/data',
            '/api/v1/files/{filename}',
            '/health',
            '/status',
        ],
    )),
    (Router, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
        routing_rules={
            'data': 'tcp://*:5554',
            'files': 'tcp://*:5555',
        },
    )),
])

Behavior: Provides multiple API endpoints with different processing paths.

Example 5: Webhook Integration​

Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
        endpoints=['/webhook/{source}'],
    )),
    (WebhookProcessor, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
    )),
    (NotificationSender, dict(
        sources='tcp://localhost:5552',
        outputs='tcp://*:5554',
    )),
])

Behavior: Accepts webhook calls from external services and processes them.

Example 6: File Management API​

Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
        resource_path='/files',
        methods=['GET', 'POST', 'PUT', 'DELETE'],
        endpoints=['/files/{filename}'],
    )),
    (FileProcessor, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
    )),
    (FileManager, dict(
        sources='tcp://localhost:5552',
        outputs='tcp://*:5554',
    )),
])

Behavior: Provides complete file management API with CRUD operations.

API Usage Examples​

Sending JSON Data​

Using curl​

# Send JSON data
curl -X POST http://localhost:8080/data \
  -H "Content-Type: application/json" \
  -d '{
    "detections": [
      {"class": "person", "confidence": 0.95, "bbox": [100, 100, 200, 200]}
    ],
    "timestamp": "2024-01-15T10:30:00Z",
    "camera_id": "cam1"
  }'

Using Python requests​

import requests
import json

# Send JSON data
data = {
    "detections": [
        {"class": "person", "confidence": 0.95, "bbox": [100, 100, 200, 200]}
    ],
    "timestamp": "2024-01-15T10:30:00Z",
    "camera_id": "cam1"
}
response = requests.post('http://localhost:8080/data', json=data)
print(response.json())

File Operations​

Download file​

# Download a file
curl -X GET http://localhost:8080/files/data1.json -o downloaded_data.json

Create file​

# Create a new file
curl -X POST http://localhost:8080/files/new_data.json \
  -H "Content-Type: application/json" \
  -d '{"data": "content"}'

Delete file​

# Delete a file
curl -X DELETE http://localhost:8080/files/old_data.json

Error Handling​

HTTP Status Codes​

  • 200 OK: Successful operation
  • 400 Bad Request: Invalid request data
  • 404 Not Found: Endpoint or file not found
  • 405 Method Not Allowed: Unsupported HTTP method
  • 413 Payload Too Large: File too large
  • 500 Internal Server Error: Server processing error

Error Response Format​

{
    "error": "Error description",
    "detail": "Detailed error information",
    "status_code": 400
}

Common Error Scenarios​

  • Invalid JSON: Malformed JSON payload
  • Missing Files: File not found in resource path
  • Permission Errors: File access permission issues
  • Large Data: Data exceeding size limits
  • Unsupported Methods: HTTP method not allowed
  • Missing Parameters: Required parameters not provided

Security Considerations​

Input Validation​

  • Data Type Validation: Restrict allowed data types
  • Size Limits: Limit data payload sizes
  • Path Traversal: Prevent directory traversal attacks
  • Content Validation: Validate received content

Access Control​

  • Authentication: Implement authentication mechanisms
  • Authorization: Control access to endpoints
  • Rate Limiting: Prevent abuse and DoS attacks
  • CORS: Configure cross-origin resource sharing

Security Best Practices​

# Example security configuration
Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='127.0.0.1',  # Bind to localhost only
        port=8080,
        resource_path='/secure/data',  # Restricted path
        # Add authentication middleware
        # Add rate limiting
        # Add input validation
    )),
])

Performance Considerations​

Request Handling​

  • Async Processing: FastAPI provides async request handling
  • Connection Pooling: Efficient HTTP connection management
  • Memory Usage: Monitor memory usage for large data payloads
  • Processing Time: Consider downstream processing time

File Operations​

  • Streaming: Stream large files instead of loading into memory
  • Caching: Cache frequently accessed files
  • Compression: Use compression for large responses
  • CDN Integration: Use CDN for static file serving

Monitoring​

  • Request Metrics: Monitor request rates and response times
  • Error Rates: Track HTTP error rates
  • Data Operations: Monitor data processing rates
  • Resource Usage: Track CPU and memory usage

Troubleshooting​

Common Issues​

Server Not Starting​

  1. Check port availability
  2. Verify host binding
  3. Check firewall settings
  4. Validate configuration parameters

Data Processing Failures​

  1. Check data size limits
  2. Verify data permissions
  3. Ensure sufficient disk space
  4. Validate data format support

JSON Processing Errors​

  1. Validate JSON syntax
  2. Check content-type headers
  3. Verify data structure
  4. Handle encoding issues

Path Parameter Issues​

  1. Validate endpoint patterns
  2. Check parameter extraction
  3. Verify file path construction
  4. Handle special characters

Debug Configuration​

import logging
logging.basicConfig(level=logging.DEBUG)

# Enable REST Connect filter debugging
export DEBUG_REST=true
export LOG_LEVEL=DEBUG

Debug Information​

  • Request Logging: Log all incoming requests
  • Response Logging: Log response details
  • Data Operations: Log data access operations
  • Error Details: Detailed error information

Advanced Usage​

Custom Middleware​

# Add custom middleware for authentication, logging, etc.
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Custom Endpoints​

# Define additional custom endpoints
@rest_app.post("/custom/endpoint")
async def custom_endpoint(data: dict):
    # Custom processing logic
    return {"status": "processed", "data": data}

Integration with External Systems​

# Integrate with external APIs and services
Filter.run_multi([
    # ... other filters above
    (Rest, dict(
        outputs='tcp://*:5550',
        host='0.0.0.0',
        port=8080,
    )),
    (ExternalAPIClient, dict(
        sources='tcp://localhost:5550',
        outputs='tcp://*:5552',
        api_url='https://external-api.com',
        api_key='your-api-key',
    )),
    (ResponseFormatter, dict(
        sources='tcp://localhost:5552',
        outputs='tcp://*:5554',
    )),
])

API Reference​

RestConfig​

class RestConfig(FilterConfig):
    outputs: str | list[str] | list[tuple[str, dict[str, Any]]]
    host: str
    port: int
    resource_path: str | None
    methods: list[str] | None
    endpoints: list[str] | None

Rest​

class Rest(Filter):
    FILTER_TYPE = 'Input'
    
    @classmethod
    def normalize_config(cls, config)
    def init(self, config)
    def setup(self, config)
    def shutdown(self)
    def process(self, frames)
    @staticmethod
    def load_file(filename: str, resource_path: str | None) -> bytes
    @staticmethod
    def parse_multipart_data(content: bytes, content_type: str) -> tuple[dict[str, Any], dict[str, bytes]]
    @staticmethod
    def parse_json_data(content: bytes) -> dict[str, Any]
    @staticmethod
    def parse_form_data(content: bytes, content_type: str) -> dict[str, Any]

Environment Variables​

  • DEBUG_REST: Enable debug logging
  • FILTER_OUTPUTS: Output destinations
  • FILTER_HOST: Server host address
  • FILTER_PORT: Server port number
  • FILTER_RESOURCE_PATH: Local file resource path
  • FILTER_METHODS: Allowed HTTP methods
  • FILTER_ENDPOINTS: Custom endpoint patterns