zerodb-prefect

ZeroDB event sensor and result storage for Prefect.

Why use this?

Installation

pip install zerodb-prefect

Quick Start

Event Sensor

from zerodb_prefect import ZeroDBSensor

sensor = ZeroDBSensor(event_type='zerodb.vector.stored')

@sensor.on_event
def process_vector(event):
    print(f"New vector: {event.data}")
    return {'processed': event.data['vector_id']}

# Start polling for events
sensor.start()

Result Storage

from zerodb_prefect import ZeroDBBlock

block = ZeroDBBlock()

# Write results
block.write_path('/results/my-flow/run-1', {
    'status': 'complete',
    'rows_processed': 42,
})

# Read results
data = block.read_path('/results/my-flow/run-1')
print(data)  # {'status': 'complete', 'rows_processed': 42}

Credentials

from zerodb_prefect import ZeroDBCredentials

creds = ZeroDBCredentials(api_key='zdb_...', project_id='proj_...')
client = creds.get_client()  # Authenticated requests.Session

API Reference

ZeroDBSensor(event_type, **kwargs)

Poll ZeroDB events and trigger handlers.

Methods:

ZeroDBBlock(**kwargs)

Store and retrieve Prefect flow results.

ZeroDBCredentials(**kwargs)

Manage ZeroDB API credentials.

Configuration

Environment Variables

export ZERODB_API_KEY="your-api-key"
export ZERODB_PROJECT_ID="your-project-id"
# Optional
export ZERODB_BASE_URL="https://api.ainative.studio"

Auto-Provisioning

If no credentials are found, zerodb-prefect automatically creates a free ZeroDB project on first use. Credentials are saved to ~/.zerodb/config.json.

Use Cases

ML Pipeline Results

from zerodb_prefect import ZeroDBBlock

block = ZeroDBBlock()

# After training
block.write_path('/models/v2/metrics', {
    'accuracy': 0.94,
    'f1': 0.91,
    'training_time': 3600,
})

# In evaluation flow
metrics = block.read_path('/models/v2/metrics')

Event-Driven ETL

from zerodb_prefect import ZeroDBSensor, ZeroDBBlock

sensor = ZeroDBSensor(event_type='zerodb.file.uploaded')
block = ZeroDBBlock()

@sensor.on_event
def etl_pipeline(event):
    file_path = event.data.get('path')
    block.write_path(f'/etl/processed/{file_path}', {
        'source': file_path,
        'status': 'processed',
    })

sensor.start()

Powered by ZeroDB + AINative

Free database for AI agents. Auto-provisions in 200ms. No signup required.

Get started | Documentation | ZeroDB | GitHub

License

MIT