Merge pull request #72 from skyflowapi/release/23.1.2.1

SK-243 added update and get interfaces

samples/get_sample.py

@@ -0,0 +1,43 @@
+'''
+	Copyright (c) 2022 Skyflow, Inc.
+'''
+from skyflow.errors import SkyflowError
+from skyflow.service_account import generate_bearer_token, is_expired
+from skyflow.vault import Client, Configuration, RedactionType
+
+
+# cache token for reuse
+bearerToken = ''
+
+
+def token_provider():
+    global bearerToken
+    if is_expired(bearerToken):
+        bearerToken, _ = generate_bearer_token('<YOUR_CREDENTIALS_FILE_PATH>')
+    return bearerToken
+
+
+try:
+    config = Configuration(
+        '<YOUR_VAULT_ID>', '<YOUR_VAULT_URL>', token_provider)
+    client = Client(config)
+
+    data = {"records": [
+        {
+            "ids": ["<SKYFLOW_ID1>", "<SKYFLOW_ID2>", "<SKYFLOW_ID3>"],
+            "table": "<TABLE_NAME>",
+            "redaction": RedactionType.PLAIN_TEXT
+        },
+        #To get records using unique column name and values.
+        {
+         "redaction" : "<REDACTION_TYPE>",
+         "table": "<TABLE_NAME>",
+         "columnName": "<UNIQUE_COLUMN_NAME>",
+         "columnValues": "[<COLUMN_VALUE_1>,<COLUMN_VALUE_2>]",
+        }
+    ]}
+
+    response = client.get(data)
+    print('Response:', response)
+except SkyflowError as e:
+    print('Error Occurred:', e)

samples/update_sample.py

@@ -0,0 +1,39 @@
+'''
+	Copyright (c) 2022 Skyflow, Inc.
+'''
+from skyflow.errors import SkyflowError
+from skyflow.service_account import generate_bearer_token, is_expired
+from skyflow.vault import Client, UpdateOptions, Configuration
+
+# cache token for reuse
+bearerToken = ''
+
+def token_provider():
+    global bearerToken
+    if is_expired(bearerToken):
+        bearerToken, _ = generate_bearer_token('<YOUR_CREDENTIALS_FILE_PATH>')
+    return bearerToken
+
+
+try:
+    config = Configuration(
+       '<YOUR_VAULT_ID>', '<YOUR_VAULT_URL>', token_provider)
+    client = Client(config)
+
+    options = UpdateOptions(True)
+
+    data = {
+        "records": [
+            {
+                "id": "<SKYFLOW_ID>",
+                "table": "<TABLE_NAME>",
+                "fields": {
+                    "<FIELDNAME>": "<VALUE>"
+                }
+            }
+        ]
+    }
+    response = client.update(data, options=options)
+    print('Response:', response)
+except SkyflowError as e:
+    print('Error Occurred:', e)

skyflow/_utils.py

@@ -68,13 +68,19 @@ class InfoMessages(Enum):
     IS_EXPIRED_TRIGGERED = "is_expired() triggered"
     EMPTY_ACCESS_TOKEN = "Give access token is empty"
     INVALID_TOKEN = "Given token is invalid"
+    UPDATE_TRIGGERED = "Update method triggered"
+    UPDATE_DATA_SUCCESS = "Data has been updated successfully"
+    GET_TRIGGERED = "Get triggered."
+    GET_SUCCESS = "Data fetched successfully."
 
 
 class InterfaceName(Enum):
     CLIENT = "client"
     INSERT = "client.insert"
     DETOKENIZE = "client.detokenize"
     GET_BY_ID = "client.get_by_id"
+    GET = "client.get"
+    UPDATE = "client.update"
     INVOKE_CONNECTION = "client.invoke_connection"
     GENERATE_BEARER_TOKEN = "service_account.generate_bearer_token"
 

skyflow/errors/_skyflow_errors.py

@@ -34,16 +34,20 @@ class SkyflowErrorMessages(Enum):
     FIELDS_KEY_ERROR = "Fields key is missing from payload"
     TABLE_KEY_ERROR = "Table key is missing from payload"
     TOKEN_KEY_ERROR = "Token key is missing from payload"
-    IDS_KEY_ERROR = "Ids key is missing from payload"
+    IDS_KEY_ERROR = "Id(s) key is missing from payload"
     REDACTION_KEY_ERROR = "Redaction key is missing from payload"
+    UNIQUE_COLUMN_OR_IDS_KEY_ERROR = "Ids or Unique column key is missing from payload"
+    UPDATE_FIELD_KEY_ERROR = "Atleast one field should be provided to update"
 
     INVALID_JSON = "Given %s is invalid JSON"
     INVALID_RECORDS_TYPE = "Records key has value of type %s, expected list"
-    INVALID_FIELDS_TYPE = "Fields key has value of type %s, expected string"
+    INVALID_FIELDS_TYPE = "Fields key has value of type %s, expected dict"
     INVALID_TABLE_TYPE = "Table key has value of type %s, expected string"
     INVALID_IDS_TYPE = "Ids key has value of type %s, expected list"
     INVALID_ID_TYPE = "Id key has value of type %s, expected string"
     INVALID_REDACTION_TYPE = "Redaction key has value of type %s, expected Skyflow.Redaction"
+    INVALID_COLUMN_NAME = "Column name has value of type %s, expected string"
+    INVALID_COLUMN_VALUE = "Column values has value of type %s, expected list"
 
     INVALID_REQUEST_BODY = "Given request body is not valid"
     INVALID_RESPONSE_BODY = "Given response body is not valid"

skyflow/vault/_client.py

@@ -4,11 +4,13 @@
 import types
 import requests
 from ._insert import getInsertRequestBody, processResponse, convertResponse
+from ._update import sendUpdateRequests, createUpdateResponseBody
 from ._config import Configuration
-from ._config import InsertOptions, ConnectionConfig
+from ._config import InsertOptions, ConnectionConfig, UpdateOptions
 from ._connection import createRequest
 from ._detokenize import sendDetokenizeRequests, createDetokenizeResponseBody
-from ._get_by_id import sendGetByIdRequests, createGetByIdResponseBody
+from ._get_by_id import sendGetByIdRequests, createGetResponseBody
+from ._get import sendGetRequests
 import asyncio
 from skyflow.errors._skyflow_errors import SkyflowError, SkyflowErrorCodes, SkyflowErrorMessages
 from skyflow._utils import log_info, InfoMessages, InterfaceName
@@ -78,6 +80,25 @@ def detokenize(self, records):
             log_info(InfoMessages.DETOKENIZE_SUCCESS.value, interface)
             return result
 
+    def get(self, records):
+        interface = InterfaceName.GET.value
+        log_info(InfoMessages.GET_TRIGGERED.value, interface)
+
+        self._checkConfig(interface)
+        self.storedToken = tokenProviderWrapper(
+            self.storedToken, self.tokenProvider, interface)
+        url = self._get_complete_vault_url()
+        responses = asyncio.run(sendGetRequests(
+            records, url, self.storedToken))
+        result, partial = createGetResponseBody(responses)
+        if partial:
+            raise SkyflowError(SkyflowErrorCodes.PARTIAL_SUCCESS,
+                               SkyflowErrorMessages.PARTIAL_SUCCESS, result, interface=interface)
+        else:
+            log_info(InfoMessages.GET_SUCCESS.value, interface)
+
+            return result
+
     def get_by_id(self, records):
         interface = InterfaceName.GET_BY_ID.value
         log_info(InfoMessages.GET_BY_ID_TRIGGERED.value, interface)
@@ -88,7 +109,7 @@ def get_by_id(self, records):
         url = self._get_complete_vault_url()
         responses = asyncio.run(sendGetByIdRequests(
             records, url, self.storedToken))
-        result, partial = createGetByIdResponseBody(responses)
+        result, partial = createGetResponseBody(responses)
         if partial:
             raise SkyflowError(SkyflowErrorCodes.PARTIAL_SUCCESS,
                                SkyflowErrorMessages.PARTIAL_SUCCESS, result, interface=interface)
