# DynamoDB SingleTable
https://pypi.org/project/ddb-single/
Python DynamoDB interface, specialized in single-table design.
DynamoDB is high-performance serverless NoSQL, but difficult to disign tables.
Single-table design needs only single table, and few GSIs (Global Secondary Indexes).
It makes effective and easy to manage your whole data models for single service.
## Getting Started
### Install
```
pip install ddb-single
```
### Start DynamoDB Local
```
docker run -d --rm -p 8000:8000 amazon/dynamodb-local
```
### Init Table
```python
from ddb_single import Table
table = Table(
table_name="sample",
endpoint_url="http://localhost:8000",
)
table.init()
```
### Data Models
Each model has al least 3 keys
- primary_key ... Hash key for single item. default: `pk: {__model_name__}_{uuid}`
- seconday_key ... Range key for item. default: `sk: {__model_name__}_item`
- unique_key ... key to identify the item is the same. Mainly used to update item.
And you can set `serch_key` to enable search via GSI
```python
from ddb_single import BaseModel, DBField, FieldType
class User(BaseModel):
__table__=table
__model_name__ = "user"
name = DBField(unique_key=True)
email = DBField(search_key=True)
age = DBField(type=FieldType.NUMBER, search_key=True)
description=DBField()
```
## Usage
need "Qurey" object for CRUD
- `query.model(foo).create`
- `query.model(foo).get`
- `query.model(foo).search`
- `query.model(foo).update`
- `query.model(foo).delete`
```python
from ddb_single import Query
query = Query(table)
```
### Create Item
If the item with same value of `unique_key` already exist, exist item is updated.
```python
user = User(name="John", email="john@example.com", description="test")
query.model(user).create()
```
Then, multible items added.
|pk|sk|data|name|email|description|
|-|-|-|-|-|-|
|user_xxxx|user_item||John|john@example.com|test|
|user_xxxx|search_user_name|John|
|user_xxxx|search_user_email|new-john@example.com|
In addition to main item (sk=`user_item`), multiple item (sk=`search_{__model_name__}_{field_name}`) added to table.
Those "search items" are used to search
The GSI `DataSearchIndex` is used to get "search items" to extract target's pk.
Then, `batch_get` items by pk.
|sk = hash|data = range|pk|
|-|-|-|
|search_user_name|John|user_xxxx|
|search_user_email|new-john@example.com|user_xxxx|
### Search Items
```python
user = query.model(User).search(User.name.eq("John"))
print(user)
# -> [{"pk":"user_xxxx", "sk":"user_item", "name":"John", "email":"john@example.com"}]
```
`pk_only=True` to extract pk without `batch_get`
```python
user_pks = query.model(User).search(User.name.eq("John"), pk_only=True)
print(user_pks)
# -> ["user_xxxx"]
```
### Get single item
`get(pk)` to get single item.
```
user = query.model(User).get("user_xxxx")
print(user)
# -> {"pk":"user_xxxx", "sk":"user_item", "name":"John", "email":"john@example.com"}
```
`get_by_unique` to get item by `unique_key`
```python
user = query.model(User).get_by_unique("John")
print(user)
# -> {"pk":"user_xxxx", "sk":"user_item", "name":"John", "email":"john@example.com"}
```
`pk_only=True` option in `get_by_unique` to get `primary key` without `get_item`
```python
pk = query.model(User).get_by_unique("John", pk_only=True)
print(pk)
# -> "user_xxxx"
```
### Update Item
```python
user = query.model(User).search(User.email.eq("john@example.com"))
new_user = User(**user[0])
new_user.email = "new-john@example.com"
query.model(new_user).update()
```
Or use unique value to detect exist item.
```python
new_user = User(name="John", email="new-john@example.com")
query.model(new_user).update()
```
Then, tha value of "main item" and "seach item" changed
|pk|sk|data|name|email|description|
|-|-|-|-|-|-|
|user_xxxx|user_item||John|new-john@example.com|test|
|user_xxxx|search_user_name|John|
|user_xxxx|search_user_email|new-john@example.com|
### Delete Item
```
user = query.model(User).search(User.email.eq("new-john@example.com"))
query.model(user[0]).delete()
```
`primary key` to detect exist item.
```
query.model(User).delete_by_pk("user_xxxx")
```
or `unique key`
```
query.model(User).delete_by_unique("John")
```
## Batch Writer
`table.batch_writer()` to create/update/delete multible items
- `query.model(foo).create(batch=batch)`
- `query.model(foo).update(batch=batch)`
- `query.model(foo).delete(batch=batch)`
### Batch Create
```python
with table.batch_writer() as batch:
for i in range(3):
user = User(name=f"test{i}", age=i+10)
query.model(user).create(batch=batch)
res = query.model(User).search(User.name.begins_with("test"))
print([(r["name"], r["age"]) for r in res])
# -> [("test0", 10), ("test1", 11), ("test2", 12)]
```
### Batch Update
```python
with table.batch_writer() as batch:
for i in range(3):
user = User(name=f"test{i}", age=i+20)
query.model(user).update(batch=batch)
res = query.model(User).search(User.name.begins_with("test"))
print([(r["name"], r["age"]) for r in res])
# -> [("test0", 20), ("test1", 21), ("test2", 22)]
```
### Batch Delete
```python
pks = query.model(User).search(User.name.begins_with("test"), pk_only=True)
with table.batch_writer() as batch:
for pk in pks:
query.model(user).delete_by_pk(pk, batch=batch)
res = query.model(User).search(User.name.begins_with("test"))
print(res)
# -> []
```
## Relationship
### Create Model
You can sat relationns to other models
`relation=BaseModel` to set relation.
```python
class BlogPost(BaseModel):
__model_name__ = "blogpost"
__table__=table
name = DBField(unique_key=True)
content = DBField()
author = DBField(reletion=User)
```
### Create Item
```python
blogpost = BlogPost(
name="Hello",
content="Hello world",
author=self.user
)
query.model(blogpost).create()
```
Then, tha value "reletion item" added
|pk|sk|data|name|author|content|
|-|-|-|-|-|-|
|user_xxxx|user_item||John|||
|user_xxxx|search_user_name|John|
|blogpost_xxxx|blogpost_item||Hello|John|Hello world|
|blogpost_xxxx|search_blogpost_title|Hello|
|blogpost_xxxx|rel_user_xxxx|author|
In addition to main item (sk=`blogpost_item`), relation item (sk=`rel_{primary_key}`) added to table. The GSI `DataSearchIndex` is used to get "relation items" to extract target's pk.
Then, `batch_get` items by pk.
|sk = hash|data = range|pk|
|-|-|-|
|rel_user_xxxx|author|blogpost_xxxx|
### Search Relations
`get_relation(model=Basemodel)` to search relations
```python
blogpost = query.model(BlogPost).get_by_unique("Hello")
blogpost = BlogPost(**blogpost)
user = query.model(blogpost).get_relation(model=User)
print(user)
# -> [{"pk":"user_xxxx", "sk":"user_item", "name":"John"}]
```
Also `get_relation(field=DBField)` to specify field
```python
user = query.model(blogpost).get_relation(field=BlogPost.author)
print(user)
# -> [{"pk":"user_xxxx", "sk":"user_item", "name":"John"}]
```
### Search Reference
In this library, "reference" is antonym to relation
`get_reference(model=Basemodel)` to search items related to the item
```python
user = query.model(User).get_by_unique("John")
user = User(**blogpost)
blogpost = query.model(blogpost).get_reference(model=BlogPost)
print(blogpost)
# -> [{"pk":"blogpost_xxxx", "sk":"blogpost_item", "name":"Hello"}]
```
Also `get_reference(field=DBField)` to specify field
```python
blogpost = query.model(user).get_reference(field=BlogPost.author)
print(blogpost)
# -> [{"pk":"blogpost_xxxx", "sk":"blogpost_item", "name":"Hello"}]
```
### Update Relation
If relation key's value changed, relationship also changed.
```python
new_user = User(name="Michael")
blogpost = query.model(BlogPost).get_by_unique("Hello")
blogpost["author"] = new_user
blogpost = BlogPost(**blogpost)
query.model(blogpost).update()
```
Then, "reletion item" changed
|pk|sk|data|name|author|content|
|-|-|-|-|-|-|
|user_xxxx|user_item||John|||
|user_xxxx|search_user_name|John|
|user_yyyy|user_item||Michael|||
|user_yyyy|search_user_name|Michael|
|blogpost_xxxx|blogpost_item||Hello|Michael|Hello world|
|blogpost_xxxx|search_blogpost_title|Hello|
|blogpost_xxxx|rel_user_yyyy|author|
### Delete Relation
If related item deleted, relationship also deleted
```python
query.model(user).delete_by_unique("Michael")
```
Then, "reletion item" deleted.
But main item's value is not chenged.
|pk|sk|data|name|author|content|
|-|-|-|-|-|-|
|user_xxxx|user_item||John|||
|user_xxxx|search_user_name|John|
|blogpost_xxxx|blogpost_item||Hello|Michael|Hello world|
|blogpost_xxxx|search_blogpost_title|Hello|
Raw data
{
"_id": null,
"home_page": "https://medaka0213.github.io/DynamoDB-SingleTable/",
"name": "ddb_single",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.10",
"maintainer_email": null,
"keywords": "aws, dynamodb, serverless",
"author": "medaka0213",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/b0/88/06e05d34550547e706800444ab923b7f560360f6b51e9e9e133670c57be7/ddb_single-0.4.17.tar.gz",
"platform": null,
"description": "# DynamoDB SingleTable\n\nhttps://pypi.org/project/ddb-single/\n\nPython DynamoDB interface, specialized in single-table design.\nDynamoDB is high-performance serverless NoSQL, but difficult to disign tables.\n\nSingle-table design needs only single table, and few GSIs (Global Secondary Indexes).\nIt makes effective and easy to manage your whole data models for single service.\n\n## Getting Started\n\n### Install\n\n```\npip install ddb-single\n```\n\n### Start DynamoDB Local\n\n```\ndocker run -d --rm -p 8000:8000 amazon/dynamodb-local\n```\n\n### Init Table\n\n```python\nfrom ddb_single import Table\n\ntable = Table(\n table_name=\"sample\",\n endpoint_url=\"http://localhost:8000\",\n)\ntable.init()\n```\n\n### Data Models\n\nEach model has al least 3 keys\n- primary_key ... Hash key for single item. default: `pk: {__model_name__}_{uuid}` \n- seconday_key ... Range key for item. default: `sk: {__model_name__}_item`\n- unique_key ... key to identify the item is the same. Mainly used to update item.\n\nAnd you can set `serch_key` to enable search via GSI \n\n```python\nfrom ddb_single import BaseModel, DBField, FieldType\n\nclass User(BaseModel):\n __table__=table\n __model_name__ = \"user\"\n name = DBField(unique_key=True)\n email = DBField(search_key=True)\n age = DBField(type=FieldType.NUMBER, search_key=True)\n description=DBField()\n```\n\n## Usage\n\nneed \"Qurey\" object for CRUD\n- `query.model(foo).create`\n- `query.model(foo).get`\n- `query.model(foo).search`\n- `query.model(foo).update`\n- `query.model(foo).delete`\n\n```python\nfrom ddb_single import Query\nquery = Query(table)\n```\n\n\n### Create Item\n\nIf the item with same value of `unique_key` already exist, exist item is updated.\n\n```python\nuser = User(name=\"John\", email=\"john@example.com\", description=\"test\")\nquery.model(user).create()\n```\n\nThen, multible items added.\n\n|pk|sk|data|name|email|description|\n|-|-|-|-|-|-|\n|user_xxxx|user_item||John|john@example.com|test|\n|user_xxxx|search_user_name|John|\n|user_xxxx|search_user_email|new-john@example.com|\n\nIn addition to main item (sk=`user_item`), multiple item (sk=`search_{__model_name__}_{field_name}`) added to table.\nThose \"search items\" are used to search\n\nThe GSI `DataSearchIndex` is used to get \"search items\" to extract target's pk.\nThen, `batch_get` items by pk.\n\n|sk = hash|data = range|pk|\n|-|-|-|\n|search_user_name|John|user_xxxx|\n|search_user_email|new-john@example.com|user_xxxx|\n\n### Search Items\n\n```python\nuser = query.model(User).search(User.name.eq(\"John\"))\nprint(user)\n# -> [{\"pk\":\"user_xxxx\", \"sk\":\"user_item\", \"name\":\"John\", \"email\":\"john@example.com\"}]\n```\n\n`pk_only=True` to extract pk without `batch_get`\n\n```python\nuser_pks = query.model(User).search(User.name.eq(\"John\"), pk_only=True)\nprint(user_pks)\n# -> [\"user_xxxx\"]\n```\n\n### Get single item\n\n`get(pk)` to get single item.\n\n```\nuser = query.model(User).get(\"user_xxxx\")\nprint(user)\n# -> {\"pk\":\"user_xxxx\", \"sk\":\"user_item\", \"name\":\"John\", \"email\":\"john@example.com\"}\n```\n\n`get_by_unique` to get item by `unique_key`\n\n```python\nuser = query.model(User).get_by_unique(\"John\")\nprint(user)\n# -> {\"pk\":\"user_xxxx\", \"sk\":\"user_item\", \"name\":\"John\", \"email\":\"john@example.com\"}\n```\n\n`pk_only=True` option in `get_by_unique` to get `primary key` without `get_item`\n\n```python\npk = query.model(User).get_by_unique(\"John\", pk_only=True)\nprint(pk)\n# -> \"user_xxxx\"\n```\n\n### Update Item\n\n```python\nuser = query.model(User).search(User.email.eq(\"john@example.com\"))\nnew_user = User(**user[0])\nnew_user.email = \"new-john@example.com\"\nquery.model(new_user).update()\n```\n\nOr use unique value to detect exist item.\n\n```python\nnew_user = User(name=\"John\", email=\"new-john@example.com\")\nquery.model(new_user).update()\n```\n\nThen, tha value of \"main item\" and \"seach item\" changed\n\n|pk|sk|data|name|email|description|\n|-|-|-|-|-|-|\n|user_xxxx|user_item||John|new-john@example.com|test|\n|user_xxxx|search_user_name|John|\n|user_xxxx|search_user_email|new-john@example.com|\n\n\n### Delete Item\n\n```\nuser = query.model(User).search(User.email.eq(\"new-john@example.com\"))\nquery.model(user[0]).delete()\n```\n\n`primary key` to detect exist item.\n\n```\nquery.model(User).delete_by_pk(\"user_xxxx\")\n```\n\n\nor `unique key`\n\n```\nquery.model(User).delete_by_unique(\"John\")\n```\n\n## Batch Writer\n\n`table.batch_writer()` to create/update/delete multible items\n- `query.model(foo).create(batch=batch)`\n- `query.model(foo).update(batch=batch)`\n- `query.model(foo).delete(batch=batch)`\n\n### Batch Create\n\n```python\nwith table.batch_writer() as batch:\n for i in range(3):\n user = User(name=f\"test{i}\", age=i+10)\n query.model(user).create(batch=batch)\nres = query.model(User).search(User.name.begins_with(\"test\"))\nprint([(r[\"name\"], r[\"age\"]) for r in res])\n# -> [(\"test0\", 10), (\"test1\", 11), (\"test2\", 12)]\n```\n\n### Batch Update\n\n```python\nwith table.batch_writer() as batch:\n for i in range(3):\n user = User(name=f\"test{i}\", age=i+20)\n query.model(user).update(batch=batch)\nres = query.model(User).search(User.name.begins_with(\"test\"))\nprint([(r[\"name\"], r[\"age\"]) for r in res])\n# -> [(\"test0\", 20), (\"test1\", 21), (\"test2\", 22)]\n```\n\n### Batch Delete\n\n```python\npks = query.model(User).search(User.name.begins_with(\"test\"), pk_only=True)\nwith table.batch_writer() as batch:\n for pk in pks:\n query.model(user).delete_by_pk(pk, batch=batch)\nres = query.model(User).search(User.name.begins_with(\"test\"))\nprint(res)\n# -> []\n```\n\n\n## Relationship\n\n### Create Model\n\nYou can sat relationns to other models\n`relation=BaseModel` to set relation.\n\n```python\nclass BlogPost(BaseModel):\n __model_name__ = \"blogpost\"\n __table__=table\n name = DBField(unique_key=True)\n content = DBField()\n author = DBField(reletion=User)\n```\n\n### Create Item\n\n\n```python\nblogpost = BlogPost(\n name=\"Hello\",\n content=\"Hello world\",\n author=self.user\n)\nquery.model(blogpost).create()\n```\n\nThen, tha value \"reletion item\" added\n\n|pk|sk|data|name|author|content|\n|-|-|-|-|-|-|\n|user_xxxx|user_item||John|||\n|user_xxxx|search_user_name|John|\n|blogpost_xxxx|blogpost_item||Hello|John|Hello world|\n|blogpost_xxxx|search_blogpost_title|Hello|\n|blogpost_xxxx|rel_user_xxxx|author|\n\nIn addition to main item (sk=`blogpost_item`), relation item (sk=`rel_{primary_key}`) added to table. The GSI `DataSearchIndex` is used to get \"relation items\" to extract target's pk.\nThen, `batch_get` items by pk.\n\n|sk = hash|data = range|pk|\n|-|-|-|\n|rel_user_xxxx|author|blogpost_xxxx|\n\n### Search Relations\n\n`get_relation(model=Basemodel)` to search relations\n\n```python\nblogpost = query.model(BlogPost).get_by_unique(\"Hello\")\nblogpost = BlogPost(**blogpost)\n\nuser = query.model(blogpost).get_relation(model=User)\nprint(user)\n# -> [{\"pk\":\"user_xxxx\", \"sk\":\"user_item\", \"name\":\"John\"}]\n```\n\nAlso `get_relation(field=DBField)` to specify field\n\n```python\nuser = query.model(blogpost).get_relation(field=BlogPost.author)\nprint(user)\n# -> [{\"pk\":\"user_xxxx\", \"sk\":\"user_item\", \"name\":\"John\"}]\n```\n\n### Search Reference\n\nIn this library, \"reference\" is antonym to relation\n\n`get_reference(model=Basemodel)` to search items related to the item\n\n```python\nuser = query.model(User).get_by_unique(\"John\")\nuser = User(**blogpost)\n\nblogpost = query.model(blogpost).get_reference(model=BlogPost)\nprint(blogpost)\n# -> [{\"pk\":\"blogpost_xxxx\", \"sk\":\"blogpost_item\", \"name\":\"Hello\"}]\n```\n\nAlso `get_reference(field=DBField)` to specify field\n\n```python\nblogpost = query.model(user).get_reference(field=BlogPost.author)\nprint(blogpost)\n# -> [{\"pk\":\"blogpost_xxxx\", \"sk\":\"blogpost_item\", \"name\":\"Hello\"}]\n```\n\n### Update Relation\n\nIf relation key's value changed, relationship also changed.\n\n\n```python\nnew_user = User(name=\"Michael\")\nblogpost = query.model(BlogPost).get_by_unique(\"Hello\")\nblogpost[\"author\"] = new_user\nblogpost = BlogPost(**blogpost)\n\nquery.model(blogpost).update()\n```\n\nThen, \"reletion item\" changed\n\n|pk|sk|data|name|author|content|\n|-|-|-|-|-|-|\n|user_xxxx|user_item||John|||\n|user_xxxx|search_user_name|John|\n|user_yyyy|user_item||Michael|||\n|user_yyyy|search_user_name|Michael|\n|blogpost_xxxx|blogpost_item||Hello|Michael|Hello world|\n|blogpost_xxxx|search_blogpost_title|Hello|\n|blogpost_xxxx|rel_user_yyyy|author|\n### Delete Relation\n\nIf related item deleted, relationship also deleted\n\n```python\nquery.model(user).delete_by_unique(\"Michael\")\n```\n\nThen, \"reletion item\" deleted.\nBut main item's value is not chenged.\n\n|pk|sk|data|name|author|content|\n|-|-|-|-|-|-|\n|user_xxxx|user_item||John|||\n|user_xxxx|search_user_name|John|\n|blogpost_xxxx|blogpost_item||Hello|Michael|Hello world|\n|blogpost_xxxx|search_blogpost_title|Hello|\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python DynamoDB interface, specialized in single-table design.",
"version": "0.4.17",
"project_urls": {
"Homepage": "https://medaka0213.github.io/DynamoDB-SingleTable/",
"Repository": "https://github.com/medaka0213/DynamoDB-SingleTable"
},
"split_keywords": [
"aws",
" dynamodb",
" serverless"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "1cb1b866a85f8c2dc330742f5cc24ec989c54d56c7e97162bc927aa110a4a2f7",
"md5": "8b22e2663aa61df5d1250f996364908e",
"sha256": "e7467f0a373d80f099cbca6341cf5bcc9a68cd2d8b71761066adc24a41a84d44"
},
"downloads": -1,
"filename": "ddb_single-0.4.17-py3-none-any.whl",
"has_sig": false,
"md5_digest": "8b22e2663aa61df5d1250f996364908e",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 16685,
"upload_time": "2024-09-21T05:06:09",
"upload_time_iso_8601": "2024-09-21T05:06:09.428528Z",
"url": "https://files.pythonhosted.org/packages/1c/b1/b866a85f8c2dc330742f5cc24ec989c54d56c7e97162bc927aa110a4a2f7/ddb_single-0.4.17-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "b08806e05d34550547e706800444ab923b7f560360f6b51e9e9e133670c57be7",
"md5": "037e50c09531a7124d17dcd57e304851",
"sha256": "96b00e34f6e052479370a891d2623d41f9c8adfed7a05cb643988cac96996a6d"
},
"downloads": -1,
"filename": "ddb_single-0.4.17.tar.gz",
"has_sig": false,
"md5_digest": "037e50c09531a7124d17dcd57e304851",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 14577,
"upload_time": "2024-09-21T05:06:10",
"upload_time_iso_8601": "2024-09-21T05:06:10.365968Z",
"url": "https://files.pythonhosted.org/packages/b0/88/06e05d34550547e706800444ab923b7f560360f6b51e9e9e133670c57be7/ddb_single-0.4.17.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-21 05:06:10",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "medaka0213",
"github_project": "DynamoDB-SingleTable",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "ddb_single"
}
JSON
