ourri

Official Python SDK for Ourri — multi-provider data extraction and orchestration.

Install

pip install ourri

Requires Python 3.10+ and httpx.

Quick Start

from ourri import OurriClient

client = OurriClient(api_key="your-api-key")

# Execute a schema
execution = client.execute("schema-version-id", input_data={"url": "https://example.com"})

# Wait for results
results = client.wait_for_completion(
    execution["id"],
    poll_interval=2.0,
    on_progress=lambda e: print(f"Status: {e.get('status')}"),
)
print(results)

Use as a context manager to ensure the HTTP client is closed:

with OurriClient(api_key="your-api-key") as client:
    schemas = client.list_schemas()

Async Client

For asyncio applications, use AsyncOurriClient — identical API surface with async/await:

import asyncio
from ourri import AsyncOurriClient

async def main():
    async with AsyncOurriClient(api_key="your-api-key") as client:
        execution = await client.execute("schema-version-id")
        results = await client.wait_for_completion(execution["id"])

        # Streaming returns AsyncIterator[str]
        async for line in client.stream_execution_logs("exec-id"):
            print(line)

asyncio.run(main())

Configuration

from ourri import OurriClient, OurriClientConfig

client = OurriClient(
    api_key="your-api-key",
    base_url="https://api.ourri.com",  # or http://localhost:8080
    timeout=30.0,                      # seconds
    max_retries=3,                     # retries on 429 / 5xx / network errors
    retry_delay=1.0,                   # initial backoff (exponential with jitter)
    debug=False,
)

# Or pass a config object
config = OurriClientConfig(api_key="...", base_url="http://localhost:8080")
client = OurriClient(config=config)

API Reference

Scraping & Extraction

result = client.scrape({"url": "https://example.com", "format": "json"})
extracted = client.extract({"url": "https://example.com/product", "schema": {"name": "string", "price": "number"}})
results = client.search({"query": "web scraping", "limit": 10})

Schemas

from ourri import SchemaInput, SchemaField

schema = client.create_schema(SchemaInput(
    name="Products",
    fields=[SchemaField(name="title", type="string", required=True)],
))
schemas = client.list_schemas()
one = client.get_schema("schema-id")
client.update_schema("schema-id", {"description": "Updated"})
client.delete_schema("schema-id")
client.duplicate_schema("schema-id")
client.toggle_schema_favorite("schema-id")

# Examples & validation
examples = client.get_schema_examples("schema-id")
fields = client.get_example_fields("schema-id")
stats = client.get_example_statistics("schema-id")
client.validate_schema_with_examples("schema-id", {...})
client.preview_aggregation("schema-id", {...})

Execution

from ourri import ExecutionOptions

# Preflight check
preflight = client.preflight_execution({
    "schema_version_id": "sv-id",
    "input": {"url": "https://example.com"},
})

# Execute
execution = client.execute("schema-version-id", ExecutionOptions(
    input_data={"url": "https://example.com"},
    output_format="json",
))

# Wait for completion
results = client.wait_for_completion(execution["id"], timeout=300.0)

# Lifecycle
client.cancel_execution("exec-id")
client.retry_execution("exec-id")
client.delete_execution("exec-id")

# Sub-resources
status = client.get_execution_status("exec-id")
results = client.get_execution_results("exec-id")
costs = client.get_execution_costs("exec-id")
attempts = client.get_execution_attempts("exec-id")
deliveries = client.get_execution_deliveries("exec-id")
explain = client.get_execution_routing_explain("exec-id")
logs = client.get_execution_logs("exec-id")
timeline = client.get_execution_timeline("exec-id")
steps = client.get_execution_steps("exec-id")

# List with filters
from ourri import ExecutionFilter, PaginationParams
executions = client.list_executions(
    filter=ExecutionFilter(status=["completed", "failed"]),
    pagination=PaginationParams(page=1, per_page=20),
)

Execution Streaming

Stream real-time data as Iterator[str]:

for line in client.stream_execution_logs("exec-id"):
    print(line)

for line in client.stream_execution_logs_sse("exec-id"):
    print(line)

