incountry 5.0.0
InCountry Storage SDK
License: MIT
Requires Python: >=3.6.0


# InCountry Storage SDK

[![Build Status](https://travis-ci.com/incountry/sdk-python.svg?branch=master)](https://travis-ci.com/incountry/sdk-python)
[![Sonarcloud Status](https://sonarcloud.io/api/project_badges/measure?project=incountry_sdk-python&metric=alert_status)](https://sonarcloud.io/dashboard?id=incountry_sdk-python)
[![Sonarcloud Status](https://sonarcloud.io/api/project_badges/measure?project=incountry_sdk-python&metric=coverage)](https://sonarcloud.io/dashboard?id=incountry_sdk-python)
[![Bugs](https://sonarcloud.io/api/project_badges/measure?project=incountry_sdk-python&metric=vulnerabilities)](https://sonarcloud.io/dashboard?id=incountry_sdk-python)

# Table of contents

- [Installation](#installation)
- [Countries list](#countries-list)
- [Quickstart guide](#quickstart-guide)
- [Storage configuration](#storage-configuration)
  - [Extra Storage options](#extra-storage-options)
  - [Request Options](#request-options)
  - [oAuth options configuration](#oauth-options-configuration)
  - [Encryption key/secret](#encryption-keysecret)
  - [Server side encryption](#server-side-encryption)
  - [Debug mode](#debug-mode)
- [Usage](#usage)
  - [Writing data to storage](#writing-data-to-storage)
    - [List of available record fields](#list-of-available-record-fields)
    - [Date fields](#date-fields)
  - [Batches](#batches)
  - [Reading stored data](#reading-stored-data)
  - [Find records](#find-records)
    - [Sorting](#Sorting)
    - [Find error handling](#find-error-handling)
  - [Find one record](#find-one-record)
  - [Deleting a single record](#deleting-a-single-record)
  - [Deleting multiple records](#deleting-multiple-records)
- [Attaching files to a record](#attaching-files-to-a-record)
  - [Adding attachments](#adding-attachments)
  - [Deleting attachments](#deleting-attachments)
  - [Downloading attachments](#downloading-attachments)
  - [Working with attachment meta info](#working-with-attachment-meta-info)
- [Health check](#health-check)
- [Data migration and key rotation support](#data-migration-and-key-rotation-support)
- [AWS KMS integration](#aws-kms-integration)
- [Error handling](#error-handling)
- [Custom encryption support](#custom-encryption-support)
- [Testing Locally](#testing-locally)

## Installation

To install Python SDK, use `pipenv` (or `pip`), as follows:

```
$ pipenv install incountry
```

## Countries List

To get the full list of supported countries and their codes, please [follow this link](countries.md).

## Quickstart guide

To access your data in InCountry Platform by using Python SDK, you need to create an instance of the Storage class. You can retrieve the `client_id`, `client_secret` and `environment_id` variables from your dashboard on InCountry Portal.

```python
from incountry import Storage, SecretKeyAccessor

storage = Storage(
    client_id="<client_id>",
    client_secret="<client_secret>",
    environment_id="<environment_id>",
    secret_key_accessor=SecretKeyAccessor(lambda: "<encryption_secret>"),
)
```

## Storage Configuration

Below you can find a full list of possible configuration options for creating a Storage instance.

```python
class Storage:
    def __init__(
        self,
        # Required.
        # Can be set via INC_ENVIRONMENT_ID env variable.
        environment_id: Optional[str] = None,

        # Required when using oAuth authorization.
        # Can be set via INC_CLIENT_ID env variable.
        client_id: Optional[str] = None,

        # Required when using oAuth authorization.
        # Can be set via INC_CLIENT_SECRET env variable.
        client_secret: Optional[str] = None,

        # Used when OAuth token is already acquired prior to Storage initialization.
        # Mutually exclusive with client_id, client_secret.
        oauth_token: Optional[str] = None,

        # Instance of the SecretKeyAccessor class.
        # Used to fetch the encryption secret. Not compatible with SSE mode.
        secret_key_accessor: Optional[SecretKeyAccessor] = None,

        # Optional. Defines API URL.
        # Can be set via INC_ENDPOINT env variable.
        endpoint: Optional[str] = None,

        # Optional. Default is EncryptionType.LOCAL.
        # Switch encryption modes:
        #   EncryptionType.LOCAL - Enables encryption using SDK. Encrypted data will be sent to InCountry platform.
        #   EncryptionType.SSE - Enables the server side encryption (SSE) mode.
        #                        Unencrypted data will be sent to InCountry platform via HTTPS,
        #                        then encrypted and saved on the server side.
        #                        Enabling this option disables local (SDK-side) encryption, hashing and normalizing.
        #   EncryptionType.OFF - Base64 encoded not encrypted data will be sent and stored at InCountry platform
        # Also supports boolean values as backward compatibility:
        #   True = EncryptionType.LOCAL
        #   False = EncryptionType.OFF
        encrypt: Optional[Union[bool, str]] = True,

        # Optional. If True enables the additional debug logging.
        debug: Optional[bool] = False,

         # Optional. Used to fine-tune some configurations.
        options: Optional[Dict[str, Any]] = {},

        # Optional. List of custom encryption configurations.
        custom_encryption_configs: Optional[List[dict]] = None,
    ):
        ...
```

---

#### Extra Storage options

You can use the `options` parameter to tweak some SDK configurations:

```python
storage = Storage(
    ...,
    options={
        "http_options": {
            "timeout": int,         # Defines http requests timeout, in seconds.
                                    # Default value is 30
            "retryBaseDelay": int,  # Sets a custom initial http request retry delay in seconds.
                                    # Retries when receiving HTTP 429 code from the server.
                                    # Default value is 1
            "retryMaxDelay": int,   # Sets a custom maximum retry delay in seconds.
                                    # Default value is 32
        },

        "auth_endpoints": dict,     # A custom endpoints map that should be used for
                                    # fetching oAuth tokens

        "countries_endpoint": str,  # If your PoPAPI configuration relies on a custom
                                    # PoPAPI server (rather than the default one)
                                    # use the `countries_endpoint` option to specify
                                    # the endpoint responsible for fetching the list
                                    # of supported countries

        "endpoint_mask": str,       # Defines API base hostname part to use.
                                    # If set, all requests will be sent to
                                    # https://${country}${endpoint_mask} host
                                    # instead of the default one
                                    # (https://${country}-mt-01.api.incountry.io)

        "normalize_keys": bool,     # If set to True all the string record keys will be stored in lowercase
                                    # (except 'body' and 'precommit_body').
                                    # Not compatible with SSE mode.

        "hash_search_keys": bool,   # If set to False key1-key20 will be stored unhashed, enabling
                                    # partial match search using 'search_keys'.
                                    # Not compatible with SSE mode.
    }
)
```

#### Request Options

Every Storage method allows to pass custom HTTP headers via `request_options` parameter, e.g.:

```python
storage.write(
    country="us",
    record_key="user_1",
    ...,
    request_options={
        "http_headers": {
            "request-id": "<uuid>",
        },
    },
)
```

#### oAuth options configuration

The SDK allows to precisely configure oAuth authorization endpoints (if needed). Use this option only if your plan configuration requires so.

Below you can find the example of how to create a storage instance with custom oAuth endpoints:

```python
storage = Storage(
    client_id="<client_id>",
    client_secret="<client_secret>",
    environment_id="<environment_id>",
    secret_key_accessor=SecretKeyAccessor(lambda: "<encryption_secret>"),
    options={
        "auth_endpoints": {
            "default": "<default_auth_endpoint>",
            "emea": "<auth_endpoint_for_emea_region>",
            "apac": "<auth_endpoint_for_apac_region>",
            "amer": "<auth_endpoint_for_amer_region>",
        },
    },
)
```

#### Encryption key/secret

The `secret_key_accessor` variable is used to pass a key or secret used for data encryption.

The `SecretKeyAccessor` class constructor allows you to pass a function that should return either a string representing your secret or a dictionary (we call it `secrets_data` object):

```python
{
  "secrets": [{
       "secret": str,
       "version": int, # Should be an integer greater than or equal to 0
       "isKey": bool,  # Should be True only for user-defined encryption keys
    }
  }, ....],
  "currentVersion": int,
}
```

Note: even though SDK uses PBKDF2 to generate a cryptographically strong encryption key, you must ensure that you provide a secret/password which follows the modern security best practices and standards.

The `secrets_data` variable allows you to specify multiple keys/secrets which SDK will use for data decryption based on the version of the key or secret used for encryption. Meanwhile SDK will encrypt data only by using a key (or secret) which matches the `currentVersion` parameter provided in the `secrets_data` object.

This enables the flexibility required to support Key Rotation policies when secrets (or keys) must be changed with time. The SDK will encrypt data by using the current secret (or key) while maintaining the ability to decrypt data records that were encrypted with old secrets (or keys). The SDK also provides a method for data migration which allows you to re-encrypt data with the newest secret (or key). For details please see the `migrate` method.

The SDK allows you to use custom encryption keys, instead of secrets. Please note that a user-defined encryption key should be a base64-encoded 32-bytes-long key as required by AES-256 cryptographic algorithm.

Below you can find several examples of how you can use `SecretKeyAccessor` module.

```python
# Get a secret from a variable
from incountry import SecretKeyAccessor

password = "password"
secret_key_accessor = SecretKeyAccessor(lambda: password)

# Get secrets via an HTTP request
from incountry import SecretKeyAccessor
import requests as req

def get_secrets_data():
    url = "<your_secret_url>"
    r = req.get(url)
    return r.json() # assuming response is a `secrets_data` object

secret_key_accessor = SecretKeyAccessor(get_secrets_data)
```

#### Server side encryption

SDK supports server side encryption mode. In this mode all the data encryption and hashing is performed server-side. The input data is sent to the InCountry platform as is.
Below you can find an example of how to use the `EncryptionType.SSE` option:

```python
from incountry import Storage, EncryptionType

storage = Storage(
    client_id="<client_id>",
    client_secret="<client_secret>",
    environment_id="<environment_id>",
    encrypt=EncryptionType.SSE,
)
```

With SSE mode enabled SDK ignores the following configuration parameters: `options.normalize_keys`, `options.hash_search_keys` and `secret_key_accessor`.

**NOTE** Availability of the SSE mode may vary per subscription plan. Please contact the InCountry support team for details.

#### Debug mode

SDK supports debug mode allowing all SDK public methods to return raw http response along with the original return data.

**Note** Some methods may perform multiple requests (e.g. `migrate`). In that case raw http response data will be represented as a `list`.

```python
from incountry import Storage

storage = Storage(
    ...,
    debug=True,
    ...,
)

read_result = storage.read(...)

# you can use original data as usual
read_result["record"]

# raw http response goes here (instance of requests.Response)
read_result[Storage.HTTP_DEBUG_KEY]

# e.g to access response headers
read_result[Storage.HTTP_DEBUG_KEY].headers["<header_name>"]

# migrate http
migrate_result = storage.migrate(...)
migrate_result[Storage.HTTP_DEBUG_KEY][0] # first request - find
migrate_result[Storage.HTTP_DEBUG_KEY][1] # second request - batch_write
```

## Usage

### Writing data to Storage

Use the `write` method in order to create/replace a record (by `record_key`).

```python
def write(self, country: str, record_key: str, request_options: dict = {}, **record_data: Union[str, int]) -> Dict[str, TRecord]:
    ...


# write returns created record dict on success
{
    "record": Dict
}
```

Below you can find an example of how you can use the `write` method.

```python
write_result = storage.write(
    country="us",
    record_key="user_1",
    body="some PII data",
    profile_key="customer",
    range_key1=10000,
    key1="english",
    key2="rolls-royce",
)

# write_result would be as follows
write_result = {
    "record": {
        "record_key": "user_1",
        "body": "some PII data",
        "profile_key": "customer",
        "range_key1": 10000,
        "key1": "english",
        "key2": "rolls-royce",
    }
}
```

For the list of possible `record_data` kwargs, see the section below.

#### List of available record fields

v3.0.0 release introduced a series of new fields available for data storage. Below you can find the full list of all the fields available for storage in InCountry Platform along with their types and storage methods. Each field is either encrypted, hashed or stored as is:

```python
# String fields, hashed
record_key
key1
key2
key3
key4
key5
key6
key7
key8
key9
key10
key11
key12
key13
key14
key15
key16
key17
key18
key19
key20
profile_key
service_key1
service_key2
service_key3
service_key4
service_key5
parent_key

# String fields, encrypted
body
precommit_body

# Date fields, plain
expires_at

# Int fields, plain
range_key1
range_key2
range_key3
range_key4
range_key5
range_key6
range_key7
range_key8
range_key9
range_key10
```

#### Date fields

Use the `created_at` and `updated_at` fields to access date-related information about records. The `created_at` field stores a date when a record was initially created in the target country. The `updated_at` field stores a date of the latest write operation for the given `record_key`.

Use `expires_at` field to limit the lifespan of a record stored in InCountry Platform. The record will be automatically deleted upon reaching the specified date.

Below is an example of how to create a record that will be deleted in a day:

```python
from datetime import datetime, timedelta, timezone
...

deletion_date = datetime.now(timezone.utc) + timedelta(days=1)

write_result = storage.write(
    country="us",
    record_key="user_1",
    expires_at=deletion_date,
)

# write_result would be as follows
write_result = {
    "record": {
        "record_key": "user_1",
        "expires_at": datetime.datetime(2021, 8, 11, 12, 36, 52, 607534, tzinfo=datetime.timezone.utc),
        "created_at": datetime.datetime(2021, 7, 26, 15, 34, 28, 910266, tzinfo=datetime.timezone.utc),
        "updated_at": datetime.datetime(2021, 8, 10, 12, 36, 54, 879828, tzinfo=datetime.timezone.utc),
    }
}
```

---

**NOTE**

All dates are stored in UTC timezone and converted to UTC timezone in case of `expires_at`.

It's highly recommended to provide timezone-aware datetime objects to avoid any timezone-related issues in future.

E.g., to create an object representing the current time in UTC you can call `datetime.now(timezone.utc)`

Please avoid using `datetime.utcnow()` as it returns naive datetime object without timezone information, which may result in wrong timezone conversion when passed as `expires_at` value.

---

### Batches

Use the `batch_write` method to create/replace multiple records at once.

```python
def batch_write(self, country: str, records: List[TRecord], request_options: dict = {}) -> Dict[str, List[TRecord]]:
    ...


# batch_write returns the following dict of created records
{
    "records": List
}
```

Below you can find an example of how to use this method.

```python
batch_result = storage.batch_write(
    country="us",
    records=[
        {"record_key": "key1", "body": "body1", ...},
        {"record_key": "key2", "body": "body2", ...},
    ],
)

# batch_result would be as follows
batch_result = {
    "records": [
        {"record_key": "key1", "body": "body1", ...},
        {"record_key": "key2", "body": "body2", ...},
    ]
}
```

### Reading stored data

You can read the stored data records by its `record_key` using the `read` method.

```python
def read(self, country: str, record_key: str, request_options: dict = {}) -> Dict[str, TRecord]:
    ...


# The read method returns the record dictionary if the record is found
{
    "record": Dict
}
```

The `read` method usage is as follows:

```python
read_result = storage.read(country="us", record_key="user1")

# read_result would be as follows
read_result = {
    "record": {
        "record_key": "user_1",
        "body": "some PII data",
        "profile_key": "customer",
        "range_key1": 10000,
        "key1": "english",
        "key2": "rolls-royce",
        "created_at": datetime.datetime(...),
        "updated_at": datetime.datetime(...),
    }
}
```

### Find records

You can look up for records by keys or version using the `find` method.

```python
def find(
        self,
        country: str,
        limit: Optional[int] = FIND_LIMIT,
        offset: Optional[int] = 0,
        sort: Optional[List[TSortFilter]] = None,
        request_options: dict = {},
        **filters: Union[TIntFilter, TStringFilter],
    ) -> Dict[str, Any]:
    ...
```

Note: SDK returns 100 records at most.

The returned object looks like the following:

```python
{
    "data": List,
    "errors": List, # optional
    "meta": {
        "limit": int,
        "offset": int,
        "total": int,  # total records matching filter, ignoring limit
    }
}
```

You can use the following options to look up for records by hashed string keys from the [list above](#list-of-available-record-fields):

```python
# single value
key1="value1" # records with key1 equal to "value1"

# list of values
key2=["value1", "value2"] # records with key2 equal to "value1" or "value2"

# dict with $not operator
key3={"$not": "value1"} # records with key3 not equal "value1"
key4={"$not": ["value1", "value2"]} # records with key4 equal to neither "value1" or "value2"
```

You can use special `search_keys` filter to search records by partial match (similar to `LIKE` SQL operator) among record's text fields `key1, ..., key20`.

```python
search_keys="text to find"
```

---

**NOTE**

`search_keys` cannot be used in combination with any of `key1, key2, ..., key20` keys and works only in combination with non-hashing Storage mode (with `hash_search_keys` Storage option set to `False`).

---

Another way to find records by partial key match is using `$like` operator. It provides a partial match search (similar to the `LIKE` SQL operator without special characters) against one of the record’s string fields `key1, key2, ..., key20`.

```python
key3={ "$like": "abc" } # Matches all records where record.key3 LIKE 'abc'
```

Note: You can use either `searchKeys` or `$like`, not both. `hash_search_keys` Storage option should also be set to `False`.

You can use the following options to look up for records by int keys from the [list above](#list-of-available-record-fields):

```python
# single value
range_key1=1 # records with range_key1 equal to 1

# list of values
range_key2=[1, 2] # records with range_key2 equal to 1 or 2

# dictionary with comparison operators
range_key3={"$gt": 1} # records with range_key3 greater than 1
range_key4={"$gte": 1} # records with range_key4 greater than or equal to 1
range_key5={"$lt": 1} # records with range_key5 less than 1
range_key6={"$lte": 1} # records with range_key6 less than or equal to 1

# you can combine different comparison operators
range_key7={"$gt": 1, "$lte": 10} # records with range_key7 greater than 1 and less than or equal to 10

# you cannot combine similar comparison operators - e.g. $gt and $gte, $lt and $lte
```

You can use the following option to look up for records by `version` (encryption key version):

```python
# single value
version=1 # records with version equal to 1

# list of values
version=[1, 2] # records with version equal to 1 or 2

# dictionary with $not operator
version={"$not": 1} # records with version not equal 1
version={"$not": [1, 2]} # records with version equal to neither 1 or 2
```

By default all the search keys are joined with the logical `AND` operator. You can switch to logical `OR` by using `$or` operator:

```python
query = {
    "$or": [
        {"key1": "123"},
        {"key2": "123"}
    ]
}
find_result = storage.find(country="us", **query])
```

Note: currently you can only use string keys inside `$or` operator. Nested `$or` operators are not supported as well.

#### Sorting

You can use `sort` parameter to sort the returned records by one or multiple keys.
SDK currently supports sorting by `key1...key20`, `range_key1...range_key10`, `created_at`, `updated_at, expires_at` keys.

```python
from incountry import SortOrder

# you can use SordOrder enum class to specify sorting order
sort=[{"key1": SortOrder.ASC}, {"range_key1": SortOrder.DESC}, {"created_at": SortOrder.DESC}]

# or you can use string values
sort=[{"key1": "asc"}, {"range_key1": "desc"}, {"updated_at": "asc"}]
```

---

**WARNING**

To use `sort` in find() call for string keys `key1...key20` you need to set `hash_search_keys` option to `False`.

---

Below you can find an example of how you can use the `find` method:

```python
find_result = storage.find(country="us", limit=10, offset=10, key1="value1", key2=["value2", "value3"], sort=[{"key1": "asc"}])

# find_result would be as follows
find_result = {
    "data": [
        {
            "record_key": "<record_key>",
            "body": "<body>",
            "key1": "value1",
            "key2": "value2",
            "created_at": datetime.datetime(...),
            "updated_at": datetime.datetime(...),
            ...
        }
    ],
    "meta": {
        "limit": 10,
        "offset": 10,
        "total": 100,
    }
}
```

#### Find error handling

There may be a situation when the `find` method receives records that cannot be decrypted.
For example, this may happen once the encryption key has been changed while the found data was encrypted with the older version of that key.
In such cases data returned by the find() method will be as follows:

```python
{
    "data": [...],  # successfully decrypted records
    "errors": [{
        "rawData",  # raw record which caused decryption error
        "error",    # decryption error description
    }, ...],
    "meta": { ... }
}
```

### Find one record

If you need to find only one of the records matching a specific filter, you can use the `find_one` method.

```python
def find_one(
        self,
        country: str,
        offset: Optional[int] = 0,
        sort: Optional[List[TSortFilter]] = None,
        request_options: dict = {},
        **filters: Union[TIntFilter, TStringFilter],
    ) -> Union[None, Dict[str, Dict]]:
    ...


# If a record is not found, the find_one method returns `None`. Otherwise it returns a record dictionary.
{
    "record": Dict
}
```

Below you can find the example of how to use the `find_one` method:

```python
find_one_result = storage.find_one(country="us", key1="english", key2=["rolls-royce", "bmw"])

# find_one_result would be as follows
find_one_result = {
    "record": {
        "record_key": "user_1",
        "body": "some PII data",
        "profile_key": "customer",
        "range_key1": 10000,
        "key1": "english",
        "key2": "rolls-royce",
    }
}
```

### Deleting a single record

Use the `delete` method in order to delete a record from InCountry Platform. It is only possible by using the `record_key` field.

```python
def delete(self, country: str, record_key: str, request_options: dict = {}) -> Dict[str, bool]:
    ...


# the delete method returns the following dictionary upon success
{
    "success": True
}
```

Below you can find the example of how to use the delete method:

```python
delete_result = storage.delete(country="us", record_key="<record_key>")

# delete_result would be as follows
delete_result = {
    "success": True
}
```

### Deleting multiple records

You can use the `batch_delete` method to delete multiple records from the InCountry platform. For now, the SDK allows deletion only by a list of `record_keys`.

```python
def batch_delete(self, country: str, record_key: List[str], request_options: dict = {}) -> Dict[str, bool]:
    ...

# the batch_delete method returns the following dictionary upon success
{
    "success": True
}
```

Below you can find the example of how to use the delete method:

```python
delete_result = storage.batch_delete(country="us", record_key=["<record_key>"])

# delete_result would be as follows
delete_result = {
    "success": True
}
```

## Attaching files to a record

**NOTE**

Attachments are currently available for InCountry dedicated instances only. Please check your subscription plan for details. This may require specifying your dedicated instance endpoint when configuring Python SDK Storage.

---

InCountry Storage allows you to attach files to the previously created records. Attachments' meta information is available through the `attachments` field of `record` dictionary, returned by `read`, `find` and `find_one` methods.

```python
read_result = storage.read(country="us", record_key="<record_key>")

# read_result with attachments would be as follows
read_result = {
    "record": {
        "record_key": "user_1",
        ...,
        "attachments": [
            {
                "created_at": datetime,
                "updated_at": datetime,
                "download_link": "<download_url>",
                "file_id": "<file_id>",
                "filename": "<file_name>.<ext>",
                "hash": "<hash>",
                "mime_type": "<mime_type>",
                "size": 1,
            },
        ],
    },
}
```

### Adding attachments

---

## Note:

The `add_attachment` method allows you to add or replace attachments.
File data can be provided either as `BinaryIO` object or `string` with a path to the file in the file system.

```python
def add_attachment(
    self,
    country: str,
    record_key: str,
    file: Union[BinaryIO, str],
    mime_type: str = None,
    upsert: bool = False,
    request_options: dict = {},
) -> Dict[str, Union[str, int, datetime]]:
    ...

# The add_attachment returns attachment_meta dictionary
{
    "attachment_meta": {
        "created_at": datetime,
        "updated_at": datetime,
        "download_link": "<download_url>",
        "file_id": "<file_id>",
        "filename": "<file_name>.<ext>",
        "hash": "<hash>",
        "mime_type": "<mime_type>",
        "size": 1,
    }
}
```

Example of usage:

```python
# using file path
storage.add_attachment(country="us", record_key="<record_key>", file="./README.md")

# using BinaryIO object
with open("./README.md", "rb") as attachment_file:
    storage.add_attachment(country="<country>", record_key="<record_key>", file=attachment_file)
```

### Deleting attachments

The `delete_attachment` method allows you to delete attachment using its `file_id`.

```python
def delete_attachment(
    self, country: str, record_key: str, file_id: str, request_options: dict = {}
) -> Dict[str, bool]::
    ...

# the delete_attachment method returns the following dictionary upon success
{
    "success": True
}
```

Example of usage:

```python
delete_result = storage.delete_attachment(country="us", record_key="<record_key>", file_id="<file_id>")

# delete_result would be as follows
delete_result = {
    "success": True
}
```

### Downloading attachments

The `get_attachment_file` method allows you to download attachment contents.
It returns dictionary with readable file body and filename.

```python
def get_attachment_file(
    self, country: str, record_key: str, file_id: str, request_options: dict = {}
) -> Dict[str, Dict]:
    ...

# the get_attachment_file method returns the following dictionary upon success
{
    "attachment_data": {
        "filename": str,
        "file": BytesIO,
    }
}
```

Example of usage:

```python
get_attachment_res = storage.get_attachment_file(country="us", record_key="<record_key>", file_id="<file_id>")
attachment_data = get_attachment_res["attachment_data"]

with open(attachment_data["filename"], "wb") as f:
    f.write(attachment_data["file"].read())
```

### Working with attachment meta info

The `get_attachment_meta` method allows you to retrieve attachment's metadata using its `file_id`.

```python
def get_attachment_meta(
    self, country: str, record_key: str, file_id: str, request_options: dict = {}
) -> Dict[str, Union[str, int, datetime]]:
    ...

# The get_attachment_meta returns attachment_meta dictionary
{
    "attachment_meta": {
        "created_at": datetime,
        "updated_at": datetime,
        "download_link": "<download_url>",
        "file_id": "<file_id>",
        "filename": "<file_name>.<ext>",
        "hash": "<hash>",
        "mime_type": "<mime_type>",
        "size": 1,
    }
}
```

Example of usage:

```python
get_attachment_meta_res = storage.get_attachment_meta(country="us", record_key="<record_key>", file_id="<file_id>")
```

The `update_attachment_meta` method allows you to update attachment's metadata (MIME type and file name).

```python
def update_attachment_meta(
    self,
    country: str,
    record_key: str,
    file_id: str,
    filename: str = None,
    mime_type: str = None,
    request_options: dict = {},
) -> Dict[str, Union[str, int, datetime]]:
    ...


# The update_attachment_meta returns attachment_meta dictionary
{
    "attachment_meta": {
        "created_at": datetime,
        "updated_at": datetime,
        "download_link": "<download_url>",
        "file_id": "<file_id>",
        "filename": "<file_name>.<ext>",
        "hash": "<hash>",
        "mime_type": "<mime_type>",
        "size": 1,
    }
}
```

Example of usage:

```python
storage.update_attachment_meta(country="us", record_key="<record_key>", file_id="<file_id>", filename="new_file_name.txt", mime_type="text/plain")
```

## Health check

The `health_check` method of `Storage` class allows you to check availability of a remote storage service by country identifier

```python
def health_check(
    self,
    country: str,
    request_options: dict = {},
) -> Dict[str, bool]:
```

Example of usage:

```python
response = storage.health_check(country="us")

# response would be as follows
response = {
    "result": True
}
```

## Data Migration and Key Rotation support

Using `secret_key_accessor` which provides the `secrets_data` object enables key rotation and data migration support.

SDK introduces the `migrate` method which allows you to re-encrypt data encrypted with older versions of the secret.

```python
def migrate(self, country: str, limit: Optional[int] = FIND_LIMIT, request_options: dict = {}) -> Dict[str, int]:
    ...


# The migrate method returns the following dictionary with meta information
{
    "migrated": int   # the number of records migrated
	"total_left": int # the number of records left to migrate (number of records with version
                      # different from `currentVersion` provided by `secret_key_accessor`)
}
```

You should specify the `country` parameter which you want to perform migration in. Additionally, you need to specify the `limit` parameter for a precise number of records to migrate.

Note: the maximum number of records that can be migrated per one request is 100.

For a detailed example of a migration script, please see `/examples/full_migration.py`

## AWS KMS integration

InCountry Python SDK supports usage of any 32-byte (256-bit) AES key, including ones produced by AWS KMS symmetric master key (CMK).

The suggested use case assumes that AWS user already got his KMS encrypted data key (AES_256) generated. Afterwards the key gets decrypted using AWS Python client library (`boto3`) and then provided to InCountry Python SDK's `SecretKeyAccessor`.

For a detailed example of AWS KMS keys usage please see [examples/aws-kms.py](examples/aws-kms.py)

## Error Handling

InCountry Python SDK may throw the following Exceptions:

- **StorageConfigValidationException** - used for Storage options validation exceptions. Can be thrown when trying to create a `Storage` class instance.

- **SecretsProviderException** - can be thrown in runtime after `Storage` instance creation when an exception occurs trying to retrieve secrets.

- **SecretsValidationException** - can be thrown if secrets returned by `SecretKeyAccessor` instance are in the wrong format.

- **InputValidationException** - used for input validation exceptions. Can be thrown by all `Storage` public methods.

- **StorageClientException** - used for various exceptions related to Storage misconfiguration or wrong input data provision. All of the exception classes `StorageConfigValidationException`, `SecretsProviderException`, `SecretsValidationException`, `InputValidationException` are inherited from `StorageClientException`.

- **StorageServerException** - thrown if SDK fails to communicate with InCountry servers or if a server response validation fails.

- **StorageCountryNotSupportedException** - thrown if InCountry platform doesn't support storage of data for the requested country. Inherited from `StorageServerException`.

- **StorageServerResponseValidationException** - thrown if a server response validation fails. Inherited from `StorageServerException`.

- **StorageNetworkException** - thrown if SDK failed to communicate with InCountry servers due to network issues, such as request timeout, unreachable endpoint etc. Inherited from `StorageServerException`.

- **StorageAuthenticationException** - can be thrown if SDK failed to authenticate in InCountry system with the provided credentials. Inherited from `StorageServerException`.

- **StorageCryptoException** - thrown during encryption/decryption procedures (both default and custom). This may be an indication of malformed/corrupted data or a wrong encryption key provided to the SDK.

- **StorageException** - general exception. Inherited by all the other exceptions.

We strongly recommend you to gracefully handle all the possible exceptions:

```python
try:
    # use InCountry Storage instance here
except StorageConfigValidationException as e:
    # storage configuration error
except SecretsProviderException as e:
    # error trying to acquire secrets
except SecretsValidationException as e:
    # secrets validation error
except InputValidationException as e:
    # input validation error
except StorageClientException as e:
    # general error related to user somehow provided wrong input data
    # when creating storage or calling a storage method
except StorageServerResponseValidationException as e:
    # error when trying to validate server response
except StorageNetworkException as e:
    # network-related error
except StorageAuthenticationException as e:
    # authentication error
except StorageCountryNotSupportedException as e:
    # requested country is not supported
except StorageServerException as e:
    # general error during SDK<->server communication
except StorageCryptoException as e:
    # some encryption error
except StorageException as e:
    # general error
except Exception as e:
    # something else happened not related to InCountry SDK
```

## Custom Encryption Support

SDK supports a capability to provide custom encryption/decryption methods if you decide to use your own algorithm instead of the default one.

The `Storage` constructor allows you to pass `custom_encryption_configs` parameters as an array of custom encryption configurations within the following schema which enables custom encryption:

```python
{
    "encrypt": Callable,
    "decrypt": Callable,
    "isCurrent": bool,
    "version": str
}
```

Both `encrypt` and `decrypt` attributes should be functions implementing the following interface (with exactly the same argument names)

```python
encrypt(input:str, key:bytes, key_version:int) -> str:
    ...

decrypt(input:str, key:bytes, key_version:int) -> str:
    ...
```

They should accept raw data to encrypt/decrypt, key data (represented as bytes array) and key version received from `SecretKeyAccessor`.
The resulted encrypted/decrypted data should be a string.

---

**NOTE**

You should provide a specific encryption key through `secrets_data` passed to `SecretKeyAccessor`. This secret should use the `isForCustomEncryption` flag instead of the regular `isKey` flag.

```python
secrets_data = {
  "secrets": [{
       "secret": "<secret for custom encryption>",
       "version": 1,
       "isForCustomEncryption": True,
    }
  }],
  "currentVersion": 1,
}

secret_accessor = SecretKeyAccessor(lambda: secrets_data)
```

---

The `version` attribute is used to differentiate custom encryptions from each other and from the default encryption as well.
This way SDK will be able to successfully decrypt any old data if encryption changes with time.

The `isCurrent` attribute allows you to specify one of the custom encryption configurations which should be used for encryption. Only one encryption configuration can be set as `"isCurrent": True`.

If none of the encryption configurations have `"isCurrent": True` then the SDK will use default encryption configuration to encrypt stored data. At the same time it will preserve the ability to decrypt old data which was encrypted with custom encryption configuration (if any).

Below you can find the example of how you can set up the SDK to use custom encryption (using Fernet encryption method from https://cryptography.io/en/latest/fernet/)

```python
import os

from incountry import InCrypto, SecretKeyAccessor, Storage
from cryptography.fernet import Fernet

def enc(input: str, key: bytes, key_version: int):
    # Fernet cipher from cryptography package accepts keys as base64-encoded string
    # while InCountry SDK decodes keys as bytes
    # thus we need to do some bytes-to-b64 encoding
    cipher = Fernet(InCrypto.b_to_base64(key))
    return cipher.encrypt(input.encode("utf8")).decode("utf8")

def dec(input: str, key: bytes, key_version: int):
    cipher = Fernet(InCrypto.b_to_base64(key))
    return cipher.decrypt(input.encode("utf8")).decode("utf8")

custom_encryption_configs = [
    {
        "encrypt": enc,
        "decrypt": dec,
        "version": "test",
        "isCurrent": True,
    }
]

key = os.urandom(InCrypto.KEY_LENGTH)  # Fernet uses 32-byte length keys

secret_key_accessor = SecretKeyAccessor(
    lambda: {
        "currentVersion": 1,
        "secrets": [{"secret": key, "version": 1, "isForCustomEncryption": True}],
    }
)

storage = Storage(
    environment_id="<env_id>",
    client_id="<client_id>",
    client_secret="<client_secret>",
    secret_key_accessor=secret_key_accessor,
    custom_encryption_configs=custom_encryption_configs,
)

storage.write(country="us", record_key="<record_key>", body="<body>")
```

## Testing Locally

1. In terminal run `pipenv run tests` for unit tests
2. In terminal run `pipenv run integrations` to run integration tests