### Install Development Requirements

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Installs all necessary development requirements for the project. This is a prerequisite for running tests and linters.

```bash
make install
```

--------------------------------

### Install corva-sdk

Source: https://github.com/corva-ai/python-sdk/blob/master/README.md

Install the corva-sdk package using pip for Python development.

```bash
pip install corva-sdk
```

--------------------------------

### Install Corva SDK

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Installs the Corva SDK using pip. Requires Python 3.8 or higher.

```bash
pip install corva-sdk
```

--------------------------------

### Setting Up Development Environment

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Guides on setting up a Python development environment for the Corva SDK. It covers creating and activating a virtual environment.

```bash
python -m venv env
source ./env/bin/activate
```

--------------------------------

### Testing Task App

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example test for a task app. It demonstrates how to set up an `AppRunner` and execute the task app with test data.

```python
from corva.testing import AppRunner

def test_task_app():
    runner = AppRunner(app='path.to.your.task_app')
    # Mock data for the test
    data = [...] 
    # Run the app with mock data
    results = runner.run(data=data)
    # Assertions on results
```

--------------------------------

### Install Sentry SDK

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Installs the Sentry SDK for Python, which is required to integrate Sentry error reporting with the Corva application.

```bash
pip install sentry-sdk
```

--------------------------------

### Install Raygun SDK

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Installs the Raygun SDK for Python, which is necessary for integrating Raygun crash reporting into Corva applications.

```bash
pip install raygun4py
```

--------------------------------

### Run Tests and Linter

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Executes the project's tests and linter to verify the setup and code quality. This ensures the project is configured correctly.

```bash
make all
```

--------------------------------

### HTTP GET Requests with Corva API

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates making GET requests using the Corva SDK's `Api` object. It highlights the use of the optional `params` parameter for URL query strings and how to unpack the returned `requests.Response` objects.

```python
from corva.api import Api

api = Api()

# Simplest GET example
response = api.get("/some/endpoint")

# Use optional `params` parameter to provide URL query string params
response = api.get("/some/endpoint", params={"param1": "value1"})

# You can unpack received data like this, as all `Api` methods return `requests.Response` objects.
data = response.json()
```

--------------------------------

### Install Rollbar SDK

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Installs the Rollbar SDK for Python, enabling integration of Rollbar for error reporting within Corva applications.

```bash
pip install rollbar
```

--------------------------------

### List Make Targets

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Lists all available `make` targets for the project. This is a useful starting point for exploring the project's functionalities and scripts.

```bash
make help
```

--------------------------------

### Convenience Method: Get Dataset

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates fetching data from a dataset endpoint using the `Api.get_dataset` convenience method provided by the Corva SDK.

```python
from corva.api import Api

api = Api()

data = api.get_dataset(dataset_name="my_dataset")
```

--------------------------------

### Corva Followable App Example

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Illustrates the creation of a followable Corva app that produces data, triggering subsequent apps in a data pipeline. It shows saving data to a dataset and then producing messages.

```python
from corva.app import App
from corva.state import State

@App
def my_followable_app(event, context):
    # Simulate data to be saved and produced
    data_to_save = {
        "timestamp": context.get("timestamp"),
        "value": 123.45,
        "sensor_id": context.get("sensor_id")
    }

    # Save data to a dataset (e.g., 'my_dataset')
    # This is a placeholder for actual dataset insertion
    # state = State("my_dataset")
    # state.insert([data_to_save])
    print(f"Saving data: {data_to_save}")

    # Produce messages to trigger downstream apps
    # This is a placeholder for actual message production
    # context.produce(data_to_save)
    print(f"Producing message: {data_to_save}")

    return "Data processed and produced"

```

--------------------------------

### Task App

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of a task app. It imports necessary functionality, defines a function receiving event and api arguments, and is decorated with the 'task' decorator.

```python
from corva.handlers import task

@task
def my_task_app(event, api):
    """Example task app function."""
    # App logic here
```

--------------------------------

