Skip to content

Instantly share code, notes, and snippets.

@ryansb
Created June 30, 2021 16:10
Show Gist options
  • Save ryansb/6aee2f4f39a2e4fca958b5414b4b9262 to your computer and use it in GitHub Desktop.
Save ryansb/6aee2f4f39a2e4fca958b5414b4b9262 to your computer and use it in GitHub Desktop.
Boto3 DynamoDB client with default table names and serialization/deserialization
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