class Response[source]

Response(body:Optional[str], statusCode:int=200, headers:dict=<factory>)

parse response from apigateway

class Event[source]

Event(body:str, httpMethod:Optional[str]=None, multiValueHeaders:Optional[dict]=None, multiValueQueryStringParameters:Optional[dict]=None, path:Optional[str]=None, pathParameters:Optional[dict]=None, queryStringParameters:Optional[dict]=None, requestContext:Optional[dict]=None, headers:dict=<factory>, statusCode:int=200, isBase64Encoded:bool=False)

parse event from apigateway

class Product[source]

Product(cprcode:str, iprcode:str, oprcode:str, ordertype:str, pr_abb:str)

Product(cprcode: str, iprcode: str, oprcode: str, ordertype: str, pr_abb: str)

class Products[source]

Products(products:List[Product])

Products(products: List[awsSchema.apigateway.Product])

Examples

event = Event(body = json.dumps({'test':'test'})).to_dict()
print(event)

{'body': '{"test":"test"}', 'httpMethod': None, 'multiValueHeaders': None, 'multiValueQueryStringParameters': None, 'path': None, 'pathParameters': None, 'queryStringParameters': None, 'requestContext': None, 'headers': {}, 'statusCode': 200, 'isBase64Encoded': False}

Event.parseBody(event)

{'test': 'test'}

Returning response

success

Response.returnSuccess(body = {'success':'true'})