### Cache: Bulk Get, Get All, Set Many

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Covers bulk operations for the `Cache` object, including retrieving multiple keys at once (`get_many`), retrieving all data (`get_all`), and storing multiple key-value pairs simultaneously (`set_many`). It also shows setting custom expiry for bulk sets.

```python
from corva.cache import Cache
import time

cache = Cache()

# Get all the data from the hash. It is empty as we have not stored anything yet.
print(cache.get_all()) # Output: {}

# Store multiple key-value pairs at once
cache.set_many({"key1": "value1", "key2": "value2"})

# You can set custom key expiry in seconds by providing additional tuple element
cache.set_many({"key3": "value3", "key4": "value4"}, ttl=10)

# Get multiple keys at once
print(cache.get_many(["key1", "key2", "non_existent_key"]))
# Output: {'key1': 'value1', 'key2': 'value2', 'non_existent_key': None}

# If you request a non-existent key it will be assigned a value of `None`.

# Get all the data from the hash.
print(cache.get_all())
# Output: {'key1': 'value1', 'key2': 'value2', 'key3': 'value3', 'key4': 'value4'}

# Wait for key with custom expiry to expire.
time.sleep(11)

# The expired key is not present anymore.
print(cache.get("key3")) # Output: None
```

--------------------------------

### Testing Stream App with Time Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example test for a stream app processing time-based data. It outlines the steps for setting up and running the test, including mock data and assertions.

```python
from corva.testing import AppRunner

def test_stream_time_app():
    runner = AppRunner(app='path.to.your.stream_app')
    # Mock data for the test
    data = [...] 
    # Run the app with mock data
    results = runner.run(data=data)
    # Assertions on results
```

--------------------------------

### Testing Scheduled App with Data Time

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example test for a scheduled app processing data based on time. It details the process of running the app in a testing environment.

```python
from corva.testing import AppRunner

def test_scheduled_data_time_app():
    runner = AppRunner(app='path.to.your.scheduled_app')
    # Mock data for the test
    data = [...] 
    # Run the app with mock data
    results = runner.run(data=data)
    # Assertions on results
```

--------------------------------

### Testing Scheduled App with Depth Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example test for a scheduled app processing depth-based data. It outlines the steps for executing the app within a test context.

```python
from corva.testing import AppRunner

def test_scheduled_depth_app():
    runner = AppRunner(app='path.to.your.scheduled_app')
    # Mock data for the test
    data = [...] 
    # Run the app with mock data
    results = runner.run(data=data)
    # Assertions on results
```

--------------------------------

### Testing Stream App with Depth Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example test for a stream app processing depth-based data. It covers the typical testing steps, including running the app with specific inputs.

```python
from corva.testing import AppRunner

def test_stream_depth_app():
    runner = AppRunner(app='path.to.your.stream_app')
    # Mock data for the test
    data = [...] 
    # Run the app with mock data
    results = runner.run(data=data)
    # Assertions on results
```

--------------------------------

### Testing Scheduled App with Natural Time

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example test for a scheduled app processing data based on natural time intervals. It covers the standard testing procedure for such apps.

```python
from corva.testing import AppRunner

def test_scheduled_natural_time_app():
    runner = AppRunner(app='path.to.your.scheduled_app')
    # Mock data for the test
    data = [...] 
    # Run the app with mock data
    results = runner.run(data=data)
    # Assertions on results
```

--------------------------------

### Stream App (Time-based)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of a time-based stream app. It imports necessary functionality, defines a function receiving event, api, and cache arguments, and is decorated with the 'stream' decorator.

```python
from corva.handlers import stream

@stream
def my_stream_app(event, api, cache):
    """Example stream app function."""
    # App logic here
```

--------------------------------

### Partial Rerun Merge Handler

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of an optional partial rerun merge handler. It defines a function receiving event, api, cache, and cache arguments, and is decorated with 'partial_rerun_merge'.

```python
from corva.handlers import partial_rerun_merge

@partial_rerun_merge
def my_rerun_handler(event, api, cache, cache):
    """Example partial rerun merge handler."""
    # App logic here
```

