Skip to main content

# DynamoQuery

Project description

DynamoQuery

PyPI - dynamoquery PyPI - Python Version Coverage

Helper for building Boto3 DynamoDB queries.

Installation

python -m pip install dynamoquery

Usage

You can find commented usage examples in examples directory.

DynamoQuery

import boto3

from dynamo_query import DynamoQuery, DataTable

table_resource = boto3.resource("dynamodb").Table('people')
query = DynamoQuery.build_scan(
    filter_expression=ConditionExpression('first_name') & ('last_name', 'in'),
).limit(
    5,
).projection(
    'first_name', 'last_name', 'age',
).table(
    table_resource=table_resource,
    table_keys=('pk', ),
)
...

# simple query
data = {
    'first_name': 'John',
    'last_name': ['Cena', 'Doe', 'Carmack'],
}

result_data_table = query.execute_dict(data)
for record in result_data_table.get_records():
    print(record)

# batch get
data_table = DataTable.create().add_record(
    {"pk": "my_pk"},
    {"pk": "my_pk2"},
    {"pk": "my_pk3"},
)

result_data_table = query.execute(data_table)
for record in result_data_table.get_records():
    print(record)

DynamoTable

from typing_extensions import TypedDict
from dynamo_query import DynamoTable

# first, define your record
class UserRecord(TypedDict, total=False):
    pk: str
    email: str
    name: str
    points: int


# Create your dynamo table manager with your record class
class UserTable(DynamoTable[UserRecord]):
    # provide a set of your table keys
    table_keys = {'pk'}

    # use this property to define your table name
    @property
    def table(self) -> str:
        return "my_table"

    # define how to get PK from a record
    def get_partition_key(self, record: UserRecord) -> str:
        return record["email"]

    # we do not have a sort key in our table
    def get_sort_key(self, record: UserRecord) -> None:
        return None

# okay, let's start using our manager
user_table = UserTable()

# add a new record to your table
user_table.upsert_record({
    "email": "user@gmail.com",
    "name": "John User",
    "age": 12,
})

# and output all the records
for user_record in user_table.scan():
    print(user_record)

Development

Install dependencies with pipenv

python -m pip install pipenv
pipenv install -d

# generate boto3 stubs index
python -m mypy_boto3

Enable pylint and mypy checks in your IDE.

Run unit tests and linting.

./scripts/before_commit.sh

Add false-positive unused entities to vulture whitelist

vulture dynamo_query --make-whitelist > vulture_whitelist.txt

VSCode

Recommended .vscode/settings.json

{
  "python.pythonPath": "<pipenv_path>/bin/python",
  "python.linting.pylintEnabled": true,
  "python.linting.enabled": true,
  "python.linting.mypyEnabled": true,
  "python.formatting.provider": "black"
}

PyCharm

  • Install mypy unofficial extension
  • Install blackextension, enable format on save
  • Run pylint on save

Versioning

dynamo_query version follows Semantic Versioning.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for dynamoquery, version 1.0.0
Filename, size File type Python version Upload date Hashes
Filename, size dynamoquery-1.0.0-py3-none-any.whl (37.7 kB) File type Wheel Python version py3 Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page