Corva Python SDK

raw JSON →
2.1.1 verified Fri May 01 auth: no python

Python SDK for building Corva DevCenter apps. Provides event-driven app framework, API client, caching, logging, and data transformations for oil and gas data applications. Current version 2.1.1 (requires Python >=3.10, <4.0). Release cadence: irregular, with recent major bumps from v1 to v2.

pip install corva-sdk
error ModuleNotFoundError: No module named 'corva_sdk'
cause Trying to import corva_sdk instead of corva.
fix
Use 'import corva' or 'from corva import ...' (the package installed as corva-sdk is imported as corva).
error ImportError: cannot import name 'ScheduledDataTimeEvent' from 'corva.models'
cause Using deprecated import path from corva.models.
fix
Change to 'from corva import ScheduledDataTimeEvent'.
error TypeError: 'NoneType' object is not subscriptable when accessing event.asset_id
cause Event can be None if app type mismatch (e.g., using ScheduledDataTimeEvent on a real-time event).
fix
Ensure your app is configured with the correct event type in the DevCenter. Add a check: if event is None: return.
breaking Version 2.0+ dropped support for Python <3.10. Upgrade your environment if still on Python 3.9.
fix Use Python 3.10+ or pin corva-sdk<2.0.0.
breaking In v2, the import path 'from corva_sdk import ...' is invalid. All public APIs are under the 'corva' top-level package.
fix Change imports to 'from corva import ...'.
deprecated The 'corva.models' module is deprecated in v2. Event and model classes should be imported directly from 'corva'.
fix Use 'from corva import ScheduledDataTimeEvent' instead of 'from corva.models import ScheduledDataTimeEvent'.
gotcha The SDK uses synchronous blocking I/O by default. Asynchronous handlers are not supported natively; use threads or run in separate processes.
fix Design your handlers as synchronous functions. For concurrency, consider using corva's built-in threading support via 'concurrent' parameter if available.
gotcha The Cache object does not support TTL expiration automatically. Keys persist indefinitely unless explicitly deleted.
fix Manually manage cache keys using cache.delete() or implement a custom eviction strategy.

Basic Corva app listening for scheduled data time events.

import os
from corva import App, ScheduledDataTimeEvent, Cache, ApiClient

app = App()

@app.on_event(ScheduledDataTimeEvent)
def handler(event: ScheduledDataTimeEvent, cache: Cache, api: ApiClient):
    data = {"message": "Hello from Corva SDK", "asset_id": event.asset_id}
    return data

if __name__ == "__main__":
    app.run()