Async fork of PynamoDB powered by aiobotocore. Requires Python 3.10+.
This fork tracks brunobelloni/AioPynamoDB and publishes fork builds on PyPI as as-aiopynamodb.
From PyPI:
$ pip install as-aiopynamodb -U
From GitHub:
$ pip install git+https://github.com/AppSolves/AioPynamoDB#egg=as-aiopynamodb
Create a model that describes your DynamoDB table.
fromaiopynamodb.modelsimportModelfromaiopynamodb.attributesimportUnicodeAttributeclassUserModel(Model):
""" A DynamoDB User """classMeta:
table_name="dynamodb-user"email=UnicodeAttribute(null=True)
first_name=UnicodeAttribute(range_key=True)
last_name=UnicodeAttribute(hash_key=True)AioPynamoDB allows you to create the table if needed (it must exist before you can use it!):
awaitUserModel.create_table(read_capacity_units=1, write_capacity_units=1)Create a new user:
user=UserModel("John", "Denver")
user.email="djohn@company.org"awaituser.save()Now, search your table for all users with a last name of 'Denver' and whose first name begins with 'J':
asyncforuserinUserModel.query("Denver", UserModel.first_name.startswith("J")):
print(user.first_name)Examples of ways to query your table with filter conditions:
asyncforuserinUserModel.query("Denver", UserModel.email=="djohn@company.org"):
print(user.first_name)Retrieve an existing user:
try:
user=awaitUserModel.get("John", "Denver")
print(user)
exceptUserModel.DoesNotExist:
print("User does not exist")Want to use indexes? No problem:
fromaiopynamodb.modelsimportModelfromaiopynamodb.indexesimportGlobalSecondaryIndex, AllProjectionfromaiopynamodb.attributesimportNumberAttribute, UnicodeAttributeclassViewIndex(GlobalSecondaryIndex):
classMeta:
read_capacity_units=2write_capacity_units=1projection=AllProjection()
view=NumberAttribute(default=0, hash_key=True)
classTestModel(Model):
classMeta:
table_name="TestModel"forum=UnicodeAttribute(hash_key=True)
thread=UnicodeAttribute(range_key=True)
view=NumberAttribute(default=0)
view_index=ViewIndex()Now query the index for all items with 0 views:
asyncforiteminTestModel.view_index.query(0):
print("Item queried from index: {0}".format(item))It's really that simple.