for line in client.stream_execution_steps("exec-id"):
    print(line)

for line in client.stream_execution_timeline("exec-id"):
    print(line)

Routing

route = client.route({
    "entity_type": "ecommerce.product",
    "strategy": "balanced",
    "volume": 1000,
})

fallback = client.route_with_fallback({...})
graph = client.route_with_graph({...})
alts = client.route_with_alternatives({..., "k": 5})
multi = client.route_multi_step({"steps": [...], "strategy": "balanced"})

# Intelligent routing
result = client.intelligent_route({"query": "Get LinkedIn profiles"})
strategies = client.get_intelligent_strategies()

Provider Keys (BYOK)

keys = client.list_provider_keys()
key = client.create_provider_key("apify", "apify_api_...")
client.update_provider_key("key-id", {"api_key": "new-key"})
client.validate_provider_key("key-id")
client.delete_provider_key("key-id")

Schedules

from ourri import ScheduleCreateRequest

schedules = client.list_schedules()
schedule = client.create_schedule(ScheduleCreateRequest(
    schema_version_id="sv-id",
    name="Daily check",
    interval_seconds=86400,
    entity_key_field="url",
    budget_monthly_limit=50.0,
))
client.pause_schedule("sched-id")
client.resume_schedule("sched-id")
client.delete_schedule("sched-id")

snapshots = client.get_schedule_snapshots("sched-id")
alerts = client.get_schedule_alerts("sched-id")
audit = client.get_schedule_provider_audit("sched-id")

Computations

comps = client.list_computations("schema-id")
comp = client.get_computation("comp-id")
analysis = client.analyze_computation_intent({"user_request": "Average price?", "schema_version_id": "sv-id"})
created = client.create_computation({...})
compiled = client.compile_computation("comp-id")
dry = client.dry_run_computation("comp-id", {"price": 29.99})
results = client.execute_computations("schema-id", {"price": 29.99})
client.delete_computation("comp-id")

stats = client.get_computation_statistics()
cache = client.get_computation_cache_stats()
client.clear_computation_cache()

Cost & Analytics

estimate = client.estimate_cost("sv-id")
summary = client.estimate_cost_summary("sv-id", estimated_executions_per_month=100)
pricing = client.calculate_pricing({...})

budget = client.get_user_budget()
spending = client.get_user_spending()
reservations = client.get_budget_reservations()

exec_stats = client.get_execution_stats("2024-01-01", "2024-12-31")
costs = client.get_analytics_costs()
timeseries = client.get_analytics_time_series()
hourly = client.get_analytics_hourly()
schema_perf = client.get_schema_performance()
customer_costs = client.get_customer_cost_breakdown()

# Local cost tracking
print(f"Total: ${client.get_total_cost():.2f}")
print(client.get_cost_summary())
client.reset_cost_tracking()

Entity Timeline

history = client.get_entity_history("entity-id", limit=50)
snapshot = client.get_entity_at_point_in_time("entity-id", "2024-01-15T10:30:00Z")
latest = client.get_latest_entity_version("entity-id")
freshness = client.get_entity_freshness("entity-id")
entities = client.get_entities_by_time_range("2024-01-01T00:00:00Z", "2024-01-31T23:59:59Z", "ecommerce.product")
daily = client.get_entity_daily_stats("ecommerce.product", days=30)
queried = client.query_entities({"entity_type": "ecommerce.product", "filters": {...}})

Datasets

datasets = client.list_datasets()
dataset = client.get_dataset_type("ecommerce.product")
exported = client.export_dataset("ecommerce.product", {...})
queried = client.query_dataset({...})

Webhooks

from ourri import WebhookInput

webhook = client.create_webhook(WebhookInput(
    url="https://myapp.com/webhooks/ourri",
    events=["execution.completed", "execution.failed"],
    secret="whsec_...",
))
webhooks = client.list_webhooks()
client.delete_webhook("webhook-id")

Export

from ourri import ExportOptions

