Created
June 30, 2021 16:10
-
-
Save ryansb/6aee2f4f39a2e4fca958b5414b4b9262 to your computer and use it in GitHub Desktop.
Boto3 DynamoDB client with default table names and serialization/deserialization
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto3 | |
from boto3.dynamodb.transform import copy_dynamodb_params, TransformationInjector | |
from boto3.dynamodb.conditions import Attr | |
def pretty_ddb(*args, default_table_name=None, **kwargs): | |
"""Builds a DynamoDB client with automatic serialization/deserialization. | |
Instead of `.get_item(...) -> {'key': 'S': 'Value', 'count': 'N': '9'}` this | |
changes the client to automatically serialize and deserialize as | |
`.get_item(...) -> {'key': 'Value', 'count': 9}` | |
default_table_name: A default to use if a table name is not passed to | |
operations such as GetItem, Query, etc | |
Other args/kwargs are passed to `boto3.client('dynamodb', *args, **kwargs)` | |
Implementation is similar to `boto3.dynamodb.transform` and the | |
`boto3.resource('dynamodb').meta.client` | |
""" | |
client = boto3.client("dynamodb", *args, **kwargs) | |
transformer = TransformationInjector() | |
def _set_table_name(params=None, model=None, **kwargs): | |
if not params or not model: | |
return | |
if "TableName" in model.input_shape.required_members: | |
if params.get("TableName"): | |
# if TableName is already set, skip it | |
return | |
params["TableName"] = default_table_name | |
return params | |
if default_table_name: | |
client.meta.events.register_first( | |
"provide-client-params.dynamodb", | |
_set_table_name, | |
unique_id="default-table-name", | |
) | |
client.meta.events.register( | |
"provide-client-params.dynamodb", | |
copy_dynamodb_params, | |
unique_id="dynamodb-create-params-copy", | |
) | |
# Apply the handler that generates condition expressions including | |
# placeholders. | |
client.meta.events.register( | |
"before-parameter-build.dynamodb", | |
transformer.inject_condition_expressions, | |
unique_id="dynamodb-condition-expression", | |
) | |
# Apply the handler that serializes the request from python | |
# types to dynamodb types. | |
client.meta.events.register( | |
"before-parameter-build.dynamodb", | |
transformer.inject_attribute_value_input, | |
unique_id="dynamodb-attr-value-input", | |
) | |
# Apply the handler that deserializes the response from dynamodb | |
# types to python types. | |
client.meta.events.register( | |
"after-call.dynamodb", | |
transformer.inject_attribute_value_output, | |
unique_id="dynamodb-attr-value-output", | |
) | |
return client | |
no_default_ddb = pretty_ddb(profile_name="some-profile") | |
print( | |
no_default_ddb.scan( | |
TableName="MyTableName", | |
FilterExpression=Attr("hash").begins_with("USER#"), | |
)["Items"][0] | |
) | |
my_ddb = pretty_ddb(default_table_name="MyTableName", region_name="us-east-1") | |
# calling non-TableName requiring args works normally | |
my_ddb.list_tables() | |
# calling a .scan with no TableName still works, and uses the default supplied | |
print(my_ddb.scan(FilterExpression=Attr("hash").begins_with("USER#"))["Items"][0]) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment