Imports

Error classes

class PynamoDBSavingError[source]

PynamoDBSavingError() :: Exception

Common base class for all non-exit exceptions.

class PynamoDBSchemaValidationError[source]

PynamoDBSchemaValidationError() :: Exception

Common base class for all non-exit exceptions.

SchemaAttribute class

a class which automatically parse and check data against json schema

class SchemaAttribute[source]

SchemaAttribute(*args, **kwds) :: Attribute

An attribute of a model

Dataclass Attribute

class DataclassJsonAttribute[source]

DataclassJsonAttribute(*args, **kwds) :: Attribute

An attribute of a model

from pynamodb.models import Model
from pynamodb.attributes import NumberAttribute
from dataclasses import dataclass
from dataclasses_json import dataclass_json
from nicHelper.dataclassUtil import enforce_types
import pytest

@enforce_types
@dataclass_json
@dataclass
class Product:
  code: str
  id_: int


class Table(Model):
  class Meta:
    table_name='test-please-delete'
    region_name='ap-southeast-1'
  id_= NumberAttribute(hash_key=True)
  product = DataclassJsonAttribute(customDataClass=Product)
  

Table.create_table(billing_mode='PAY_PER_REQUEST')


Table(id_=123,product={'code':'123','id_':123 }).save()

with pytest.raises(TypeError):
  Table(id_=123,product={'code':123,'id_':123 }).save()
  raise Exception('type error should occur')

Table.delete_table()

{'TableDescription': {'BillingModeSummary': {'BillingMode': 'PAY_PER_REQUEST',
   'LastUpdateToPayPerRequestDateTime': 1627143790.004},
  'ItemCount': 0,
  'ProvisionedThroughput': {'NumberOfDecreasesToday': 0,
   'ReadCapacityUnits': 0,
   'WriteCapacityUnits': 0},
  'TableArn': 'arn:aws:dynamodb:us-east-1:394922924679:table/test-please-delete',
  'TableId': 'a49c1723-d94e-4ef2-ae1d-e1dbde4d54d5',
  'TableName': 'test-please-delete',
  'TableSizeBytes': 0,
  'TableStatus': 'DELETING',
  'TableThroughputModeSummary': {'LastUpdateToPayPerRequestDateTime': 1627143790.004,
   'TableThroughputMode': 'PAY_PER_REQUEST'}}}

Supermodel

a class which add some functionalities on top of the standard pynamodb model, it sets id_ as the hash key and gives

• fromDict functions
• repr as a dict

class SuperModel[source]

SuperModel(hash_key:Optional[Any]=None, range_key:Optional[Any]=None, _user_instantiated:bool=True, **attributes:Any) :: Model

Defines a PynamoDB Model

This model is backed by a table in DynamoDB. You can create the table by with the create_table method.

Test

testSchema = 'https://gist.githubusercontent.com/thanakijwanavit/e2720d091ae0cef710a49b57c0c9cd4c/raw/ed2d322eac4900ee0f95b431d0f9067a40f3e0f0/squirrelOpenApiV0.0.3.yaml'
schema = yaml.load(requests.get(testSchema).text, Loader= yaml.Loader)
path = 'components/schemas/Location'
dpath.util.get(schema, path)

{'type': 'object',
 'required': ['id',
  'type',
  'street_address',
  'city',
  'state',
  'zip',
  'capacity',
  'status'],
 'properties': {'id': {'type': 'string', 'format': 'uuid'},
  'type': {'type': 'string', 'enum': ['pick up', 'drop off', 'overnight']},
  'street_address': {'type': 'string'},
  'city': {'type': 'string'},
  'state': {'type': 'string',
   'pattern': '^(?:(A[KLRZ]|C[AOT]|D[CE]|FL|GA|HI|I[ADLN]|K[SY]|LA|M[ADEINOST]|N[CDEHJMVY]|O[HKR]|PA|RI|S[CD]|T[NX]|UT|V[AT]|W[AIVY]))$'},
  'zip': {'type': 'string', 'pattern': '(^\\d{5}$)|(^\\d{5}-\\d{4}$)'},
  'status': {'type': 'string', 'enum': ['open', 'in use']},
  'created': {'type': 'string', 'format': 'date-time'},
  'modified': {'type': 'string', 'format': 'date-time'}}}

TestModel

schemaUrl = 'https://raw.githubusercontent.com/thanakijwanavit/villaMasterSchema/master/Product.json'
from typing import Any
class TestModel(SuperModel):
  class Meta:
    table_name="colab-test-sensitive-column"
    region = 'ap-southeast-1'
  data = SchemaAttribute(schemaUrl = schemaUrl, null=True)
  phoneHash = UnicodeAttribute(hash_key=True)
  
    
  # Overrides
  def pullOutKeys(self)->None:
    self.phoneHash = str(self.data.get('phoneHash') or self.data.get('iprcode') or self.data.get('id') )

  @beartype
  def toDict(self)->dict:
    return self.data
    
  @classmethod
  @beartype
  def fromDict(cls, inputDict:dict)->Any:
    return cls(data=inputDict)
    
  @beartype  
  def update(self,inputDict:dict)->None:
    self.data.update(inputDict)

d = TestModel('123', data={'iprcode': 4, 'cprcode': 123 , 'oprCode': '123', 'orderId': 123})
assert d.pullOutKeys() == None
assert type(d.toDict()) == dict
assert d.update({'cprcode': 234}) ==None

success

from nicHelper.exception import errorString
try:
  test = TestModel.fromDict({'iprcode': 4, 'cprcode': 123 , 'oprCode': '123', 'orderId': 123})
  test.save()
except Exception as e:
  print(e)
  print(errorString())


next(TestModel.query('1'))

{"data": {"type": "pick up", "street_address": "123", "id": "123", "city": "sth", "state": "CA", "zip": "23523", "capacity": 5, "status": "open"}}

next(TestModel.query('1'))

{"data": {"type": "pick up", "street_address": "123", "id": "123", "city": "sth", "state": "CA", "zip": "23523", "capacity": 5, "status": "open"}}

fail

try:
  TestModel(
    data = {'iprcode': '4', 'cprcode': 123 , 'oprCode': '123'}
  ).save()
except Exception as e:
  print(e)



next(TestModel.query('1'))

failed validation 
 '4' is not of type 'integer'

Failed validating 'type' in schema['properties']['iprcode']:
    {'type': 'integer'}

On instance['iprcode']:
    '4'

{"data": {"type": "pick up", "street_address": "123", "id": "123", "city": "sth", "state": "CA", "zip": "23523", "capacity": 5, "status": "open"}}

nested

schemaUrl = 'https://gist.githubusercontent.com/thanakijwanavit/e2720d091ae0cef710a49b57c0c9cd4c/raw/ed2d322eac4900ee0f95b431d0f9067a40f3e0f0/squirrelOpenApiV0.0.3.yaml'
path = '/components/schemas/Location'
class ProductModel(SuperModel):
  class Meta:
    table_name="colab-test-sensitive-column"
    region = 'ap-southeast-1'
  phoneHash = UnicodeAttribute(hash_key=True)
  data = SchemaAttribute(schemaUrl = schemaUrl,path=path, null=True)
  
  def pullOutKeys(self):
    self.phoneHash = self.data.get('id')

  
def test_nested():
  result = {}
  try:
    ProductModel(
      data = {'type': 'something invalid', 'street_address': '123' }
    ).save()
  except Exception as e:
    print('faulty data is rejected')
    result['errorModel'] = True

  try:
    data = {'type': 'pick up', 'street_address': '123' , 'id': '123', 'city':'sth', 'state': 'CA', 'zip':'23523', 'capacity':5, 'status':'open'}
    product:ProductModel = ProductModel.fromDict(data)
    result['successModel'] = True
  except Exception as e:
    print(f'valid data is rejected\n{e}')
    result['successModel'] = False
  

  assert next(TestModel.query('1')).data == {'type': 'pick up', 'street_address': '123' , 'id': '123', 'city':'sth', 'state': 'CA', 'zip':'23523', 'capacity':5, 'status':'open'}
  assert result['successModel'] == True, 'success model didnt save properly'
  assert result['errorModel'] == True, 'error model went through'
  
test_nested()

faulty data is rejected

Standard functions

Create data

createData[source]

createData(event:dict, hashKeyName:str, mainClass:Model, schemaUrl:Optional[str]=None, schemaFormat:str='yaml', *args)

create a new row of data

Usage example

def create (event, *args):
  body = Event.parseBody(event)
  body['id'] = body['phoneHash']
  
  event2 = Event.getInput(body)
  r = createData(event2, hashKeyName='phoneHash', mainClass=TestModel)
  if r.get('statusCode') != 200: return r
  r2 = next(TestModel.query(body['phoneHash']), None)
  if not r2: return Response.returnError('st wrong with saving, saving but didnt go through')
  return Response.returnSuccess(r2)

Test

Success

schemaUrl = 'https://raw.githubusercontent.com/thanakijwanavit/villaMasterSchema/master/Product.json'
data = {'phoneHash': '123','iprcode': 4, 'cprcode': 123 , 'oprCode': '123'}
event = Event.getInput(data)
item = next(TestModel.queryId('123'), None)
print('existing item is :',item)
# delete item if exist
if item:
  print(item.delete())
create(event)

existing item is : {"lastEdited": 1627176860.619993, "creationTime": 1627176860.521518, "data": {"phoneHash": "123", "iprcode": 5, "cprcode": 123, "oprCode": "1234", "id": "123"}, "phoneHash": "123"}
{'ConsumedCapacity': {'CapacityUnits': 1.0, 'TableName': 'colab-test-sensitive-column'}}

{'body': '{"phoneHash":"123","iprcode":4,"cprcode":123,"oprCode":"123","id":"123"}',
 'statusCode': 200,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

createData(Event.getInput(data), hashKeyName='phoneHash', mainClass =TestModel)
next(TestModel.query('123'))

{"lastEdited": 1627188189.149829, "creationTime": 1627188189.149844, "data": {"phoneHash": "123", "iprcode": 4, "cprcode": 123, "oprCode": "123", "id": "123"}}

test = TestModel(data = data)
test.save()
test.phoneHash
next(TestModel.query('123'))

{"lastEdited": 1627188189.190334, "creationTime": 1627188189.190346, "data": {"phoneHash": "123", "iprcode": 4, "cprcode": 123, "oprCode": "123"}}

GetData

getData[source]

getData(hashKeyName:str, mainClass:Model)

create a new basket

usage

def lambdaGet(event, *args):
  query = Event.parseBody(event)
  if 'key' not in query: return Response.returnError(f'missing key')
  return getData(query['key'], TestModel)

Test

Success

data = {'phoneHash': '123','iprcode': 4, 'cprcode': 123 , 'oprCode': '123'}
event = Event.getInput(data)
create(event)

lambdaGet(Event.getInput({'key': '123'}))

{'body': '{"phoneHash":"123","iprcode":4,"cprcode":123,"oprCode":"123"}',
 'statusCode': 200,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

failed wrong key type

lambdaGet(Event.getInput({'key': 123}))

{'body': '{"error":"failed to query with error object of type \'int\' has no len()"}',
 'statusCode': 400,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}

update data

updateData[source]

updateData(event:dict, hashKeyName:str, mainClass:Model, schemaUrl:Optional[str]=None, schemaFormat:str='yaml', *args)

updating data based on the new input

event:dict: object gathered from apiGatewayProxy hashKeyName:str: the name of the hash key mainClass:Model: pynamodb model class for this object schemaUrl:Optional[str]: url of the input schema for validation schemaFormat:Enum['yaml', 'json']

sample Usage

def update(event, *args):
  body = Event.parseBody(event)
  body['id'] = body['phoneHash']
  
  event2 = Event.getInput(body)
  hashKeyname = 'id'
  return updateData(event2, hashKeyName=hashKeyname, mainClass=TestModel)

r = create(Event.getInput({'phoneHash': '123','iprcode': 5, 'cprcode': 123 , 'oprCode': '123'}))
r = update(Event.getInput({'phoneHash': '123','iprcode': 5, 'cprcode': 123 , 'oprCode': '1234'}))
lambdaGet(Event.getInput({'key':'123'}))

{'body': '{"phoneHash":"123","iprcode":5,"cprcode":123,"oprCode":"1234","id":"123"}',
 'statusCode': 200,
 'headers': {'Access-Control-Allow-Headers': '*',
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': '*'}}