job = client.export_execution_history(
    ExportOptions(date_range={"start": "2024-01-01", "end": "2024-03-31"}, format="parquet"),
    wait=True,
)
status = client.get_export_job("job-id")
client.download_export("job-id", "./output.parquet")

Observations

client.create_observation({"provider": "apify", "endpoint": "/v2/acts/run", "status_code": 200})
client.create_observations_batch([...])

Catalog

categories = client.list_categories()
category = client.get_category("ecommerce")
use_cases = client.list_use_cases()
use_case = client.get_use_case("price-monitoring")
canonical = client.list_canonical_schemas()
stats = client.get_system_stats()
cost_comp = client.get_cost_comparison()

API Key Management

keys = client.list_api_keys()
key = client.create_api_key({"name": "Production", "scopes": ["execute"]})
client.revoke_api_key("key-id")

perms = client.list_api_key_permissions("key-id")
client.add_api_key_permission("key-id", {"permission": "schemas:write"})
client.remove_api_key_permission("key-id", "permission-id")
limits = client.get_api_key_rate_limits("key-id")
client.update_api_key_rate_limits("key-id", {"requests_per_minute": 100})
settings = client.get_api_key_settings("key-id")
client.update_api_key_settings("key-id", {...})

Dashboard & User

metrics = client.get_dashboard_metrics()
activity = client.get_dashboard_activity()
limits = client.get_user_limits()
key_info = client.get_api_key_info()

Admin Analytics

Requires admin role JWT:

health = client.get_admin_overview_health()
timeline = client.get_admin_request_timeline()
heatmap = client.get_admin_provider_heatmap()
perf = client.get_admin_provider_performance()
dedup = client.get_admin_deduplication()
cost_trends = client.get_admin_cost_trends()
cost_entity = client.get_admin_cost_by_entity_type()
cache_save = client.get_admin_cache_savings()
graph_net = client.get_admin_graph_network()
graph_stats = client.get_admin_graph_stats()
test_sum = client.get_admin_tests_summary()
test_time = client.get_admin_tests_timeline()
test_det = client.get_admin_tests_details()
cost_break = client.get_cost_breakdown("2024-01-01", "2024-12-31")

Error Handling

All errors extend OurriError, including OurriTimeoutError and OurriNetworkError. A single except OurriError catches every SDK error.

from ourri import (
    OurriError,
    OurriAuthenticationError,
    OurriRateLimitError,
    OurriValidationError,
    OurriTimeoutError,
    OurriNetworkError,
)

try:
    result = client.execute("sv-id")
except OurriAuthenticationError as e:
    # 401/403
    print(f"Auth failed: {e.message}")
except OurriRateLimitError as e:
    # 429
    print(f"Rate limited, retry after {e.retry_after}s")
except OurriValidationError as e:
    # 400/422
    print(f"Invalid param: {e.param} — {e.message}")
except OurriTimeoutError:
    print("Request timed out")
except OurriNetworkError:
    print("Connection failed")
except OurriError as e:
    # Any other API error
    print(f"Error {e.status_code}: {e.message} (request_id={e.request_id})")

Automatic Retries

The SDK retries on 429 (rate limit), 5xx (server errors), timeout, and connection errors with exponential backoff + jitter. Override per request:

from ourri import RequestOptions

result = client.scrape(
    {"url": "https://example.com"},
    options=RequestOptions(max_retries=1, timeout=10.0),
)

Observe Mode

Automatically capture provider API calls:

from ourri.observe import observe

teardown = observe(api_key="your-api-key")
# All requests/httpx calls to supported providers are now observed

Gateway Mode

Route provider calls through Ourri with stored credentials:

from ourri.observe.gateway import rewrite_for_gateway, GatewayOptions

result = rewrite_for_gateway(
    "https://api.apify.com/v2/acts/test/run-sync",
    GatewayOptions(api_key="your-api-key"),
)

Response Sanitization

The SDK automatically strips internal routing details (provider IDs, scraper names, routing scores) from all responses. Your application code never sees implementation internals.

Development

pip install -e ".[dev]"
pytest tests/
ruff check ourri/
ruff format ourri/

License

MIT