--------------------------------

### Stream App (Depth-based)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of a depth-based stream app. It imports necessary functionality, defines a function receiving event, api, and cache arguments, and is decorated with the 'stream' decorator.

```python
from corva.handlers import stream

@stream
def my_stream_app(event, api, cache):
    """Example stream app function."""
    # App logic here
```

--------------------------------

### Scheduled App (Data Time-based)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of a data time-based scheduled app. It imports necessary functionality, defines a function receiving event, api, and cache arguments, and is decorated with the 'scheduled' decorator.

```python
from corva.handlers import scheduled

@scheduled
def my_scheduled_app(event, api, cache):
    """Example scheduled app function."""
    # App logic here
```

--------------------------------

### Scheduled App (Depth-based)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of a depth-based scheduled app. It imports necessary functionality, defines a function receiving event, api, and cache arguments, and is decorated with the 'scheduled' decorator.

```python
from corva.handlers import scheduled

@scheduled
def my_scheduled_app(event, api, cache):
    """Example scheduled app function."""
    # App logic here
```

--------------------------------

### Scheduled App (Natural Time-based)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Example of a natural time-based scheduled app. It imports necessary functionality, defines a function receiving event, api, and cache arguments, and is decorated with the 'scheduled' decorator.

```python
from corva.handlers import scheduled

@scheduled
def my_scheduled_app(event, api, cache):
    """Example scheduled app function."""
    # App logic here
```

--------------------------------

### Testing App with Cache (Reset)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Shows how to test an app using cache where the cache is reset for each `app_runner` call. This ensures isolated test runs.

```python
from corva.testing import AppRunner

def test_app_with_reset_cache():
    runner = AppRunner(app='path.to.your.app')
    # Run the app, cache is reset by default
    results1 = runner.run()
    # Run again, cache is reset again
    results2 = runner.run()
    # Assertions: results1 and results2 should be the same
```

--------------------------------

### Testing App with Secrets

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates how to test an app that relies on secrets. It shows passing a dictionary of secrets to the `AppRunner`.

```python
from corva.testing import AppRunner

def test_app_with_secrets():
    runner = AppRunner(app='path.to.your.app')
    # Define required secrets
    test_secrets = {
        'my_api_key': 'test_key',
        'timeout': '30'
    }
    # Run the app, passing the secrets
    results = runner.run(secrets=test_secrets)
    # Assertions on results
```

--------------------------------

### Basic Corva Logger Usage

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates the basic usage of the Corva SDK's Logger object, which is a standard Python logging instance with added contextual information and message/count limits.

```python
from corva.logger import Logger

logger = Logger()

logger.info("This is an informational message.")
logger.warning("This is a warning message.")
logger.error("This is an error message.")
```

--------------------------------

### Cache: Bulk Delete, Delete All

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Explains how to perform bulk deletion of multiple keys using `delete_many` and how to clear the entire cache using `delete_all`.

```python
from corva.cache import Cache

cache = Cache()

# Cache is empty as we have not stored anything yet.
print(cache.get_all()) # Output: {}

# Store some data.
cache.set_many({"key_a": "val_a", "key_b": "val_b", "key_c": "val_c"})
print(cache.get_all()) # Output: {'key_a': 'val_a', 'key_b': 'val_b', 'key_c': 'val_c'}

# Delete multiple keys at once.
cache.delete_many(["key_a", "key_c"])

# Deleted keys are non-existent anymore.
print(cache.get("key_a")) # Output: None
print(cache.get("key_b")) # Output: val_b
print(cache.get("key_c")) # Output: None

# Delete all the data.
cache.delete_all()

# Cache is empty.
print(cache.get_all()) # Output: {}
```

--------------------------------

### Corva App with Raygun Integration

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates how to integrate Raygun crash reporting into a Corva application by initializing the Raygun handler and passing it to the app decorator.

