Skip to main content

SDK for interacting with Corva

Project description

Corva python-sdk is a framework for building Corva DevCenter apps.

Contents

Requirements

Python 3.8+

Installation

$ pip install corva-sdk

App types

There are three app types that you can build:

  1. stream - works with real-time data
  2. scheduled - works with data at defined schedules/intervals (e.g. once an hour)
  3. task - works with data on-demand

Note: Use type hints like those in the examples below for better support from editors and tools.

Stream

Stream apps can be of two types - time or depth based.

Stream time app
from typing import List

from corva import Api, Cache, Corva, StreamTimeEvent


def stream_time_app(event: StreamTimeEvent, api: Api, cache: Cache):
    # get some data from the api
    drillstrings = api.get_dataset(
        provider='corva',
        dataset='data.drillstring',
        query={
            'asset_id': event.asset_id,
        },
        sort={'timestamp': 1},
        limit=1,
    )  # type: List[dict]

    # do some calculations/modifications to the data
    for drillstring in drillstrings:
        drillstring['id'] = drillstring.pop('_id')

    # save the data to private collection
    api.post(path='api/v1/data/my_provider/my_collection/', data=drillstrings)


def lambda_handler(event, context):
    Corva(context).stream(stream_time_app, event)
Stream depth app
from typing import List

from corva import Api, Cache, Corva, StreamDepthEvent


def stream_depth_app(event: StreamDepthEvent, api: Api, cache: Cache):
    # get some data from the api
    drillstrings = api.get_dataset(
        provider='corva',
        dataset='data.drillstring',
        query={
            'asset_id': event.asset_id,
        },
        sort={'timestamp': 1},
        limit=1,
    )  # type: List[dict]

    # do some calculations/modifications to the data
    for drillstring in drillstrings:
        drillstring['id'] = drillstring.pop('_id')

    # save the data to private collection
    api.post(path='api/v1/data/my_provider/my_collection/', data=drillstrings)


def lambda_handler(event, context):
    Corva(context).stream(stream_depth_app, event)

Scheduled

from corva import Api, Cache, Corva, ScheduledEvent


def scheduled_app(event: ScheduledEvent, api: Api, cache: Cache):
    print('Hello, World!')


def lambda_handler(event, context):
    Corva(context).scheduled(scheduled_app, event)

Task

from corva import Api, Corva, TaskEvent


def task_app(event: TaskEvent, api: Api):
    print('Hello, World!')


def lambda_handler(event, context):
    Corva(context).task(task_app, event)

Event

An event is an object that contains data for an app function to process. event instance is inserted automatically as a first parameter to each app type. There are different event types for every app type: StreamTimeEvent, StreamDepthEvent, ScheduledEvent and TaskEvent.

Api

Apps might need to communicate with the Corva Platform API and Corva Data API. This SDK provides an Api class, which wraps the Python requests library and adds automatic authorization, convenient URL usage and reasonable timeouts to API requests. Api instance is inserted automatically as a second parameter to each app type. Api supports following HTTP methods: GET, POST, PATCH, PUT and DELETE.

Examples:

from corva import Api, Corva


def my_app(event, api: Api, cache):
    # Corva API calls
    api.get('/v2/pads', params={'param': 'val'})
    api.post('/v2/pads', data={'key': 'val'})
    api.patch('/v2/pads/123', data={'key': 'val'})
    api.delete('/v2/pads/123')

    # Corva Data API calls
    api.get('/api/v1/data/provider/dataset/', params={'param': 'val'})
    api.post('/api/v1/data/provider/dataset/', data={'key': 'val'})
    api.put('/api/v1/data/provider/dataset/', data={'key': 'val'})
    api.delete('/api/v1/data/provider/dataset/')


def lambda_handler(event, context):
    corva = Corva(context)
    corva.scheduled(my_app, event)

Cache

Apps might need to share some data between runs. The sdk provides a Cache class, that allows you to store, load and do other operations with data. Cache instance is inserted automatically as a third parameter to stream and scheduled apps.
Note: task apps don't get a Cache parameter as they aren't meant to store data between invokes.

Cache uses a dict-like database, so the data is stored as key:value pairs. key should be of str type, and value can have any of the following types: str, int, float and bytes.

