Typed Mongo
A production-ready modern Python MongoDB ODM
In addition to synchronous mode, you can use asynchronous mode, just export from typedmongo.asyncio
.
Install
pip install typedmongo
Usage
Usage examples trump all usage documentation. So please look at the Example below first.
Example
from motor.motor_asyncio import AsyncIOMotorClient as MongoClient
import typedmongo.asyncio as mongo
from typedmongo.marshamallow import MarshamallowObjectId
class Wallet(mongo.Table):
balance: mongo.DecimalField
class User(mongo.Table):
_id: mongo.ObjectIdField = mongo.ObjectIdField(
marshamallow=MarshamallowObjectId(required=False)
)
name: mongo.StringField
age: mongo.IntegerField
tags: mongo.ListField[str]
wallet: mongo.EmbeddedField[Wallet]
created_at: mongo.DateTimeField = mongo.DateTimeField(
default=lambda: datetime.datetime.now(datetime.timezone.utc)
)
children: mongo.ListField[User]
extra: mongo.DictField = mongo.DictField(default=dict)
async def main():
await mongo.initial_collections(
MongoClient().mongo,
User,
)
document_id = await User.objects.insert_one(
User.load(
{
"name": "Aber",
"age": 18,
"tags": ["a", "b"],
"wallet": {"balance": 100},
"children": [],
},
)
)
user = await User.objects.find_one(User._id == document_id, sort=[+User.age])
update_result = await User.objects.update_one(
User._id == document_id, {"$set": {"tags": ["a", "b", "e", "r"]}}
)
delete_result = await User.objects.delete_one(User._id == document_id)
user = await User.objects.find_one_and_update(
User._id == document_id, {"$set": {"tags": ["a", "b", "e"]}}
)
user = await User.objects.find_one_and_replace(
User._id == document_id,
User.load({"name": "Aber", "age": 0}),
after_document=True,
)
user = await User.objects.find_one_and_delete(User._id == document_id)
users = [user async for user in User.objects.find(User.age == 18, sort=[-User.age])]
update_result = await User.objects.update_many(
User.wallet._.balance == Decimal("100"), {"$inc": {"wallet.balance": 10}}
)
await User.objects.count_documents(User.age >= 0)
await User.objects.bulk_write(
mongo.DeleteOne(User._id == 0),
mongo.DeleteMany(User.age < 18),
mongo.InsertOne(User.load({"name": "InsertOne"}, partial=True)),
mongo.ReplaceOne(User.name == "Aber", User.load({}, partial=True)),
mongo.UpdateMany({}, {"$set": {"age": 25}}),
mongo.UpdateMany(User.name == "Yue", {"$set": {"name": "yue"}}),
)
Table
Table.load
: Load data from dict to instance, and validate the data.Table.dump
: Dump the instance to jsonable dict.
Field
ObjectIdField
StringField
IntegerField
DecimalField
DateTimeField
DictField
EmbeddedField
ListField
Conditional expressions
Comparison expressions
Table.field == value
Table.field != value
Table.field > value
Table.field >= value
Table.field < value
Table.field <= value
Logical expressions
(Table.field == value) & (Table.field == value)
(Table.field == value) | (Table.field == value)
~(Table.field == value)
~((Table.field == value) & (Table.field == value))
~((Table.field == value) | (Table.field == value))
Sort expressions
+Table.field
: Ascending-Table.field
: Descending