aminalaee/mongox

Familiar async Python MongoDB ODM

---

MongoX

MongoX is an async python ODM (Object Document Mapper) for MongoDB which is built on top of Motor and Pydantic.

The main features include:

• Fully type annotated
• Async support Python 3.7+ (since it's built on top of Motor)
• Elegant editor support (since it's built on top of Pydantic)
• Autocompletion everywhere, from object creation to query results
• Custom query builder which is more intuitive and pythonic
• 100% test coverage

MongoX models are at the same time Pydantic models and have the same functionalitties, so you can use them with your existing Pydantic models.

---

Documentation: https://aminalaee.github.io/mongox

---

Installation

$ pip install mongox

---

Quickstart

You can define mongox models the same way you define Pydantic models. The difference is they should inherit from mongox.Model now:

import asyncio

import mongox

client = mongox.Client("mongodb://localhost:27017")
db = client.get_database("test_db")


class Movie(mongox.Model, db=db, collection="movies"):
    name: str
    year: int

Now you can create some instances and insert them into the database:

movie = await Movie(name="Forrest Gump", year=1994).insert()

The returned result will be a Movie instance, and mypy will understand that this is a Movie instance. So you will have type hints and validations everywhere.

Now you can fetch some data from the database.

You can use the same pattern as PyMongo/Motor:

movie = await Movie.query({"name": "Forrest Gump"}).get()

Or you can use Movie fields instead of dictionaries in the query (less room for bugs):

movie = await Movie.query({Movie.name: "Forrest Gump"}).get()

And finally you can use a more intuitive query (limited yet):

movie = await Movie.query(Movie.name == "Forrest Gump").get()

Notice how we omitted the dictionary and passed the Movie fields in comparison.

---

Please refer to the documentation here or the full examples here.

---

mongox's Issues

Add `get_or_none` method

Hey, I'm planing to add get_or_none method to the Queryset class.
Since most of the python ORMs have implemented this method, I think it's nice for us to have it too.
What is your opinion?

Would be nice to have default connection and collection

Hi

Nice idea

what would be nice is to have some global connection configuration... like in one module you do:

client = mongox.Client(
    "mongodb://localhost:27017", get_event_loop=asyncio.get_running_loop
)

mongox.set_default_client("mongodb://localhost:27017", get_event_loop=asyncio.get_running_loop)

and then in places where I define models:

class Movie(mongox.Model):
    name: str
    year: int

# !!! no Meta 

Now Movie should get client from set_default_client and collection name will become "movie" (lowercase from model name)

to override collection

class Movie(mongox.Model, collection="movies"):
    name: str
    year: int

Define ref field in the model

Hi, thank you for your perfect repo,
I wonder that can I define a ref field by mongx? See image for detail

Bulk Inserts of Dictionary

Currently it insert only one document at a time. It will improve performance if dictionary insert works.

Query builder for magic method `__contains__`

Query within a list type field with __contains__

import mongox


class Movie(mongox.Model):
    name: str
    year: int
   tags: List[str]


movies = await Movie.query(Movie.tag in ["A"]).all()

Error when working with multiple models

When working with 2 different models, it seems like mongox is ignoring the collection name and inserting into the first one's collection.

Currently, I have 2 models defined in 2 different files:

class UserAction(mongox.Model, db=db, collection="user_actions"):
    server_id: int
    name: str
    aliases: list[str]
    emoji: str
    self_text: str
    receivers_text: list[str]

class ServerSetting(mongox.Model, db=db, collection="server_settings"):
    server_id: int
    ignored_channels: list[int]

Firstly, I insert into server_settings, but when I try to insert into user_actions, the object gets created inside the server_settings collection instead.

Integration Tests

Integration tests with starlette.

Data validation for `get_or_create` method

Just as the update method required a validation of the data to be saved, the get_or_create method also requires it to avoid the construction of documents with erroneous data.

The idea would be to do something like this:

mongox/mongox/models.py

Lines 223 to 228 in ada1886

	values, _, validation_error = pydantic.validate_model(
	pydantic_model, kwargs
	)
	if validation_error:
	raise validation_error

Add `Q` method `regex` for matched strings

Need a way to search for a regular expression in a string type field.

movie = await Movie.query(Q.regex(Movie.name, r"\w+")).first()

Insert isn't working

My class
class Minutedsx(mongox.Model, db=db, indexes=ohlcindx):
tradingsymbol: str
date: datetime
open: float
high: float
low: float
close: float

class Meta:
    collection = db.get_collection("mindsx")

I import it in another class and tried insert in loop. but nothing gets inserted. tried the basic.py in example folder that also didnt insert anything. DB was created though.

EmbeddedModel

EmbeddedModel wil be the same as Model except that it doesn't have default _id and will be store inside models only.

Allow generator in `.all()` method

Right now we return a list when we call the Model.query().all() method.

This should be exposed from the motor client to allow async generators over the result set.

So something like:

models = Model.query()

async for model in models:
    print(model)


# And for lists

models = await Model.query().all()

Any feedback would be appreciated.

maybe can provide Model.get() function for get obj by id？

Although we can use Model.query({"_id": id}).first() to get object by id，But maybe it's better to use a shotcut function like:

Model.get("id")

Separating raw queries from query builder

Right now we support both at the same time:

# Raw query
await Movie.query({"name": "X"}).first()

# Query builder
await Movie.query(Movie.name == "X").first()

This leas to complexity handling both in the same method. I'm thinking maybe adding a .raw() method which accepts the query and doesn't do any validation/query building and sends it directly to MongoDB.

Possible error in the return of the update method

I have been creating the update_or_create method based on the get_or_create and update methods, doing some tests I found a possible bug in the update method.

in the database I have the following data.

[
  {
    "name": "Venom",
    "year": 2023
  },{
    "name": "Venom2",
    "year": 2021
  }
]

and when executing the next query Movie.query({Movie.name: "Venom"}).update({Movie.year: 2021}), what I would expect is for the method to return

[
  {
    "name": "Venom",
    "year": 2021
  }
]

But what it returns is

[
  {
    "name": "Venom",
    "year": 2021
  },{
    "name": "Venom2",
    "year": 2021
  }
]

Is this a bug or is expected to happen?

Querying inside lists

Need tests and probably fixes for querying inside a List field:

import mongox


class Movie(mongox.Model):
    name: str
    year: int
   tags: List[str]

Need to be able to query inside the tags fields. Maybe like this:

await Movie.query(Movie.tags == ["A", "B"])

await Movie.query(Q.contains(Movie.tags, ["A", "B"]))

Any feedback would be appreciated.

Setting default io_loop

Right now we don't specify any default io_loop for the Motor client.

I think it would be a good idea to add asyncio.get_event_loop if no loop is specified.

https://github.com/aminalaee/mongox/blob/main/mongox/database.py#L47