Examples:

  1. Store and load:

    from corva import Cache, Corva
    
    
    def my_app(event, api, cache: Cache):
       cache.store(key='key', value='val')
       # cache: {'key': 'val'}
    
       cache.load(key='key')  # returns 'val'
    
    
    def lambda_handler(event, context):
       corva = Corva(context)
       corva.scheduled(my_app, event)
    
  2. Store and load multiple:

    from corva import Cache, Corva
    
    
    def my_app(event, api, cache: Cache):
       cache.store(mapping={'key1': 'val1', 'key2': 'val2'})
       # cache: {'key1': 'val1', 'key2': 'val2'}
    
       cache.load_all()  # returns {'key1': 'val1', 'key2': 'val2'}
    
    
    def lambda_handler(event, context):
       corva = Corva(context)
       corva.scheduled(my_app, event)
    
  3. Delete and delete all:

    from corva import Cache, Corva
    
    
    def my_app(event, api, cache: Cache):
       cache.store(mapping={'key1': 'val1', 'key2': 'val2', 'key3': 'val3'})
       # cache: {'key1': 'val1', 'key2': 'val2', 'key3': 'val3'}
    
       cache.delete(keys=['key1'])
       # cache: {'key2': 'val2', 'key3': 'val3'}
    
       cache.delete_all()
       # cache: {}
    
    
    def lambda_handler(event, context):
       corva = Corva(context)
       corva.scheduled(my_app, event)
    
  4. Expiry, ttl and exists. Note: by default Cache sets an expiry to 60 days.

    import time
    from corva import Cache, Corva
    
    
    def my_app(event, api, cache: Cache):
       cache.store(key='key', value='val', expiry=60)  # 60 seconds
       # cache: {'key': 'val'}
    
       cache.ttl()  # 60 seconds
       cache.pttl() # 60000 milliseconds
       cache.exists()  # True
    
       time.sleep(60)  # wait 60 seconds
       # cache: {}
    
       cache.exists()  # False
       cache.ttl()  # -2: doesn't exist
       cache.pttl() # -2: doesn't exist
    
    
    def lambda_handler(event, context):
       corva = Corva(context)
       corva.scheduled(my_app, event)
    

Testing

Testing Corva applications is easy and enjoyable.

The SDK provides convenient tools for testing through pytest plugin. Write your tests using pytest to get the access to the plugin. To install the library run pip install pytest.

Stream time app example test

from corva import Corva, StreamTimeEvent, StreamTimeRecord


def stream_app(event, api, cache):
    return 'Hello, World!'


def lambda_handler(event, context):
    return Corva(context).stream(fn=stream_app, event=event)


def test_stream_time_app(app_runner):
    event = StreamTimeEvent(
        asset_id=0, company_id=0, records=[StreamTimeRecord(timestamp=0)]
    )

    result = app_runner(fn=lambda_handler, event=event)

    assert result == 'Hello, World!'

Stream depth app example test

from corva import Corva, StreamDepthEvent, StreamDepthRecord


def stream_app(event, api, cache):
    return 'Hello, World!'


def lambda_handler(event, context):
    return Corva(context).stream(fn=stream_app, event=event)


def test_stream_depth_app(app_runner):
    event = StreamDepthEvent(
        asset_id=0, company_id=0, records=[StreamDepthRecord(measured_depth=0)]
    )

    result = app_runner(fn=lambda_handler, event=event)

    assert result == 'Hello, World!'

Scheduled app example test

from corva import Corva, ScheduledEvent


def scheduled_app(event, api, cache):
    return 'Hello, World!'


def lambda_handler(event, context):
    return Corva(context).scheduled(fn=scheduled_app, event=event)


def test_scheduled_app(app_runner):
    event = ScheduledEvent(asset_id=0, start_time=0, end_time=0)

    result = app_runner(fn=lambda_handler, event=event)

    assert result == 'Hello, World!'

Task app example test

from corva import Corva, TaskEvent


def task_app(event, api):
    return 'Hello, World!'


def lambda_handler(event, context):
    return Corva(context).task(fn=task_app, event=event)


def test_task_app(app_runner):
    event = TaskEvent(asset_id=0, company_id=0)

    result = app_runner(fn=lambda_handler, event=event)

    assert result == 'Hello, World!'

Contributing

Set up the project

$ cd ~/YOUR_PATH/python-sdk
$ python -m venv venv
$ source venv/bin/activate
(venv) $ pip install -e .[dev]

Run tests

(venv) $ pytest

Run code linter

(venv) $ flake8

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

corva-sdk-0.0.16.tar.gz (19.1 kB view hashes)

Uploaded Source

Built Distribution

corva_sdk-0.0.16-py3-none-any.whl (21.7 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page