corva-sdk 0.0.15

SDK for interacting with Corva

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

Contents

• Requirements
• Installation
• App types
• Event
• Api
• Cache
• Testing
• Contributing

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

import json

from corva import Api, Cache, Corva, StreamEvent


def stream_app(event: StreamEvent, api: Api, cache: Cache):
    # get some data from api
    drillstrings = api.get(
        'api/v1/data/corva/data.drillstring/',
        params={
            'query': json.dumps({'asset_id': event.asset_id, }),
            'sort': json.dumps({'timestamp': 1}),
            'limit': 1}
    ).json()  # 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(
        f'api/v1/data/my_provider/my_collection/',
        data=drillstrings
    )


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

Corva.stream provides an optional parameter:

• filter_mode - set to timestamp or depth to clear event data with previously processed timestamp or measured_depth.

Scheduled

from corva import Api, Cache, Corva, ScheduledEvent


def scheduled_app(event: ScheduledEvent, api: Api, cache: Cache):
    pass


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

Task

from corva import Api, Corva, TaskEvent


def task_app(event: TaskEvent, api: Api):
    pass


def lambda_handler(event, context):
    corva = Corva(context)
    corva.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: StreamEvent, 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 example test

# lambda_function.py

from corva import Corva


def app(event, api, cache):
    print("Hello, World!")
    pass


def lambda_handler(event, context):
    corva = Corva(context)
    corva.stream(app, event)


# tests/test_app.py

# corva_context is provided by the SDK's pytest plugin. No imports required.
def test_app(corva_context):
    event = {"records": [{"asset_id": 0, "timestamp": 0}]}

    lambda_handler(event, corva_context)

Scheduled example test

# lambda_function.py

from corva import Corva


def app(event, api, cache):
    print("Hello, World!")
    pass


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


# tests/test_app.py

# corva_context is provided by the SDK's pytest plugin. No imports required.
def test_app(corva_context):
    event = {
        "schedule": 0,
        "interval": 0,
        "schedule_start": 0,
        "asset_id": 0,
    }

    lambda_handler(event, corva_context)

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