altitudenetworks / dynamoquery Goto Github PK

Python AWS DynamoDB ORM

Home Page: https://altitudenetworks.github.io/dynamoquery/

License: MIT License

Python 99.78% Shell 0.22%

boto3 dynamodb python orm aws aws-dynamodb serverless serverless-architectures serverless-applications type-hints type-safety

dynamoquery's Introduction

DynamoQuery

Helper for building Boto3 DynamoDB queries.

DynamoQuery
- Installation
- Usage
  - DynamoQuery
  - DynamoTable
- Development
  - VSCode
  - PyCharm
- Versioning

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
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="[email protected]",
        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.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.

dynamoquery's People

Contributors

Stargazers

Watchers

Forkers

howbazaar njgrisafi amirkav fwereade stacklet

dynamoquery's Issues

How could I control the serialization of an enum in a DynamoRecord?

Hi folks,

We are looking to use the DynamoRecords closely with ariadne graphql.

The graphql type wraps enum.Enum quite nicely, and I have an attribute in our record class of that enum type. However when I try to upsert_record, I get:

TypeError: Unsupported type "<enum 'CloudProvider'>" for value "CloudProvider.AWS"

This is my second attempt because I was using enum.IntEnum, which was happily accepting the type from the graphql parts and creating the records in dynamodb. However it was storing the integer value in the DB, and I was wanting a string value stored.

Is there an existing pattern in the library that I should be using to control the serialization and deserialization of a record field?

added extra fields in upsert_record breaks _convert_record

upsert_record on DynamoTable breaks when using DynamoDictClass record. In the upsert_record code it converts the passed in record to a dict and then adds extra fields which it passes to _convert_record. The default implementation attempts to pass the dict into the record_class constructor. It doesn't look like dataclasses are designed to handle this method of initialization or maybe there is a bug as I get an error about missing required positional arguments for all of the required fields in the record.

Below is a test (pytest) that exhibits the behavior.
Python version: 3.8.5
OS: ubuntu 20.04 (WSL)
requires moto

import logging
from dataclasses import dataclass
from typing import Any, Optional

import boto3
from dynamo_query import DynamoDictClass, DynamoTable
from moto import mock_dynamodb2

@dataclass
class MyTestRecord(DynamoDictClass):
  required1: str
  required2: str
  optional: Optional[str] = None

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

  @DynamoDictClass.compute_key("sort")
  def get_sort(self) -> str:
    return self.required2


# Create your dynamo table manager with your record class
class MyTestTable(DynamoTable[MyTestRecord]):
  sort_key_name = "sort"
  record_class = MyTestRecord

  def __init__(self, table_name: str, endpoint_url: str = None, logger: Optional[logging.Logger] = None) -> None:
    super().__init__(logger=logger)
    self.table_name = table_name
    self.endpoint_url = endpoint_url

  # use this property to define your table resource
  @property
  def table(self) -> Any:
    return boto3.resource("dynamodb", endpoint_url=self.endpoint_url,
    region_name="us-west-1",
            aws_access_key_id="null",
            aws_secret_access_key="null").Table(self.table_name)

@mock_dynamodb2
def test_upsert():
  table = MyTestTable("test")
  res = table.upsert_record(MyTestRecord(required1="req1", required2="req2", optional="opt"))
  assert res is not None

The tests fails with this output:

=================================== FAILURES ===================================
_________________________________ test_upsert __________________________________
   @mock_dynamodb2
    def test_upsert():
      table = MyTestTable("test")
>     res = table.upsert_record(MyTestRecord(required1="req1", required2="req2", optional="opt"))