```python
import raygun
from corva.app import App
from corva.logger import Logger

# Initialize Raygun client
raygun_client = raygun.RaygunClient('YOUR_RAYGUN_API_KEY')

# Create a Raygun handler
class RaygunHandler(logging.Handler):
    def __init__(self, client):
        super().__init__()
        self.client = client

    def emit(self, record):
        if record.exc_info:
            self.client.captureException()
        else:
            self.client.captureMessage(record.getMessage())

raygun_handler = RaygunHandler(raygun_client)

@App(custom_handlers=[raygun_handler])
def my_raygun_app(event, context):
    logger = Logger()
    try:
        # Simulate an error
        my_list = [1, 2]
        print(my_list[5])
    except Exception as e:
        logger.error(f"Exception caught: {e}")
        # The RaygunHandler will capture this exception
    return "Raygun error reported"

```

--------------------------------

### Corva App with Rollbar Integration

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Sets up a Corva application to report errors to Rollbar by initializing the Rollbar handler and passing it to the app decorator.

```python
import rollbar
from corva.app import App
from corva.logger import Logger
from rollbar.logger import RollbarHandler

rollbar.init(
    "YOUR_ROLLBAR_ACCESS_TOKEN",
    environment="production",
    # Other Rollbar configuration options
)

rollbar_handler = RollbarHandler()

@App(custom_handlers=[rollbar_handler])
def my_rollbar_app(event, context):
    logger = Logger()
    try:
        # Simulate an error
        data = {'value': 10}
        processed_data = data['non_existent_key']
    except Exception as e:
        logger.error(f"Processing error: {e}")
        # RollbarHandler will capture and send the exception
    return "Rollbar error captured"

```

--------------------------------

### Cache: Storing and Retrieving String Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates basic usage of the `Cache` object for storing and retrieving string data. It shows how to save a string value using a key and load it back, noting that `Cache` only stores string data.

```python
from corva.cache import Cache

cache = Cache()

# Save `str` data to `Cache`
cache.set("my_string_key", "hello world")

# Load the value using its key
retrieved_value = cache.get("my_string_key")
# retrieved_value is "hello world"
```

--------------------------------

### Convenience Method: Produce Messages

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Shows how to use the `Api.produce_messages` convenience method to post data to the message producer endpoint. This method supports both 'stream' and 'scheduled' app types.

```python
from corva.api import Api

api = Api()

# Example of producing time messages
api.produce_messages(payload=[{"timestamp": 1678886400, "value": 100}], topic="time_data")

# Example of producing depth messages
api.produce_messages(payload=[{"depth": 10, "value": 50}], topic="depth_data")
```

--------------------------------

### HTTP POST, DELETE, PUT, and PATCH Requests

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Illustrates how to perform common HTTP methods (POST, DELETE, PUT, PATCH) using the Corva SDK's `Api` object. The `data` parameter is used to provide a request body, which is automatically cast to JSON.

```python
from corva.api import Api

api = Api()

# Simplest POST example
response_post = api.post("/some/endpoint", data={"key": "value"})

# Simplest DELETE example
response_delete = api.delete("/some/endpoint/123")

# Simplest PUT example
response_put = api.put("/some/endpoint/123", data={"key": "new_value"})

# Simplest PATCH example
response_patch = api.patch("/some/endpoint/123", data={"key": "updated_value"})
```

--------------------------------

### Convenience Method: Insert Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Illustrates posting data to a dataset endpoint using the `Api.insert_data` method. This method allows saving multiple documents to a dataset and optionally producing the data simultaneously.

```python
from corva.api import Api

api = Api()

# Save two documents to dataset
documents = [{"field1": "value1"}, {"field1": "value2"}]
api.insert_data(dataset_name="my_dataset", data=documents)

# Save and produce the data at once
api.insert_data(dataset_name="my_dataset", data=documents, produce=True)
```

--------------------------------

### Corva SDK API Usage

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates how to use the Corva SDK's Api object to make requests to Corva Platform and Data APIs. The Api object wraps the requests library and handles authorization and timeouts.

