dynamodb_patriQL_CRUD_statements

patriQL_CRUD.py

#%%
import boto3
client = boto3.client('dynamodb')
table_name  = ''

#%%
# INSERT INTO -- CREATE A RECORD WITH Multiple Values
#example args
primary_key = ''
secondary_key = ''
status = '' # value #1
#example Values dict
values_dict = {"primary_key":primary_key, "secondary_key":secondary_key, "status":status}

stmt = f"""INSERT INTO "{table_name}" value {values_dict}"""
response = client.execute_statement(Statement= stmt)
print(response)



#%%
# UPDATE -- UPDATE A RECORD
dynamodb_client = boto3.client('dynamodb')
tenant = {"email":"xx@xx.com",
          "password":"Password@1",
          "plan_id":"ssss",
          "invoice_item_uuid":"www",
          "uuid":"rrr"}
stmt = f'''
UPDATE "{table_name}"
SET email='{tenant['email']}'
SET plan_id='{tenant['plan_id']}'
SET invoice_item_uuid='{tenant['invoice_item_uuid']}'
SET uuid='{tenant['uuid']}'
WHERE client_id='{client_id}'
RETURNING ALL NEW *
'''
update_res = dynamodb_client.execute_statement(Statement= stmt)
#%%
# SELECT -- READ A RECORD
stmt = f"SELECT * FROM {table_name} where PK = '1'"
response = client.execute_statement(Statement= stmt)
print(response["Items"])

#%%
#DELETE -- DELETE A RECORD
stmt = f"DELETE FROM {table_name} WHERE PK = '1'"
response = client.execute_statement(Statement= stmt)
print(response)


### Unmarshall dynamodb
def unmarshall(dynamo_obj: dict) -> dict:
    """Convert a DynamoDB dict into a standard dict."""
    deserializer = TypeDeserializer()
    return {k: deserializer.deserialize(v) for k, v in dynamo_obj.items()}