# Mongodantic
[](https://github.com/cocreators-ee/mongodantic/actions/workflows/publish.yaml)
[](https://github.com/psf/black)
[](https://github.com/PyCQA/bandit)
[](https://github.com/cocreators-ee/mongodantic/blob/master/.pre-commit-config.yaml)
[](https://pypi.org/project/mongodantic-python/)
[](https://pypi.org/project/mongodantic-python/)
[](https://opensource.org/licenses/BSD-3-Clause)
Python async database models for MongoDB using Pydantic base models. It should work on Python 3.9+, maybe 3.8,
not quite sure.
## Motivation
It's usually a good idea to have a simple layer on your DB that doesn't try to do too much, but takes care of
the basic things like validating your data and mapping database records to class instances, and overall
providing basic database access helpers. [Pydantic](https://docs.pydantic.dev) does a great job at the typing
of models and validating data, and just needs a bit of database logic around it to provide all the
capabilities commonly needed.
There are similar libraries already for other databases that serve as inspiration for this, e.g.
[firedantic](http://github.com/ioxiocom/firedantic) for Firestore, and
[arangodantic](https://github.com/ioxiocom/arangodantic) for ArangoDB.
## Installation
It's a Python library, what do you expect?
```bash
pip install mongodantic-python
# OR
poetry add mongodantic-python
```
## Usage
Small example of how you can use this library (also in [readme_example.py](./readme_example.py)).
```python
import asyncio
from datetime import datetime
from typing import Optional, Sequence
import pymongo
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import Field
# IndexModel and ASCENDING are just passed through from pymongo
from mongodantic import ASCENDING, IndexModel, Model, ModelNotFoundError, set_database
MONGODB_CONNECT_STR = "mongodb://localhost:27017" # Point to your MongoDB server
class User(Model):
# Indexes are automatically created when model is accessed
indexes: Sequence[IndexModel] = [
IndexModel(
keys=[
("name", ASCENDING),
]
),
]
# id properly is automatically added - stored as _id in MongoDB
# Pydantic typing + Field usage works great
created: datetime = Field(default_factory=datetime.now)
name: Optional[str] = None
# You can of course add methods
def greet(self):
print(f"Hello, {self.name} from {self.created}")
async def rename(self):
self.name = f"Another {self.name}"
await self.save()
# You can also run code after loading objects from DB
async def after_load(self) -> None:
self.greet()
async def main():
# Configure the DB connection at the start of your application
print("Connecting to DB")
client = AsyncIOMotorClient(MONGODB_CONNECT_STR)
db = client["my_test_db"]
set_database(db)
# You can use this for cleanup
# for user in await User.find({}):
# await user.delete()
# And just use the models
print("Creating user")
user = User()
await user.save()
print("Updating user")
user.name = "Test"
await user.save()
print("Renaming user")
await user.rename()
# Load up a specific one if you know the str representation of its id
print("Searching by ID")
user_again = await User.get_by_id(user.id)
assert user_again.name == "Another Test"
# Find many
# {} is a Pymongo filter, if filtering by id make sure you use "_id" key and ObjectId() for value
print("Finding all users")
users = await User.find({})
assert len(users) == 1
# Counting
for idx in range(0, 9):
u = User(name=f"user-{idx + 1}")
await u.save()
# Add a user that sorts to the end
u = User(name=f"zuser")
await u.save()
assert await User.count() == 11
assert await User.count({"name": user.name}) == 1
# Pagination
users = await User.find({"name": {"$ne": user.name}}, skip=3, limit=3)
assert len(users) == 3
for u in users:
print(u.name)
# Load up the first matching entry
print("Finding a user by name")
test_user = await User.find_one({"name": "Another Test"})
assert test_user.id == user.id
# Sorting
print("Sorting")
users = await User.find({}, sort="name")
for u in users:
print(u.name)
last_by_name = await User.find_one({}, sort=[("name", pymongo.DESCENDING)])
print(last_by_name.name)
print("Deleting users")
for u in users:
await u.delete()
try:
print("Attempting reload")
await user.reload()
raise Exception("User was supposed to be deleted")
except ModelNotFoundError:
print("User not found")
if __name__ == "__main__":
asyncio.run(main())
```
## Development
Issues and PRs are welcome!
Please open an issue first to discuss the idea before sending a PR so that you know if it would be wanted or
needs re-thinking or if you should just make a fork for yourself.
For local development, make sure you install [pre-commit](https://pre-commit.com/#install), then run:
```bash
pre-commit install
poetry install
poetry run ptw . # Hit Ctrl+C when done with your changes
poetry run python readme_example.py
```
## License
The code is released under the BSD 3-Clause license. Details in the [LICENSE.md](./LICENSE.md) file.
# Financial support
This project has been made possible thanks to [Cocreators](https://cocreators.ee) and
[Lietu](https://lietu.net). You can help us continue our open source work by supporting us on
[Buy me a coffee](https://www.buymeacoffee.com/cocreators).
[](https://www.buymeacoffee.com/cocreators)
Raw data
{
"_id": null,
"home_page": "https://github.com/cocreators-ee/mongodantic/",
"name": "mongodantic-python",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4",
"maintainer_email": "",
"keywords": "database,mongodb,pymongo,odm,async",
"author": "Janne Enberg",
"author_email": "janne.enberg@lietu.net",
"download_url": "https://files.pythonhosted.org/packages/e3/c7/d22de80228be40821a18fdbc1c4fd7b6b455c1c6c5be9a2bdc3922f80182/mongodantic_python-0.4.0.tar.gz",
"platform": null,
"description": "# Mongodantic\n\n[](https://github.com/cocreators-ee/mongodantic/actions/workflows/publish.yaml)\n[](https://github.com/psf/black)\n[](https://github.com/PyCQA/bandit)\n[](https://github.com/cocreators-ee/mongodantic/blob/master/.pre-commit-config.yaml)\n[](https://pypi.org/project/mongodantic-python/)\n[](https://pypi.org/project/mongodantic-python/)\n[](https://opensource.org/licenses/BSD-3-Clause)\n\nPython async database models for MongoDB using Pydantic base models. It should work on Python 3.9+, maybe 3.8,\nnot quite sure.\n\n## Motivation\n\nIt's usually a good idea to have a simple layer on your DB that doesn't try to do too much, but takes care of\nthe basic things like validating your data and mapping database records to class instances, and overall\nproviding basic database access helpers. [Pydantic](https://docs.pydantic.dev) does a great job at the typing\nof models and validating data, and just needs a bit of database logic around it to provide all the\ncapabilities commonly needed.\n\nThere are similar libraries already for other databases that serve as inspiration for this, e.g.\n[firedantic](http://github.com/ioxiocom/firedantic) for Firestore, and\n[arangodantic](https://github.com/ioxiocom/arangodantic) for ArangoDB.\n\n## Installation\n\nIt's a Python library, what do you expect?\n\n```bash\npip install mongodantic-python\n# OR\npoetry add mongodantic-python\n```\n\n## Usage\n\nSmall example of how you can use this library (also in [readme_example.py](./readme_example.py)).\n\n```python\nimport asyncio\nfrom datetime import datetime\nfrom typing import Optional, Sequence\n\nimport pymongo\nfrom motor.motor_asyncio import AsyncIOMotorClient\nfrom pydantic import Field\n\n# IndexModel and ASCENDING are just passed through from pymongo\nfrom mongodantic import ASCENDING, IndexModel, Model, ModelNotFoundError, set_database\n\nMONGODB_CONNECT_STR = \"mongodb://localhost:27017\" # Point to your MongoDB server\n\n\nclass User(Model):\n # Indexes are automatically created when model is accessed\n indexes: Sequence[IndexModel] = [\n IndexModel(\n keys=[\n (\"name\", ASCENDING),\n ]\n ),\n ]\n\n # id properly is automatically added - stored as _id in MongoDB\n\n # Pydantic typing + Field usage works great\n created: datetime = Field(default_factory=datetime.now)\n name: Optional[str] = None\n\n # You can of course add methods\n def greet(self):\n print(f\"Hello, {self.name} from {self.created}\")\n\n async def rename(self):\n self.name = f\"Another {self.name}\"\n await self.save()\n\n # You can also run code after loading objects from DB\n async def after_load(self) -> None:\n self.greet()\n\n\nasync def main():\n # Configure the DB connection at the start of your application\n print(\"Connecting to DB\")\n client = AsyncIOMotorClient(MONGODB_CONNECT_STR)\n db = client[\"my_test_db\"]\n set_database(db)\n\n # You can use this for cleanup\n # for user in await User.find({}):\n # await user.delete()\n\n # And just use the models\n print(\"Creating user\")\n user = User()\n await user.save()\n\n print(\"Updating user\")\n user.name = \"Test\"\n await user.save()\n\n print(\"Renaming user\")\n await user.rename()\n\n # Load up a specific one if you know the str representation of its id\n print(\"Searching by ID\")\n user_again = await User.get_by_id(user.id)\n assert user_again.name == \"Another Test\"\n\n # Find many\n # {} is a Pymongo filter, if filtering by id make sure you use \"_id\" key and ObjectId() for value\n print(\"Finding all users\")\n users = await User.find({})\n assert len(users) == 1\n\n # Counting\n for idx in range(0, 9):\n u = User(name=f\"user-{idx + 1}\")\n await u.save()\n\n # Add a user that sorts to the end\n u = User(name=f\"zuser\")\n await u.save()\n\n assert await User.count() == 11\n assert await User.count({\"name\": user.name}) == 1\n\n # Pagination\n users = await User.find({\"name\": {\"$ne\": user.name}}, skip=3, limit=3)\n assert len(users) == 3\n for u in users:\n print(u.name)\n\n # Load up the first matching entry\n print(\"Finding a user by name\")\n test_user = await User.find_one({\"name\": \"Another Test\"})\n assert test_user.id == user.id\n\n # Sorting\n print(\"Sorting\")\n users = await User.find({}, sort=\"name\")\n for u in users:\n print(u.name)\n\n last_by_name = await User.find_one({}, sort=[(\"name\", pymongo.DESCENDING)])\n print(last_by_name.name)\n\n print(\"Deleting users\")\n for u in users:\n await u.delete()\n\n try:\n print(\"Attempting reload\")\n await user.reload()\n raise Exception(\"User was supposed to be deleted\")\n except ModelNotFoundError:\n print(\"User not found\")\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## Development\n\nIssues and PRs are welcome!\n\nPlease open an issue first to discuss the idea before sending a PR so that you know if it would be wanted or\nneeds re-thinking or if you should just make a fork for yourself.\n\nFor local development, make sure you install [pre-commit](https://pre-commit.com/#install), then run:\n\n```bash\npre-commit install\npoetry install\npoetry run ptw . # Hit Ctrl+C when done with your changes\npoetry run python readme_example.py\n```\n\n## License\n\nThe code is released under the BSD 3-Clause license. Details in the [LICENSE.md](./LICENSE.md) file.\n\n# Financial support\n\nThis project has been made possible thanks to [Cocreators](https://cocreators.ee) and\n[Lietu](https://lietu.net). You can help us continue our open source work by supporting us on\n[Buy me a coffee](https://www.buymeacoffee.com/cocreators).\n\n[](https://www.buymeacoffee.com/cocreators)\n\n",
"bugtrack_url": null,
"license": "BSD-3-Clause",
"summary": "Pydantic models for MongoDB",
"version": "0.4.0",
"project_urls": {
"Documentation": "https://github.com/cocreators-ee/mongodantic/",
"Homepage": "https://github.com/cocreators-ee/mongodantic/",
"Repository": "https://github.com/cocreators-ee/mongodantic/"
},
"split_keywords": [
"database",
"mongodb",
"pymongo",
"odm",
"async"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "ed3d341f74775ba0e9c53fffcc1c1323b55e1c95db1e6a7b900658af6c62deb5",
"md5": "fc2b5416cf40d7a7911cb2e2d41f4495",
"sha256": "e405c617cd631cdc20af91f5ff4c33badd4ee1839ab10022a142865440b36b0e"
},
"downloads": -1,
"filename": "mongodantic_python-0.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "fc2b5416cf40d7a7911cb2e2d41f4495",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4",
"size": 8403,
"upload_time": "2024-03-18T19:22:24",
"upload_time_iso_8601": "2024-03-18T19:22:24.710347Z",
"url": "https://files.pythonhosted.org/packages/ed/3d/341f74775ba0e9c53fffcc1c1323b55e1c95db1e6a7b900658af6c62deb5/mongodantic_python-0.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e3c7d22de80228be40821a18fdbc1c4fd7b6b455c1c6c5be9a2bdc3922f80182",
"md5": "5c4d1885f7104a39e642a7a8492b4979",
"sha256": "e1aba74d73d0f821549830fde8560210979196e697a3fd585a1d6058e5dab001"
},
"downloads": -1,
"filename": "mongodantic_python-0.4.0.tar.gz",
"has_sig": false,
"md5_digest": "5c4d1885f7104a39e642a7a8492b4979",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4",
"size": 7181,
"upload_time": "2024-03-18T19:22:26",
"upload_time_iso_8601": "2024-03-18T19:22:26.461798Z",
"url": "https://files.pythonhosted.org/packages/e3/c7/d22de80228be40821a18fdbc1c4fd7b6b455c1c6c5be9a2bdc3922f80182/mongodantic_python-0.4.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-03-18 19:22:26",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cocreators-ee",
"github_project": "mongodantic",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "mongodantic-python"
}
JSON