```APIDOC
Corva SDK API Object:

Provides methods to interact with Corva Platform API and Corva Data API.
Wraps Python 'requests' library with automatic authorization, convenient URL usage, and timeouts.
API methods return 'requests.Response' objects.

URL Usage:
  - Use Corva Platform API URL suffixes for Platform API calls.
  - Use Corva Data API URL suffixes for Data API calls.
  - Full URLs can also be provided.

Example:
  # Making a Platform API call
  response_platform = api.get('/path/to/platform/resource')

  # Making a Data API call
  response_data = api.post('/path/to/data/resource', json={'key': 'value'})

  # Using a full URL
  response_full = api.get('https://example.com/api/resource')

HTTP GET Example:
  # Example of making an HTTP GET request
  response = api.get('/api/v1/data/provider/dataset/some_provider/some_dataset')
  # Process response.json() or response.text
```

--------------------------------

### Cache: Storing and Retrieving Integer Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Explains how to store integer data in the `Cache` by casting it to a string before saving. It also shows how to retrieve the string value and cast it back to an integer for use.

```python
from corva.cache import Cache

cache = Cache()

# Cast `int` to `str` before saving data to `Cache`
cache.set("my_int_key", str(12345))

# Load the value using its key
retrieved_str_value = cache.get("my_int_key")
# retrieved_str_value is "12345"

# Cast the value back to `int` as needed
retrieved_int_value = int(retrieved_str_value)
# retrieved_int_value is 12345
```

--------------------------------

### Testing App with Cache (Reused)

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates testing an app with cache where the same cache object is reused across multiple `app_runner` calls. This allows for stateful cache behavior.

```python
from corva.testing import AppRunner
from corva.cache import Cache

def test_app_with_reused_cache():
    runner = AppRunner(app='path.to.your.app')
    # Define a cache object
    cache = Cache()
    # Run the app, passing the cache object
    results1 = runner.run(cache=cache)
    # Run again with the same cache object
    results2 = runner.run(cache=cache)
    # Assertions: results1 and results2 might differ due to cache reuse
```

--------------------------------

### Corva App with Sentry Integration

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Configures a Corva application to send errors to Sentry by initializing the Sentry SDK and passing a Sentry handler to the app decorator.

```python
import sentry_sdk
from corva.app import App
from corva.logger import Logger
from sentry_sdk.integrations.logging import SentryHandler

sentry_sdk.init(
    dsn="YOUR_SENTRY_DSN",
    # Set traces_sample_rate to 1.0 to capture 100% of transactions
    # Adjust based on your needs
    traces_sample_rate=1.0,
)

sentry_handler = SentryHandler()

@App(custom_handlers=[sentry_handler])
def my_sentry_app(event, context):
    logger = Logger()
    try:
        result = 1 / 0  # This will raise a ZeroDivisionError
    except Exception as e:
        logger.error(f"An error occurred: {e}")
        # SentryHandler will automatically capture the exception
    return "Error handled"

```

--------------------------------

### Cache: Deleting Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates the process of deleting data from the `Cache`. It shows saving data, retrieving it to confirm its presence, deleting it using its key, and then attempting to retrieve it again to confirm its absence.

```python
from corva.cache import Cache

cache = Cache()

# Save some data to `Cache`
cache.set("key_to_delete", "some_value")

# Load the value using its key
print(cache.get("key_to_delete")) # Output: some_value

# Delete the data, when it is no longer needed
cache.delete("key_to_delete")

# The data is not present anymore
print(cache.get("key_to_delete")) # Output: None
```

--------------------------------

### Custom Headers and Timeouts

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Shows how to customize HTTP requests by adding custom headers and overriding the default timeout value using the `headers` and `timeout` parameters of the `Api` object's methods.

```python
from corva.api import Api

api = Api()

# Use `headers` parameter to add custom headers to the request
custom_headers = {"X-Custom-Header": "MyValue"}
response = api.get("/some/endpoint", headers=custom_headers)

# Use `timeout` parameter to override default timeout value
response = api.get("/some/endpoint", timeout=30.0)
```

