corva-sdk 0.0.7

SDK for interacting with Corva

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

Contents

• Requirements
• Installation
• App types
• Event
• Api
• Cache
• 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 a 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: it is recommended to use type hints like in examples below, so that editors and tools can give you better support.

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 two optional parameters:

• filter_by_timestamp - enable to clear event from data with previously processed timestamp;
• filter_by_depth - enable to clear event from data with previously processed 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)

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

The apps might need to communicate with Corva API and Corva Data API. The sdk provides an Api class - a thin wrapper around requests library that handles Corva authorization, adds timeouts and retries to request. 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 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)
   

Contributing

Set up the project

$ cd ~/YOUR_PATH/python-sdk
$ python3 -m venv venv
$ source venv/bin/activate
$ pip install -r requirements.txt

Run tests

$ venv/bin/python3 -m pytest tests

Run code linter

$ venv/bin/python3 -m flake8