Description
This Python SDK is designed to help developers easily implement Skyflow into their python backend.
Table of Contents
Features
Authentication with a Skyflow Service Account and generation of a bearer token
Vault API operations to insert, retrieve and tokenize sensitive data
Invoking connections to call downstream third party APIs without directly handling sensitive data
Installation
Requirements
- Python 3.7.0 and above
Configuration
The package can be installed using pip:
pip install skyflow
Service Account Bearer Token Generation
The Service Account python module is used to generate service account tokens from service account credentials file which is downloaded upon creation of service account. The token generated from this module is valid for 60 minutes and can be used to make API calls to vault services as well as management API(s) based on the permissions of the service account.
The generate_bearer_token(filepath)
function takes the credentials file path for token generation, alternatively, you can also send the entire credentials as string, by using generate_bearer_token_from_creds(credentials)
from skyflow.errors import SkyflowError
from skyflow.service_account import generate_bearer_token, is_expired
# cache token for reuse
bearerToken = ''
tokenType = ''
def token_provider():
if is_expired(bearerToken):
bearerToken, tokenType = generate_bearer_token('<YOUR_CREDENTIALS_FILE_PATH>')
return bearerToken, tokenType
try:
accessToken, tokenType = token_provider()
print("Access Token:", accessToken)
print("Type of token:", tokenType)
except SkyflowError as e:
print(e)
Vault APIs
The Vault python module is used to perform operations on the vault such as inserting records, detokenizing tokens, retrieving tokens for a skyflow_id and to invoke a connection.
To use this module, the skyflow client must first be initialized as follows.
from skyflow.vault import Client, Configuration
from skyflow.service_account import generate_bearer_token, is_expired
# cache for reuse
bearerToken = ''
# User defined function to provide access token to the vault apis
def token_provider():
if is_expired(bearerToken):
return bearerToken
bearerToken, _ = generate_bearer_token('<YOUR_CREDENTIALS_FILE_PATH>')
return bearerToken
#Initializing a Skyflow Client instance with a SkyflowConfiguration object
config = Configuration('<YOUR_VAULT_ID>', '<YOUR_VAULT_URL>', token_provider)
client = Client(config)
All Vault APIs must be invoked using a client instance.
Insert
To insert data into the vault use the insert(records: dict, options: InsertOptions) method. The records parameter is a dictionary that must have a records
key which has an array of records to be inserted into the vault as it's value. The options parameter takes a Skyflow.InsertOptions object, as shown below:
from skyflow.vault import InsertOptions
from skyflow.errors import SkyflowError
#Initialize Client
try:
options = InsertOptions(True) #indicates whether or not tokens should be returned for the inserted data. Defaults to 'True'
data = {
"records": [
{
"table": "<TABLE_NAME>",
"fields": {
"<FIELDNAME>": "<VALUE>"
}
}
]
}
response = client.insert(data, options=options)
print('Response:', response)
except SkyflowError as e:
print('Error Occurred:', e)
An example of an insert call is given below:
client.insert(
{
"records": [
{
"table": "cards",
"fields": {
"cardNumber": "41111111111",
"cvv": "123",
},
}
]
},
InsertOptions(True),
)
Sample response :
{
"records": [
{
"table": "cards",
"fields": {
"cardNumber": "f3907186-e7e2-466f-91e5-48e12c2bcbc1",
"cvv": "1989cb56-63da-4482-a2df-1f74cd0dd1a5",
},
}
]
}
Detokenize
In order to retrieve data from your vault using tokens that you have previously generated for that data, you can use the detokenize(records: dict) method. The records parameter takes a dictionary that contains the records
key that takes an array of records to be fetched from the vault as shown below.
{
"records":[
{
"token": str #token for the record to be fetched
}
]
}
An example of a detokenize call:
try:
client.detokenize(
{
"records": [
{"token": "45012507-f72b-4f5c-9bf9-86b133bae719"},
{'token': 'invalid-token'}
]
}
)
except SkyflowError as e:
if e.data:
print(e.data) # see note below
else:
print(e)
Sample response:
{
"records": [
{
"token": "131e70dc-6f76-4319-bdd3-96281e051051",
"value": "1990-01-01"
}
],
"errors": [
{
"token": "invalid-token",
"error": {
"code": 404,
"description": "Tokens not found for invalid-token"
}
}
]
}
Get By Id
For retrieving using SkyflowID's, use the get_by_id(records: dict) method. The records parameter takes a Dictionary that contains records to be fetched as shown below:
{
"records": [
{
"ids": [str], # List of SkyflowID's of the records to be fetched
"table": str, # name of table holding the above skyflow_id's
"redaction": Skyflow.RedactionType, # redaction to be applied to retrieved data
}
]
}
There are 4 accepted values in Skyflow.RedactionTypes:
PLAIN_TEXT
MASKED
REDACTED
DEFAULT
An example of get_by_id call:
from skyflow.vault import RedactionType
skyflowIDs = [
"f8d8a622-b557-4c6b-a12c-c5ebe0b0bfd9",
"da26de53-95d5-4bdb-99db-8d8c66a35ff9"
]
record = {"ids": skyflowIDs, "table": "cards", "redaction": RedactionType.PLAIN_TEXT}
invalidID = ["invalid skyflow ID"]
badRecord = {"ids": invalidID, "table": "cards", "redaction": RedactionType.PLAIN_TEXT}
records = {"records": [record, badRecord]}
try:
client.get_by_id(records)
except SkyflowError as e:
if e.data:
print(e.data) # see note below
else:
print(e)
Sample response:
{
"records": [
{
"fields": {
"card_number": "4111111111111111",
"cvv": "127",
"expiry_date": "11/35",
"fullname": "myname",
"skyflow_id": "f8d8a622-b557-4c6b-a12c-c5ebe0b0bfd9"
},
"table": "cards"
},
{
"fields": {
"card_number": "4111111111111111",
"cvv": "317",
"expiry_date": "10/23",
"fullname": "sam",
"skyflow_id": "da26de53-95d5-4bdb-99db-8d8c66a35ff9"
},
"table": "cards"
}
],
"errors": [
{
"error": {
"code": "404",
"description": "No Records Found"
},
"skyflow_ids": ["invalid skyflow id"]
}
]
}
Note:
While using detokenize and get_by_id methods, there is a possibility that some or all of the tokens might be invalid. In such cases, the data from response consists of both errors and detokenized records. In the SDK, this will raise a SkyflowError Exception and you can retrieve the data from this Exception object as shown above.
Invoke Connection
Using Skyflow Connection, end-user applications can integrate checkout/card issuance flow with their apps/systems. To invoke connection, use the invoke_connection(config: Skyflow.ConnectionConfig) method of the Skyflow client.
config = ConnectionConfig(
connectionURL: str, # connection url received when creating a skyflow connection integration
methodName: Skyflow.RequestMethod,
pathParams: dict, # optional
queryParams: dict, # optional
requestHeader: dict, # optional
requestBody: dict, # optional
)
client.invokeConnection(config)
methodName
supports the following methods:
- GET
- POST
- PUT
- PATCH
- DELETE
pathParams, queryParams, requestHeader, requestBody are the JSON objects represented as dictionaries that will be sent through the connection integration url.
An example of invoke_connection:
from skyflow.vault import ConnectionConfig, Configuration, RequestMethod
bearerToken = ''
def token_provider():
if is_expired(bearerToken):
return bearerToken
bearerToken, _ = generate_bearer_token('<YOUR_CREDENTIALS_FILE_PATH>')
return bearerToken
try:
config = Configuration('<YOUR_VAULT_ID>', '<YOUR_VAULT_URL>', token_provider)
connectionConfig = ConnectionConfig('<YOUR_CONNECTION_URL>', RequestMethod.POST,
requestHeader={
'Content-Type': 'application/json',
'Authorization': '<YOUR_CONNECTION_BASIC_AUTH>'
},
requestBody= # For third party integration
{
"expirationDate": {
"mm": "12",
"yy": "22"
}
},
pathParams={'cardID': '<CARD_VALUE>'}) # param as in the example
client = Client(config)
response = client.invoke_connection(connectionConfig)
print('Response:', response)
except SkyflowError as e:
print('Error Occurred:', e)
Sample response:
{
"receivedTimestamp": "2021-11-05 13:43:12.534",
"processingTimeinMs": 12,
"resource": {
"cvv2": "558"
}
}
Logging
The skyflow python SDK provides useful logging using python's inbuilt logging
library. By default the logging level of the SDK is set to LogLevel.ERROR
. This can be changed by using set_log_level(logLevel)
as shown below:
import logging
from skyflow import set_log_level, LogLevel
logging.basicConfig() # You can set the basic config here
set_log_level(LogLevel.INFO) # sets the skyflow SDK log level to INFO
Current the following 5 log levels are supported:
-
DEBUG
:When
LogLevel.DEBUG
is passed, all level of logs will be printed(DEBUG, INFO, WARN, ERROR) -
INFO
:When
LogLevel.INFO
is passed, INFO logs for every event that has occurred during the SDK flow execution will be printed along with WARN and ERROR logs -
WARN
:When
LogLevel.WARN
is passed, WARN and ERROR logs will be printed -
ERROR
:When
LogLevel.ERROR
is passed, only ERROR logs will be printed. -
OFF
:LogLevel.OFF
can be used to turn off all logging from the Skyflow SDK.
Note
: The ranking of logging levels is as follows : DEBUG
< INFO
< WARN
< ERROR
< OFF