add layercake
This commit is contained in:
400
layercake/layercake/dynamodb.py
Normal file
400
layercake/layercake/dynamodb.py
Normal file
@@ -0,0 +1,400 @@
|
||||
from datetime import datetime
|
||||
from ipaddress import IPv4Address
|
||||
from typing import Any
|
||||
|
||||
from aws_lambda_powertools import Logger
|
||||
from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
logger = Logger(__name__)
|
||||
|
||||
|
||||
def _serialize(v):
|
||||
if isinstance(v, datetime):
|
||||
return v.isoformat()
|
||||
|
||||
if isinstance(v, IPv4Address):
|
||||
return str(v)
|
||||
|
||||
if isinstance(v, (list, tuple)):
|
||||
return [_serialize(x) for x in v]
|
||||
|
||||
if isinstance(v, dict):
|
||||
return {k: _serialize(dv) for k, dv in v.items()}
|
||||
|
||||
return v
|
||||
|
||||
|
||||
def serialize(obj: dict) -> dict:
|
||||
return {k: TypeSerializer().serialize(_serialize(v)) for k, v in obj.items()}
|
||||
|
||||
|
||||
def deserialize(obj: dict) -> dict:
|
||||
return {k: TypeDeserializer().deserialize(v) for k, v in obj.items()}
|
||||
|
||||
|
||||
def Key(pk: str, sk: str) -> dict[str, str]:
|
||||
return {
|
||||
'id': pk,
|
||||
'sk': sk,
|
||||
}
|
||||
|
||||
|
||||
class TransactItems:
|
||||
"""
|
||||
Documentation:
|
||||
--------------
|
||||
- https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_TransactWriteItems.html
|
||||
- https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_TransactGetItems.html
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.transact_write_items
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.transact_get_items
|
||||
"""
|
||||
|
||||
def __init__(self, table_name: str) -> None:
|
||||
self.table_name = table_name
|
||||
self.items: list[dict] = []
|
||||
|
||||
def put(
|
||||
self,
|
||||
*,
|
||||
item: dict,
|
||||
table_name: str | None = None,
|
||||
cond_expr: str | None = None,
|
||||
) -> None:
|
||||
attrs: dict = {}
|
||||
|
||||
if cond_expr:
|
||||
attrs['ConditionExpression'] = cond_expr
|
||||
|
||||
if not table_name:
|
||||
table_name = self.table_name
|
||||
|
||||
self.items.append(
|
||||
dict(
|
||||
Put=dict(
|
||||
TableName=table_name,
|
||||
Item=serialize(item),
|
||||
**attrs,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
def update(
|
||||
self,
|
||||
*,
|
||||
key: dict,
|
||||
update_expr: str,
|
||||
cond_expr: str | None = None,
|
||||
table_name: str | None = None,
|
||||
expr_attr_names: dict = {},
|
||||
expr_attr_values: dict = {},
|
||||
) -> None:
|
||||
attrs: dict = {}
|
||||
|
||||
if cond_expr:
|
||||
attrs['ConditionExpression'] = cond_expr
|
||||
|
||||
if expr_attr_names:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_names
|
||||
|
||||
if expr_attr_values:
|
||||
attrs['ExpressionAttributeValues'] = serialize(expr_attr_values)
|
||||
|
||||
if not table_name:
|
||||
table_name = self.table_name
|
||||
|
||||
self.items.append(
|
||||
dict(
|
||||
Update=dict(
|
||||
TableName=table_name,
|
||||
Key=serialize(key),
|
||||
UpdateExpression=update_expr,
|
||||
**attrs,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
def get(
|
||||
self,
|
||||
*,
|
||||
table_name: str | None = None,
|
||||
key: dict,
|
||||
expr_attr_names: str | None = None,
|
||||
) -> None:
|
||||
attrs: dict = {}
|
||||
|
||||
if expr_attr_names:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_names
|
||||
|
||||
if not table_name:
|
||||
table_name = self.table_name
|
||||
|
||||
self.items.append(
|
||||
dict(
|
||||
Get=dict(
|
||||
TableName=table_name,
|
||||
Key=serialize(key),
|
||||
**attrs,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
def delete(
|
||||
self,
|
||||
*,
|
||||
key: dict,
|
||||
table_name: str | None = None,
|
||||
cond_expr: str | None = None,
|
||||
expr_attr_names: dict = {},
|
||||
expr_attr_values: dict = {},
|
||||
) -> None:
|
||||
attrs: dict = {}
|
||||
|
||||
if cond_expr:
|
||||
attrs['ConditionExpression'] = cond_expr
|
||||
|
||||
if expr_attr_names:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_names
|
||||
|
||||
if expr_attr_values:
|
||||
attrs['ExpressionAttributeValues'] = serialize(expr_attr_values)
|
||||
|
||||
if not table_name:
|
||||
table_name = self.table_name
|
||||
|
||||
self.items.append(
|
||||
dict(
|
||||
Delete=dict(
|
||||
TableName=table_name,
|
||||
Key=serialize(key),
|
||||
**attrs,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
def condition(
|
||||
self,
|
||||
*,
|
||||
key: dict,
|
||||
cond_expr: str,
|
||||
table_name: str | None = None,
|
||||
expr_attr_names: dict = {},
|
||||
expr_attr_values: dict = {},
|
||||
) -> None:
|
||||
attrs: dict = {'ConditionExpression': cond_expr}
|
||||
|
||||
if expr_attr_names:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_names
|
||||
|
||||
if expr_attr_values:
|
||||
attrs['ExpressionAttributeValues'] = serialize(expr_attr_values)
|
||||
|
||||
if not table_name:
|
||||
table_name = self.table_name
|
||||
|
||||
self.items.append(
|
||||
dict(
|
||||
ConditionCheck=dict(
|
||||
TableName=table_name,
|
||||
Key=serialize(key),
|
||||
**attrs,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class DynamoDBPersistenceLayer:
|
||||
"""
|
||||
Documentation:
|
||||
--------------
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.query
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.get_item
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.put_item
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.update_item
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.delete_item
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.transact_get_items
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.transact_write_items
|
||||
"""
|
||||
|
||||
def __init__(self, table_name: str, dynamodb_client) -> None:
|
||||
self.table_name = table_name
|
||||
self.dynamodb_client = dynamodb_client
|
||||
|
||||
def query(
|
||||
self,
|
||||
*,
|
||||
key_cond_expr: str,
|
||||
expr_attr_name: dict = {},
|
||||
expr_attr_values: dict = {},
|
||||
start_key: dict = {},
|
||||
filter_expr: str | None = None,
|
||||
limit: int | None = None,
|
||||
index_forward: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""You must provide the name of the partition key attribute and a single value for that attribute.
|
||||
|
||||
Query returns all items with that partition key value.
|
||||
Optionally, you can provide a sort key attribute and use a comparison operator to refine the search results.
|
||||
|
||||
...
|
||||
|
||||
A `Query` operation always returns a result set. If no matching items are found, the result set will be empty.
|
||||
Queries that do not return results consume the minimum number of read capacity units for that type of read operation.
|
||||
|
||||
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/client/query.html
|
||||
"""
|
||||
attrs: dict = {'ScanIndexForward': index_forward}
|
||||
|
||||
if expr_attr_name:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_name
|
||||
|
||||
if expr_attr_values:
|
||||
attrs['ExpressionAttributeValues'] = serialize(expr_attr_values)
|
||||
|
||||
if start_key:
|
||||
attrs['ExclusiveStartKey'] = start_key
|
||||
|
||||
if filter_expr:
|
||||
attrs['FilterExpression'] = filter_expr
|
||||
|
||||
if limit:
|
||||
attrs['Limit'] = limit
|
||||
|
||||
try:
|
||||
response = self.dynamodb_client.query(
|
||||
TableName=self.table_name,
|
||||
KeyConditionExpression=key_cond_expr,
|
||||
**attrs,
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return dict(
|
||||
items=[deserialize(v) for v in response.get('Items', [])],
|
||||
last_key=response.get('LastEvaluatedKey', None),
|
||||
)
|
||||
|
||||
def get_item(self, key: dict) -> dict:
|
||||
"""The GetItem operation returns a set of attributes for the item with the given primary key.
|
||||
If there is no matching item, GetItem does not return any data and there will be no Item element in the response.
|
||||
"""
|
||||
|
||||
try:
|
||||
response = self.dynamodb_client.get_item(
|
||||
TableName=self.table_name,
|
||||
Key=serialize(key),
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return deserialize(response.get('Item', {}))
|
||||
|
||||
def put_item(self, item: dict, *, cond_expr: str | None = None) -> bool:
|
||||
attrs = {}
|
||||
|
||||
if cond_expr:
|
||||
attrs['ConditionExpression'] = cond_expr
|
||||
|
||||
try:
|
||||
self.dynamodb_client.put_item(
|
||||
TableName=self.table_name,
|
||||
Item=serialize(item),
|
||||
**attrs,
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return True
|
||||
|
||||
def update_item(
|
||||
self,
|
||||
key: dict,
|
||||
*,
|
||||
update_expr: str,
|
||||
cond_expr: str | None = None,
|
||||
expr_attr_names: dict | None = None,
|
||||
expr_attr_values: dict | None = None,
|
||||
) -> bool:
|
||||
attrs: dict = {}
|
||||
|
||||
if cond_expr:
|
||||
attrs['ConditionExpression'] = cond_expr
|
||||
|
||||
if expr_attr_names:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_names
|
||||
|
||||
if expr_attr_values:
|
||||
attrs['ExpressionAttributeValues'] = serialize(expr_attr_values)
|
||||
|
||||
try:
|
||||
self.dynamodb_client.update_item(
|
||||
TableName=self.table_name,
|
||||
Key=serialize(key),
|
||||
UpdateExpression=update_expr,
|
||||
**attrs,
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return True
|
||||
|
||||
def delete_item(
|
||||
self,
|
||||
key: dict,
|
||||
*,
|
||||
cond_expr: str | None = None,
|
||||
expr_attr_names: dict | None = None,
|
||||
expr_attr_values: dict | None = None,
|
||||
) -> bool:
|
||||
"""Deletes a single item in a table by primary key. You can perform a conditional delete operation that deletes
|
||||
the item if it exists, or if it has an expected attribute value.
|
||||
"""
|
||||
attrs: dict = {}
|
||||
|
||||
if cond_expr:
|
||||
attrs['ConditionExpression'] = cond_expr
|
||||
|
||||
if expr_attr_names:
|
||||
attrs['ExpressionAttributeNames'] = expr_attr_names
|
||||
|
||||
if expr_attr_values:
|
||||
attrs['ExpressionAttributeValues'] = serialize(expr_attr_values)
|
||||
|
||||
try:
|
||||
self.dynamodb_client.delete_item(
|
||||
TableName=self.table_name, Key=serialize(key), **attrs
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return True
|
||||
|
||||
def transact_get_items(self, transact_items: TransactItems) -> list[dict]:
|
||||
try:
|
||||
response = self.dynamodb_client.transact_get_items(
|
||||
TransactItems=transact_items.items
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return [
|
||||
deserialize(response.get('Item', {}))
|
||||
for response in response.get('Responses', [])
|
||||
]
|
||||
|
||||
def transact_write_items(self, transact_items: TransactItems) -> bool:
|
||||
try:
|
||||
self.dynamodb_client.transact_write_items(
|
||||
TransactItems=transact_items.items
|
||||
)
|
||||
except ClientError as err:
|
||||
logger.exception(err)
|
||||
raise
|
||||
else:
|
||||
return True
|
||||
Reference in New Issue
Block a user