tests/test_dynamoquery_bug.py:45: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../venvs/project/lib/python3.8/site-packages/dynamo_query/dynamo_table.py:914: in upsert_record
    new_record = self._convert_record(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.test_dynamoquery_bug.MyTestTable object at 0x7f804f5a9a00>
record = {'dt_created': '2021-03-12T19:19:23.649553', 'dt_modified': '2021-03-12T19:19:23.649553', 'optional': 'opt', 'pk': 'req1', ...}

    def _convert_record(self, record: Union[_RecordType, Dict[str, Any]]) -> _RecordType:
        # pylint: disable=isinstance-second-argument-not-valid-type
        if self.record_class and not isinstance(record, self.record_class):
            # pylint: disable=not-callable
>           return self.record_class(record)  # type: ignore
E           TypeError: __init__() missing 1 required positional argument: 'required2'

../../venvs/project/lib/python3.8/site-packages/dynamo_query/dynamo_table.py:222: TypeError

Example for conditional upsert

I have a record that I'd like to conditionally update.

The use case is this:

I have a record I'd like to insert as long as it doesn't exist, if it does exist I'd like it to fail
Having read the record, it now has the dt_modified value thanks to the DynamoTable, I'd only like the update to happen if the dt_modified matches the last read modified timestamp.

I tried the following code snippet:

table.upsert_record(new_record, ConditionExpression('dt_modified', '=', 'last_dt_modified'), extra_data={'last_dt_modified': 'not same'})

However that hit this traceback:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/tim/stacklet/platform/.venv/lib/python3.8/site-packages/dynamo_query/dynamo_table.py", line 923, in upsert_record
    self.dynamo_query_class.build_update_item(
  File "/home/tim/stacklet/platform/.venv/lib/python3.8/site-packages/dynamo_query/dynamo_query_main.py", line 646, in execute_dict
    return self.execute(data_table=[data or {"dummy": True}])
  File "/home/tim/stacklet/platform/.venv/lib/python3.8/site-packages/dynamo_query/dynamo_query_main.py", line 611, in execute
    return method_map[self._query_type](data_table)
  File "/home/tim/stacklet/platform/.venv/lib/python3.8/site-packages/dynamo_query/base_dynamo_query.py", line 341, in _execute_method_update_item
    self._validate_required_value_keys(data_table)
  File "/home/tim/stacklet/platform/.venv/lib/python3.8/site-packages/dynamo_query/base_dynamo_query.py", line 275, in _validate_required_value_keys
    raise DynamoQueryError(
dynamo_query.base_dynamo_query.DynamoQueryError: Column "last_dt_modified" is missing in input data, but present in ConditionExpression = "{dt_modified} = {last_dt_modified__value}"

First question: "how do you go about providing values for a conditional upsert?"
Second question: "how could I have an upsert fail if the item exists?"

[BUG] DynamoQuery.build_projection_expression does not exist

I am trying to use the projection expression and instead of hardcoding the values, I am trying to pass the values as a list at runtime. I am running into problems where it does not evaluate individual items from the list and creates a properly formatted .projection statement within the query. further looking into the source code and docs it was mentioned that ProjectionExpression can be built using the method within the DynamoQuery object. inspecting the DynamoQuery object no methods with "build_projection_expression" exists.

can you help me figure out how to actually make the projections work passing the attribute list at runtime?

projection_expression = DynamoQuery.build_projection_expression(
['first_name', 'last_name']

Is it possible to create a batch_get_records type query that uses an alternate index?

I feel like this should be possible.

I have a secondary index that has all the item values. I feel like it should be relatively easy to pass the index through from the table batch_get_records to the query builder.

Am I on the right track @vemel?

cut a release - last release 9/2020

there's roughly 30 commits to trunks since the last release almost two years ago.

Provide consistent_read flags for query and scan on DynamoTable

The underlying query builder supports it for build_query, and it should be trivial to add for build_scan.

This would be very useful to be able to set on the higher level methods.

altitudenetworks / dynamoquery Goto Github PK

dynamoquery's Introduction

DynamoQuery

Installation

Usage

DynamoQuery

DynamoTable

Development

VSCode

PyCharm

Versioning

dynamoquery's People

Contributors

Stargazers

Watchers

Forkers

dynamoquery's Issues

Recommend Projects

Recommend Topics

Recommend Org