--------------------------------

### Corva Logger with Custom Handler

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Shows how to add a custom logging handler to the Corva Logger, allowing logs to be sent to additional destinations like Sentry or Rollbar alongside the default Corva handler.

```python
from corva.app import App
from corva.logger import Logger
import logging

# Assume 'my_custom_handler' is an instance of a logging handler
# e.g., from sentry_sdk.integrations.logging import SentryHandler
# my_custom_handler = SentryHandler()

# Placeholder for a custom handler
class MyCustomHandler(logging.Handler):
    def emit(self, record):
        print(f"CUSTOM HANDLER: {record.getMessage()}")

my_custom_handler = MyCustomHandler()

@App(custom_handlers=[my_custom_handler])
def my_app(event, context):
    logger = Logger()
    logger.info("Message sent to both default and custom handlers.")
    return "OK"
```

--------------------------------

### Cache: Key Expiry

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Explains how to set a Time To Live (TTL) for cache keys, which determines how long data remains accessible before being automatically deleted. By default, keys expire in 60 days.

```python
from corva.cache import Cache
import time

cache = Cache()

# Specify how many seconds you want your key to live using `ttl` parameter
cache.set("expiring_key", "will_disappear", ttl=5) # Key expires in 5 seconds

print(cache.get("expiring_key")) # Output: will_disappear

# Wait for key to expire
time.sleep(6)

# The data is not present anymore
print(cache.get("expiring_key")) # Output: None
```

--------------------------------

### Cache: Storing and Retrieving Dictionary Data

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Details how to store dictionary data in the `Cache` by first converting it to a JSON string. It covers saving the JSON string and then retrieving and parsing it back into a dictionary.

```python
import json
from corva.cache import Cache

cache = Cache()

my_dict = {"name": "Corva", "version": 1.0}

# Cast `dict` to JSON `str` before saving data to `Cache`
cache.set("my_dict_key", json.dumps(my_dict))

# Load the value using its key
retrieved_json_str = cache.get("my_dict_key")
# retrieved_json_str is '{"name": "Corva", "version": 1.0}'

# Parse JSON `str` and convert it back into a `dict`
retrieved_dict = json.loads(retrieved_json_str)
# retrieved_dict is {'name': 'Corva', 'version': 1.0}
```

--------------------------------

### Accessing Application Secrets

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Demonstrates how to import and use the `secrets` object provided by the Corva SDK to access sensitive application data. Secrets are stored as strings and may need type casting.

```python
from corva.app import App
from corva.middleware import secrets

app = App()

@app.corva_handler(event='my_event')
def my_handler(event, context):
    # Access secrets using dictionary-like access
    api_key = secrets['my_api_key']
    # Values are strings, cast to required type
    timeout_seconds = int(secrets['timeout'])
    return 'Hello World'
```

--------------------------------

### Enabling Retries for HTTP Errors

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Explains how to enable the retry functionality for requests that fail due to specific HTTP error codes (428, 500, 502, 503, 504). The SDK automatically applies exponential back-off for these errors.

```python
from corva.api import Api

# Retries are enabled by default for the specified error codes.
# No explicit code change is needed to enable this feature.
api = Api()

# Example of a request that might trigger a retry
# response = api.get("/potentially/failing/endpoint")
```

--------------------------------

### Corva Event Merging

Source: https://github.com/corva-ai/python-sdk/blob/master/docs/modules/ROOT/pages/index.adoc

Explains how to enable event merging in Corva apps, allowing multiple incoming events to be processed as a single batch. This is useful for optimizing I/O operations but requires careful handling to avoid timeouts.

```python
from corva.app import App

@App(merge_events=True)
def my_merged_app(event, context):
    # 'event' will contain multiple events if merge_events=True
    # Process the batch of events here
    print(f"Received {len(event)} events in this batch.")
    return "Batch processed"

@App(merge_events=False)
def my_unmerged_app(event, context):
    # 'event' will contain a single event
    # Process the single event
    print("Received a single event.")
    return "Single event processed"

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.