Implementation:Evidentlyai Evidently Legacy Collector App

Knowledge Sources	Evidentlyai_Evidently
Domains	Collector, REST_API, Data_Monitoring, Legacy
Last Updated	2026-02-14 12:00 GMT

Overview

legacy/collector/app.py implements a Litestar-based REST API service that collects data from production systems, buffers it, periodically generates Evidently reports/test suites, and uploads the resulting snapshots to an Evidently workspace.

Description

This module provides the full lifecycle of the legacy data collector service:

Route Handlers:

• POST /{id} (create_collector) -- Creates or updates a collector configuration. Persists to config file if autosave is enabled.
• GET /{id} (get_collector) -- Retrieves a collector configuration by ID.
• POST /{id}/reference (set_reference) -- Uploads reference data as a parquet file for a collector.
• POST /{id}/data (push_data) -- Pushes current/production data into the collector's buffer (async with lock).
• GET /{id}/logs (get_logs) -- Retrieves log events for a collector.

Background Processing:

• check_snapshots_factory -- Iterates all collectors and triggers snapshot creation and upload when conditions are met.
• create_snapshot -- Flushes the data buffer, runs the configured report or test suite, and stores the result.
• send_snapshot -- Uploads generated reports to the configured workspace (Evidently Cloud or self-hosted).

Application Factory:

• create_app -- Constructs the Litestar application with dependency injection, authentication middleware (token or no-security), periodic snapshot checking via a lifespan background task, and error handling.
• run -- Convenience function to start the server with uvicorn.

The service supports token-based authentication and integrates with Evidently's telemetry system.

Usage

Use the collector app to set up a real-time data monitoring pipeline. Deploy it as a service endpoint that receives production data, then automatically generates and uploads monitoring reports at configured intervals or row-count thresholds.

Code Reference

Source Location

• Repository: Evidentlyai_Evidently
• File: src/evidently/legacy/collector/app.py

Signature

@post("/{id:str}", sync_to_thread=True)
def create_collector(id, parsed_json, service, storage, service_config_path) -> CollectorConfig: ...

@get("/{id:str}")
async def get_collector(id, service) -> CollectorConfig: ...

@post("/{id:str}/reference", sync_to_thread=True)
def set_reference(id, parsed_json, service, service_config_path, service_workspace) -> Dict[str, str]: ...

@post("/{id:str}/data")
async def push_data(id, data, service, storage) -> Dict[str, str]: ...

@get("/{id:str}/logs")
async def get_logs(id, service, storage) -> List[LogEvent]: ...

async def check_snapshots_factory(service, storage) -> None: ...
async def create_snapshot(collector, storage) -> None: ...
async def send_snapshot(collector, storage) -> None: ...

def create_app(config_path: str = CONFIG_PATH, secret: Optional[str] = None, debug: bool = False) -> Litestar: ...
def run(host: str = "127.0.0.1", port: int = 8001, config_path: str = CONFIG_PATH, secret: Optional[str] = None): ...

Import

from evidently.legacy.collector.app import create_app, run

I/O Contract

Inputs

Name	Type	Required	Description
id	str	Yes	Unique identifier for the collector instance (URL path parameter).
parsed_json	CollectorConfig or Any	Yes	JSON body parsed into the appropriate type for each endpoint.
config_path	str	No (default "collector_config.json")	Path to the service configuration file.
secret	Optional[str]	No	Authentication token. If provided, enables token-based security.
host	str	No (default "127.0.0.1")	Host address for the uvicorn server.
port	int	No (default 8001)	Port number for the uvicorn server.

Outputs

Name	Type	Description
create_collector return	CollectorConfig	The created/updated collector configuration.
get_collector return	CollectorConfig	The requested collector configuration.
push_data return	Dict[str, str]	Empty dict on success.
get_logs return	List[LogEvent]	List of log events for the specified collector.
create_app return	Litestar	Configured Litestar ASGI application instance.

Usage Examples

from evidently.legacy.collector.app import create_app, run

# Create the app with default config
app = create_app(config_path="my_config.json", secret="my-secret-token")

# Or run directly
run(host="0.0.0.0", port=8001, config_path="my_config.json", secret="my-secret-token")

Related Pages

• Environment:Evidentlyai_Evidently_Python_Core_Environment