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 import Optional
from dynamo_query import DynamoTable, DynamoDictClass

# first, define your record
@dataclass
class UserRecord(DynamoDictClass):
    pk: str
    email: str
    name: str
    points: Optional[int] = None

    @DynamoDictClass.compute_key("pk")
    def get_pk(self) -> str:
        return self.email


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

    # use this property to define your table resource
    @property
    def table(self) -> Any:
        return boto3.resource("dynamodb").Table("user_table")


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

# add a new record to your table
user_table.upsert_record(
    UserRecord(
        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

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.py

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.

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

dynamoquery-2.7.1-py3-none-any.whl (41.8 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page