{'body': '{"success":"true"}',
 'statusCode': 200,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

Response.returnSuccess(statusCode = 206, body = {'success':'true'})

{'body': '{"success":"true"}',
 'statusCode': 206,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

Response.returnSuccess(body = None)

{'body': 'null',
 'statusCode': 200,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

Response.returnSuccess(body = {})

{'body': '{}',
 'statusCode': 200,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

Error

Response.returnError(statusCode = 402, message='error')

{'body': '{"error":"error"}',
 'statusCode': 402,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

responseDict = Response.returnSuccess(body = {'success':'true'}) Response.fromDict(responseDict)

event = Event.getInput(body={'input':'123'}, queryStringParameters={'test':'123'})
event

{'body': '{"input":"123"}',
 'httpMethod': None,
 'multiValueHeaders': None,
 'multiValueQueryStringParameters': None,
 'path': None,
 'pathParameters': None,
 'queryStringParameters': {'test': '123'},
 'requestContext': None,
 'headers': {},
 'statusCode': 200,
 'isBase64Encoded': False}

Parsing requests

with open('./testData/sampleEvent.json') as f:
  sampleEvent = json.load(f)['input']

get inidividual features

body

body = Event.parseBody(sampleEvent)
assert body == json.loads(Event.from_dict(sampleEvent).body)
print(body)

{'a': 1}

headers = Event.parseHeaders(sampleEvent)
assert headers == Event.from_dict(sampleEvent).headers
print(headers)

{'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'cache-control': 'no-cache', 'CloudFront-Forwarded-Proto': 'https', 'CloudFront-Is-Desktop-Viewer': 'true', 'CloudFront-Is-Mobile-Viewer': 'false', 'CloudFront-Is-SmartTV-Viewer': 'false', 'CloudFront-Is-Tablet-Viewer': 'false', 'CloudFront-Viewer-Country': 'US', 'Content-Type': 'application/json', 'headerName': 'headerValue', 'Host': 'gy415nuibc.execute-api.us-east-1.amazonaws.com', 'Postman-Token': '9f583ef0-ed83-4a38-aef3-eb9ce3f7a57f', 'User-Agent': 'PostmanRuntime/2.4.5', 'Via': '1.1 d98420743a69852491bbdea73f7680bd.cloudfront.net (CloudFront)', 'X-Amz-Cf-Id': 'pn-PWIJc6thYnZm5P0NMgOUglL1DYtl0gdeJky8tqsg8iS_sgsKD1A==', 'X-Forwarded-For': '54.240.196.186, 54.182.214.83', 'X-Forwarded-Port': '443', 'X-Forwarded-Proto': 'https'}

query = Event.parseQuery(sampleEvent)
assert query ==Event.from_dict(sampleEvent).queryStringParameters
print(query)

{'name': 'me', 'multivalueName': 'me'}

path = Event.parsePath(sampleEvent)
assert path == Event.from_dict(sampleEvent).path
print(path)

/hello/world

Parsing data class directly

productDict = Product(
  cprcode='123',
  iprcode='123',
  oprcode= '12343',
  ordertype='3225',
  pr_abb='4563'
).to_dict()
event = Event.getInput(productDict)
Event.parseDataClass(Product, event)

Product(cprcode='123', iprcode='123', oprcode='12343', ordertype='3225', pr_abb='4563')

import re
from beartype import beartype
@beartype
def increaseVersion(string:str)->str:
  '''
  increase version number of the pip package by 1
  '''
  match = next(re.finditer(r'\d.\d.\d*', string))
  lastDigits = int(match.group().split('.')[-1])
  newSubVersionList = match.group().split('.')
  newSubVersionList[-1] = str(lastDigits + 1)
  newVersionString = '.'.join(newSubVersionList)
  wholeString = re.sub(r'\d.\d.\d*',newVersionString,string)
  return wholeString

@beartype
def loadFile(fileName:str= './settings.ini')->str:
  with open(fileName, 'r') as f:
    file = f.read()
    return file
  
@beartype
def saveFile(data,fileName='./settings.ini.bak')->str:
  with open(fileName, 'w') as f:
    f.write(data)
    return data
  
@beartype
def replaceContent(originalContent:str)->str:
  newFile = ''
  for line in file.split('\n'):
    newFile += '\n'
    if 'version' in line:
      newFile += increaseVersion(line)
    else:
      newFile += line

def testIncreaseVersion():
  fileName = 'settings.ini'
  tempFileName = 'settings.ini.bak'
  fileString:str = loadFile(fileName)
  print('testing load file')
  assert fileString == open(fileName, 'r').read()
  print('testing save file')
  saveFile(fileString, fileName = tempFileName)
  assert fileString == open(tempFileName, 'r').read()
  print('test increase version')

# !cat settings.ini.bak
testIncreaseVersion()
      
    

testing load file
testing save file
test increase version

string = 'version = 0.0.14'
increaseVersion(string)

'version = 0.0.15'

APIGATEWAY schemas

Requests

import yaml
schema = {
    "resource": "Resource path",
    "path": "Path parameter",
    "httpMethod": "Incoming request's method name",
    "headers": "{String containing incoming request headers}",
    "multiValueHeaders": "{List of strings containing incoming request headers}",
    "queryStringParameters": "{query string parameters }",
    "multiValueQueryStringParameters": "{List of query string parameters}",
    "pathParameters":  "{path parameters}",
    "stageVariables": "{Applicable stage variables}",
    "requestContext": "{Request context, including authorizer-returned key-value pairs}",
    "body": "A JSON string of the request payload.",
    "isBase64Encoded": "A boolean flag to indicate if the applicable request payload is Base64-encoded",
}
print(yaml.dump(schema))

body: A JSON string of the request payload.
headers: '{String containing incoming request headers}'
httpMethod: Incoming request's method name
isBase64Encoded: A boolean flag to indicate if the applicable request payload is Base64-encoded
multiValueHeaders: '{List of strings containing incoming request headers}'
multiValueQueryStringParameters: '{List of query string parameters}'
path: Path parameter
pathParameters: '{path parameters}'
queryStringParameters: '{query string parameters }'
requestContext: '{Request context, including authorizer-returned key-value pairs}'
resource: Resource path
stageVariables: '{Applicable stage variables}'

Response

schema = {
    "isBase64Encoded": 'bool',
    "statusCode": 'int',
    "headers": { "headerName": "headerValue" },
    "multiValueHeaders": { "headerName": ["headerValue", "headerValue2"]   },
    "body": "..."
}
print(yaml.dump(schema))

body: '...'
headers:
  headerName: headerValue
isBase64Encoded: bool
multiValueHeaders:
  headerName:
  - headerValue
  - headerValue2
statusCode: int

Headers

class Headers[source]

Headers()

Headers.cors()

{'Access-Control-Allow-Headers': '*',
 'Access-Control-Allow-Origin': '*',
 'Access-Control-Allow-Methods': '*'}

Headers.cacheMaxAge()

{'Cache-Control': 'max-age=86400'}