Skip to main content

Task App

What is a Task App?​

A task app is an app that gets invoked by an API Post request (https://api.corva.ai/v2/tasks) via a frontend app, another backend app with an API key or externally via the API Post url with an API Key.

The task app receives on demand data in the form of events from the body of the API Post request. It receives the event whenever the API Post request is sent.

The task app will sit dormant until it is invoked only when an API Post request is sent. At no other time will the task app be invoked.

Unlike a stream or scheduled app, task apps are NOT assigned to an asset stream.

Task apps execute in their own short-lived Lambdas on AWS. Task apps are best used for data processing, single API requests, or calculations that need to be done once and have the outputs saved to a dataset.

While a frontend app can invoke a task app via an API Post request, the task app does not return data to the frontend app. The task app can communicate processed data by posting the values in a dataset (https://data.corva.ai/api/v1/data/{provider}/{dataset}/) or posting to Corva's Data API subscription endpoint (https://data.corva.ai/api/v1/subscriptions/{provider}/{dataset}/{asset_id}/).

Dev Center task apps are built on AWS Lambda. To learn more about AWS Lambda functions please see Best practices for working with AWS Lambda functions.


When to build a Task App?​

Typical use cases for a task app include invoking initial API request for an external/non-Corva API. Invoking Back End app to Back End app in order to either abstract common tasks/functions, farm out computations to new lambdas, or update custom collections. Performing single calculations based on inputs from the UI.

  1. I require processing or calculations of input data from a frontend app, then posted to a dataset

  2. I require the task to run on a separate/new lambda (memory/size concerns)

  3. I require a scheduled, stream or another task app to invoke a task app to execute a function or process data

  4. I require this app to be reusable or invoked by several other apps

  5. I require my external data's well identifiers to be mapped to Corva's well identifiers, then posted to a dataset, a.k.a mapping data

  6. I require one-time, triggered execution

  7. I do not require the app to be attached to an asset stream

  8. I do not require the app to receive real-time data events from Corva

If you require real-time data at one second or 1 foot intervals (or the next available measurement), you should use a Stream App instead.


What is the difference between a Task App and a Scheduled or Stream App?​

A task app receives an event when it is invoked when an API Post is made to the app. A stream or scheduled app receives an event from a drilling rig, frac fleet, wireline unit or drill out unit.


Where to find the data to build a Task App?​

The Corva Dev Center Dataset Explorer is an easy way to to see a list of Corva datasets and the data stored in each dataset. An additional option is to utilize Corva's API tools located in the Dev Center Intro page and make a GET API request to your desired dataset.


How to build a Task App?​

The following examples will demonstrate the simple building blocks of a task application while utilizing all imported functionalities.

1. Task Event​

The use case for the example below is the app is being used as an external function that calculates the Net Present Value of a natural gas well, then POST the calculated value to a custom dataset named big-data-energy#example-task-app.


1.1 Follow the Getting Started Tutorial to install the prerequisites, create and copy app to your local machine.​


1.2. Open and review the README.md file.​

# Dev Center Python Task App

## Getting started

[Python SDK Documentation](https://corva-ai.github.io/python-sdk)

### 1. Install

You need to have `make` installed:
    
- Windows: [Windows Subsystem for Linux](https://docs.microsoft.com/en-us/windows/wsl/install)
- OS X: [Command Line Tools for Xcode](https://developer.apple.com/download/more/)
- Linux: refer to your distro-specific docs

To install dependencies run: 

`make`

### 2. Run Tests

`make test`

### 3. Create app zip

`make bundle`

Example of a README.md file


1.3. Update to latest Python SDK version in requirements.txt file​

corva-sdk==1.8.0
pytest==7.1.1

Example of a README.md file for a TaskEvent app

1.3.1. In the command line run​
pip install -U corva-sdk
1.3.2 Or within requirements.txt file set to latest sdk version e.g. corva-sdk==1.8.0 , then run​
pip install -r requirements.txt

1.4. Optional: Install make dependency​

1.4.1 Install make​
make install
make all

1.5 Make adjustments to the manifest.json file​

{
   "format":1,
   "license":{
      "type":"MIT",
      "url":"https://www.oandgexample.com/license/"
   },
   "developer":{
      "name":"O&G Company",
      "identifier":"oandgc",
      "authors":[
         
      ]
   },
   "application":{
      "type":"task",
      "key":"big-data-energy.example_task_app",
      "visibility":"private",
      "name":"Example Task App",
      "description":"This is the description of my app. You can do great things with it!",
      "summary":"More information about this app goes here",
      "category":"analytics",
      "website":"https://www.oandgexample.com/my-app/",
      "segments":[
         "drilling"
      ]
   },
   "settings":{
      "entrypoint":{
         "file":"lambda_function",
         "function":"lambda_handler"
      },
      "timeout":120,
      "memory":128,
      "environment":{
         "LOG_LEVEL":"DEBUG"
      },
      "runtime":"python3.8",
      "app":{}
   },
   "datasets":{
      "big-data-energy.example-task-app":{
         "permissions":[
            "read",
            "write"
         ]
      }
   }
}

Example of a manifest.json file for a TaskEvent app


1.5.1 Set read and/or write permissions to datasets in the manifest.json file​

Read and write permissions can only be granted to company datasets (non-Corva), e.g. big-data-energy#example-task-app. Read only permissions can be granted to Corva datasets where "provider":"corva", e.g. corva#wits. The example below shows how to grant read and write permissions to big-data-energy#example-task-app. 

   "datasets":{
      "big-data-energy.example-task-app":{
         "permissions":[
            "read",
            "write"
         ]
      }
   }

Example of dataset read/write permissions in the manifest.json file for a TaskEvent app


1.5.2 Optional: Set Python Log Levels in environment variables in the manifest.json file​

The default Log Level in the Corva SDK is Logger.info(). In order to change the default Log Level, you must set the log level in the environment variables within the manifest.json file. The example below shows how to change the default Log Level to Logger.debug().

"settings": {
    "entrypoint": {
      "file": "lambda_function",
      "function": "lambda_handler"
    },
    "timeout": 120,
    "memory": 128,
    "environment": {"LOG_LEVEL": "DEBUG" },
    "runtime": "python3.8",
    "app": {}
  }

Example of Python log levels environment variable setting in the manifest.json file for a TaskEvent app

Note: Please refer to Python's Logging Documentation to learn about the different log levels here Logging HOWTO and the Corva Logging Documentation.


1.5.3 Optional: Increase the application timeout time in the manifest.json file​

If your app is running longer than the desired timeout value, then you may need to make sure that the Lambda function is idempotent. If increasing the app timeout value is required, then you may need to increase the timeout value. The default timeout value for an application is 120 seconds. The maximum value for the application timeout is 900 seconds. The example below shows the application timeout increased to 240 seconds.

"settings": {
    "entrypoint": {
      "file": "lambda_function",
      "function": "lambda_handler"
    },
    "timeout": 240,
    "memory": 128,
    "environment": {"LOG_LEVEL": "DEBUG" },
    "runtime": "python3.8",
    "app": {}
  }

Example of app timeout setting in the manifest.json file for a TaskEvent app

Note: Please refer to AWS Lambda's Documentation to learn about lambda function best practices here Best practices for working with AWS Lambda functions.


1.5.4 Optional: Increase the application memory in the manifest.json file​

If your app is importing large code libraries, completing memory intensive tasks, or is running much slower than expected, then you may need to increase the memory setting. The default memory value and minimum value for an application is 128 MB. The maximum value for the application memory is 10,240 MB. The example below shows the application timeout increased by 128 MB increments to 640 MB.

"settings": {
    "entrypoint": {
      "file": "lambda_function",
      "function": "lambda_handler"
    },
    "timeout": 120,
    "memory": 640,
    "environment": {"LOG_LEVEL": "DEBUG" },
    "runtime": "python3.8",
    "app": {}
  }

Example of app memory setting in the manifest.json file for a TaskEvent app

Note: Please refer to AWS Lambda's Documentation to learn about lambda function best practices here Best practices for working with AWS Lambda functions.


1.6 Set up the POST Task​

Now that the app is configured, you can now set up the body of the POST task that will be sent to the task app.

The following is required for the POST task API Request:

  1. The API request is a POST
  2. The url is https://api.corva.ai/v2/tasks
  3. The required Authorization from a frontend app is the user's Bearer Token, e.g. "authorization: Bearer eyJhbGciOiJIUzI..."
  4. A stream, scheduled or task backend app is not authorized to invoke a task app, therefore, you must pass in an API Key. API Keys can be securely stored utilizing Secrets. The required Authorization from another backend app to a task app is an API Key, e.g. "authorization: API bca70d650da6..."
  5. The "task": { } is an object { }, therefore wrapped in { }, e.g. { "task":{ } }
  6. The "task": { } object must wrap the "provider" index, "app_key" index, "asset_id" index, "properties": { } object and "payload": { } object
  7. The required indexes of a POST request are "provider", "app_key", "asset_id"
  8. The data can be sent in either the "properties": { } object or "payload": { } object or both the "properties": { } object and "payload": { } object
{
   "task":{
      "provider":"big-data-energy",
      "app_key":"provider.app_name",
      "asset_id":5678,
      "properties":{
         "key_1":"string",
         "key_2":12345,
         "key_3":1.2345,
         "key_4":true
      },
      "payload":{
         "key_array":[
            123,
            45678
         ],
         "key_int":10
      }
   }
}

Example of the POST task body.

In this use case example, the values required by the task app are being sent in the properties {} object. 

{
   "task":{
      "provider":"big-data-energy",
      "app_key":"big-data-energy.example_task_app",
      "asset_id": 67307237,
      "properties":{
         "discounted_revenue":1574832,
         "discounted_operating_costs":200000,
         "drilling_and_completions_costs":1000000,
         "timestamp":1672444526
      }
   }
}

Example of the use case POST task body.

curl -X POST "https://api.corva.ai/v2/tasks" -H  "accept: application/json" -H  "authorization: Bearer eyJhbGciOiJIUzI1NiJ" -H  "Content-Type: application/json" -d "{   \"task\":{      \"provider\":\"big-data-energy\",      \"app_key\":\"big-data-energy.example_task_app\",      \"asset_id\": 67307237,      \"properties\":{         \"discounted_revenue\":1574832,         \"discounted_operating_costs\":200000,         \"drilling_and_completions_costs\":1000000,         \"timestamp\":1672444526      }   }}"

Example of the use case curl of POST task body.


1.7 Implement logic in the lambda_function.py file​

Now that the app is configured and the POST body is configured, you can now implement the logic in the lambda_function.py file.

Note: Implementing the logic in the lambda_function.py file is the most basic way to implement the logic. The user has the option to create directories and use Python libraries like Pydantic. 

# 1. Import required functionality.
from corva import Api, Logger, TaskEvent, task

# 2. - Decorate your function using @task. Use the the existing lambda_handler function or define your own function. It must receive two argumets: event and api. The arguments serve as building blocks for your app.
@task
def lambda_handler(event: TaskEvent, api: Api) -> list:

# 3. Here is where you can declare your variables from the argument event: TaskEvent and start using Api and Logger functionalities. 

    # The task app can declare the following attributes from the TaskEvent: company_id: The company identifier; asset_id: The asset identifier.
    asset_id = event.asset_id
    company_id = event.company_id

    #The task app can declare variables from the properties {} object. 
    timestamp = event.properties["timestamp"]
    discounted_revenue = event.properties["discounted_revenue"]
    discounted_operating_costs = event.properties["discounted_operating_costs"]
    drilling_and_completions_costs = event.properties["drilling_and_completions_costs"]

# 4. Implement calculations and logic

    #Compute Net Present Value
    npv = discounted_revenue - discounted_operating_costs - drilling_and_completions_costs

    # Utilize the Logger functionality. The default log level is Logger.info. To use a different log level, the log level must be specified in the manifest.json file in the "settings.environment": {"LOG_LEVEL": "DEBUG"}. See the Logger documentation for more information.
    Logger.debug(f"{asset_id=} {company_id=} {timestamp=}")
    Logger.debug(f"{npv=}")

# 5. This is how to set up a body of a POST request to store the data.
    output = {
        "asset_id": asset_id,
        "company_id": company_id,
        "provider": "big-data-energy",
        "collection": "example-task-app",
        "data": {
            "discounted_revenue": discounted_revenue,
            "discounted_operating_costs": discounted_operating_costs,
            "drilling_and_completions_costs": drilling_and_completions_costs,
            "npv": npv
        },
        "version": 1,
        "timestamp": timestamp
    }

    # Utilize the Logger functionality.
    Logger.debug({f"{output=}"})

# 6. Save the newly calculated data in a custom dataset

    # Utilize the Api functionality. The data=outputs needs to be an an array because Corva's data is saved as an array of objects. Objects being records. See the Api documentation for more information.
    # Checking if any data is present and then seding a POST request to Corva Data API. Please note that data is always a list.
    if output:
        # if request fails, lambda will be reinvoked. so no exception handling
        api.post(
            f"/api/v1/data/big-data-energy/example-task-app/", data=[output],
        ).raise_for_status()
    return output

1.7 Locally test your application​

1.7.1 Running a local test for your Corva App​

To locally test your Corva app, you need to follow these steps:

  1. Create a local_run.py file in your project directory.

    • This file will contain the code that simulates the environment in which your app will run.
    • It uses environment variables for authentication and API access.

  2. Set environment variables on your local machine.

    • Ensure you have set the appropriate environment variables, as shown in the code below.

    • Here's an example of how to export environment variables in your terminal:

      export API_ROOT_URL="https://api.example.com"
      export DATA_API_ROOT_URL="https://data-api.example.com"
      export CORVA_API_KEY="your_api_key"
      export APP_KEY="your_app_key"

    • You can add these export statements to your shell profile (e.g., .bashrc, .zshrc) for persistence, or run them directly in your terminal for the current session.

  3. Run the local_run.py file.

    • Once you've created the file and set your environment variables, run the script using Python to simulate the app behavior in a local environment.

    • Example:

      python local_run.py

  4. Interpret the output.

    • The output of your app will be printed in the terminal. You can use this output to verify the results of your function's execution in a local testing scenario.

Here’s an example of what the local_run.py file should look like:

from corva import Api, Logger, TaskEvent, task
from lambda_function import lambda_handler

# Main entry point for the script
if __name__ == '__main__':
    import os  # Importing os to access environment variables

    # Define test values for the asset and company IDs
    asset_id = 44315514  # Example asset ID for testing
    company_id = 1  # Example company ID for testing

    # Creating a mock TaskEvent object with test properties
    _event = TaskEvent(asset_id=asset_id, company_id=company_id, properties={'foo': 'bar'})

    # Custom BearerApi class extends the base Api class to include authentication headers
    class BearerApi(Api):
        @property
        def default_headers(self):
            # Add Bearer token and Corva-specific app key to API request headers
            return {
                'Authorization': f'Bearer {self.api_key}',  # Bearer token for authentication
                'X-Corva-App': self.app_key,  # App key for Corva apps
            }

    # Instantiate the BearerApi class with the required environment variables for authentication
    api = BearerApi(
        api_url=os.environ['API_ROOT_URL'],  # Base URL for the Corva API
        data_api_url=os.environ['DATA_API_ROOT_URL'],  # Base URL for the data API
        api_key=os.environ['CORVA_API_KEY'],  # API key from environment
        app_key=os.environ['APP_KEY'],  # App key from environment
    )

    # Call the example task app function with the mocked event and API instance
    output = lambda_handler(event=_event, api=api)

    # Print the output to verify the result of the function call
    print(output)

1.8 Deploy your application​

Please see Getting Started section 4. Upload and Publish.


1.8.1 App Runner production testing​

Task apps cannot be used with the App Runner feature. Only Stream and Scheduled apps receiving events from an asset stream can be used with the App Runner feature.


1.9 Provision the application​

For Company provisioning please see App Provisioning for more information.

For Corva Partners please see App Provisioning with a focus on App Purchases.