@@ -130,3 +151,21 @@ def _get_complete_vault_url(self):
             Get the complete vault url from given vault url and vault id
         '''
         return self.vaultURL + "/v1/vaults/" + self.vaultID
+
+    def update(self, updateInput, options: UpdateOptions = UpdateOptions()):
+        interface = InterfaceName.UPDATE.value
+        log_info(InfoMessages.UPDATE_TRIGGERED.value, interface=interface)
+
+        self._checkConfig(interface)
+        self.storedToken = tokenProviderWrapper(
+            self.storedToken, self.tokenProvider, interface)
+        url = self._get_complete_vault_url()
+        responses = asyncio.run(sendUpdateRequests(
+            updateInput, options, url, self.storedToken))
+        result, partial = createUpdateResponseBody(responses)
+        if partial:
+            raise SkyflowError(SkyflowErrorCodes.PARTIAL_SUCCESS,
+                               SkyflowErrorMessages.PARTIAL_SUCCESS, result, interface=interface)
+        else:
+            log_info(InfoMessages.UPDATE_DATA_SUCCESS.value, interface)
+            return result

skyflow/vault/_config.py

@@ -34,6 +34,10 @@ def __init__(self, tokens: bool=True,upsert :List[UpsertOption]=None):
         self.tokens = tokens
         self.upsert = upsert
 
+class UpdateOptions:
+    def __init__(self, tokens: bool=True):
+        self.tokens = tokens
+
 class RequestMethod(Enum):
     GET = 'GET'
     POST = 'POST'

skyflow/vault/_get.py

@@ -0,0 +1,99 @@
+'''
+	Copyright (c) 2022 Skyflow, Inc.
+'''
+from skyflow.errors._skyflow_errors import SkyflowError, SkyflowErrorCodes, SkyflowErrorMessages
+import asyncio
+from aiohttp import ClientSession
+from ._config import RedactionType
+from skyflow._utils import InterfaceName
+from ._get_by_id import get
+
+interface = InterfaceName.GET.value
+
+def getGetRequestBody(data):
+    ids = None
+    if "ids" in data:
+        ids = data["ids"]
+        if not isinstance(ids, list):
+            idsType = str(type(ids))
+            raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT,
+                            SkyflowErrorMessages.INVALID_IDS_TYPE.value % (idsType), interface=interface)
+        for id in ids:
+            if not isinstance(id, str):
+                idType = str(type(id))
+                raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_ID_TYPE.value % (
+                    idType), interface=interface)
+    try:
+        table = data["table"]
+    except KeyError:
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT,
+                           SkyflowErrorMessages.TABLE_KEY_ERROR, interface=interface)
+    if not isinstance(table, str):
+        tableType = str(type(table))
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_TABLE_TYPE.value % (
+            tableType), interface=interface)
+    try:
+        redaction = data["redaction"]
+    except KeyError:
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT,
+                           SkyflowErrorMessages.REDACTION_KEY_ERROR, interface=interface)
+    if not isinstance(redaction, RedactionType):
+        redactionType = str(type(redaction))
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_REDACTION_TYPE.value % (
+            redactionType), interface=interface)
+
+    columnName = None
+    if "columnName" in data:   
+        columnName = data["columnName"] 
+        if not isinstance(columnName, str):
+            columnNameType = str(type(columnName))
+            raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_COLUMN_NAME.value % (
+            columnNameType), interface=interface)
+
+    columnValues = None
+    if columnName is not None and "columnValues" in data:   
+        columnValues = data["columnValues"]
+        if not isinstance(columnValues, list):
+            columnValuesType= str(type(columnValues))
+            raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_COLUMN_VALUE.value % (
+            columnValuesType), interface=interface)
+
+    if(ids is None and (columnName is None or columnValues is None)):
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, 
+        SkyflowErrorMessages.UNIQUE_COLUMN_OR_IDS_KEY_ERROR.value, interface= interface)
+    return ids, table, redaction.value, columnName, columnValues
+
+
+async def sendGetRequests(data, url, token):
+    tasks = []
+    try:
+        records = data["records"]
+    except KeyError:
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT,
+                           SkyflowErrorMessages.RECORDS_KEY_ERROR, interface=interface)
+    if not isinstance(records, list):
+        recordsType = str(type(records))
+        raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_RECORDS_TYPE.value % (
+            recordsType), interface=interface)
+
+    validatedRecords = []
+    for record in records:
+        ids, table, redaction, columnName, columnValues = getGetRequestBody(record)
+        validatedRecords.append((ids, table, redaction, columnName, columnValues))
+    async with ClientSession() as session:
+        for record in validatedRecords:
+            headers = {
+                "Authorization": "Bearer " + token
+            }
+            params = {"redaction": redaction}
+            if ids is not None:
+                params["skyflow_ids"] = ids
+            if columnName is not None:
+                params["column_name"] = columnName
+                params["column_values"] = columnValues
+            task = asyncio.ensure_future(
+                get(url, headers, params, session, record[1]))
+            tasks.append(task)
+        await asyncio.gather(*tasks)
+        await session.close()
+    return tasks

skyflow/vault/_get_by_id.py

@@ -85,7 +85,7 @@ async def get(url, headers, params, session, table):
             return (await response.read(), response.status, table)
 
 
-def createGetByIdResponseBody(responses):
+def createGetResponseBody(responses):
     result = {
         "records": [],
         "errors": []
@@ -114,6 +114,5 @@ def createGetByIdResponseBody(responses):
             if len(r) > 3 and r[3] != None:
                 temp["error"]["description"] += ' - Request ID: ' + str(r[3])
             result["errors"].append(temp)
-            result["errors"].append(temp)
             partial = True
     return result, partial

skyflow/vault/_insert.py

@@ -74,7 +74,7 @@ def getTableAndFields(record):
                            SkyflowErrorMessages.FIELDS_KEY_ERROR, interface=interface)
 
     if not isinstance(fields, dict):
-        fieldsType = str(type(table))
+        fieldsType = str(type(fields))
         raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.INVALID_FIELDS_TYPE.value % (
             fieldsType), interface=interface)