I'm happy to introduce to you the new version of Beanie and a lot of new features, that come with it.

Here is the feature list:

• Relations
• Event-based actions
• Cache
• Revision

Relations

This feature is perhaps the most anticipated of all. It took some time, but finally, it is here. Relations.

The document can contain links to other documents in their fields.

Only top-level fields are fully supported for now.

Direct link to the document:

from beanie import Document, Link

class Door(Document):
    height: int = 2
    width: int = 1


class House(Document):
    name: str
    door: Link[Door]  # This is the link

List of the links:

from typing import List

from beanie import Document, Link

class Window(Document):
    x: int = 10
    y: int = 10


class House(Document):
    name: str
    door: Link[Door]
    windows: List[Link[Window]]  # This is the list of the links

Other link patterns are not supported for now. If you need something more specific for your use-case, please leave an issue on the GitHub page - https://github.com/roman-right/beanie

Write

The next write methods support relations:

• insert(...)
• replace(...)
• save(...)

To apply the writing method to the linked documents, you should set the respective link_rule parameter

house.windows = [Window(x=100, y=100)]
house.name = "NEW NAME"

# The next call will insert a new window object 
# and replace the house instance with updated data
await house.save(link_rule=WriteRules.WRITE)

# `insert` and `replace` methods will work the same way

Or Beanie can ignore internal links with the link_rule parameter WriteRules.DO_NOTHING

house.door.height = 3
house.name = "NEW NAME"

# The next call will just replace the house instance 
# with new data, but the linked door object will not be synced
await house.replace(link_rule=WriteRules.DO_NOTHING)

# `insert` and `save` methods will work the same way

Fetch

Prefetch

You can fetch linked documents on the find query step, using the parameter fetch_links

houses = await House.find(
    House.name == "test", 
    fetch_links=True
).to_list()

All the find methods supported:

• find
• find_one
• get

Beanie uses a single aggregation query under the hood to fetch all the linked documents. This operation is very effective.

On-demand fetch

If you don't use prefetching, linked documents will be presented as objects of the Link class. You can fetch them manually then.
To fetch all the linked documents you can use the fetch_all_links method

await house.fetch_all_links()

It will fetch all the linked documents and replace Link objects with them.

Or you can fetch a single field:

await house.fetch_link(House.door)

This will fetch the Door object and put it in the door field of the house object.

Delete

Delete method works the same way as write operations, but it uses other rules:

To delete all the links on the document deletion you should use the DeleteRules.DELETE_LINKS value for the link_rule parameter

await house.delete(link_rule=DeleteRules.DELETE_LINKS)

To keep linked documents you can use the DO_NOTHING rule

await house.delete(link_rule=DeleteRules.DO_NOTHING)

Event-based actions

You can register methods as pre- or post- actions for document events like insert, replace and etc.

Currently supported events:

• Insert
• Replace
• SaveChanges
• ValidateOnSave

To register an action you can use @before_event and @after_event decorators respectively.

from beanie import Insert, Replace

class Sample(Document):
    num: int
    name: str

    @before_event(Insert)
    def capitalize_name(self):
        self.name = self.name.capitalize()

    @after_event(Replace)
    def num_change(self):
        self.num -= 1

It is possible to register action for a list of events:

from beanie import Insert, Replace

class Sample(Document):
    num: int
    name: str

    @before_event([Insert, Replace])
    def capitalize_name(self):
        self.name = self.name.capitalize()

This will capitalize the name field value before each document insert and replace

And sync and async methods could work as actions.

from beanie import Insert, Replace

class Sample(Document):
    num: int
    name: str

    @after_event([Insert, Replace])
    async def send_callback(self):
        await client.send(self.id)

Cache

All the query results could be locally cached.

This feature must be turned on in the Settings inner class explicitly.

class Sample(Document):
    num: int
    name: str

    class Settings:
        use_cache = True

Beanie uses LRU cache with expiration time. You can set capacity (the maximum number of the cached queries) and expiration time in the Settings inner class.

class Sample(Document):
    num: int
    name: str

    class Settings:
        use_cache = True
        cache_expiration_time = datetime.timedelta(seconds=10)
        cache_capacity = 5

Any query will be cached for this document class.

# on the first call it will go to the database
samples = await Sample.find(num>10).to_list()

# on the second - it will use cache instead
samples = await Sample.find(num>10).to_list()

await asyncio.sleep(15)

# if the expiration time was reached 
# it will go to the database again
samples = await Sample.find(num>10).to_list()

Revision

This feature helps with concurrent operations.

It stores revision_id together with the document and changes it on each document update. If the application with the old local copy of the document will try to change it, an exception will be raised. Only when the local copy will be synced with the database, the application will be allowed to change the data. It helps to avoid losses of data.

This feature must be turned on in the Settings inner class explicitly too.

class Sample(Document):
    num: int
    name: str

    class Settings:
        use_revision = True

Any changing operation will check if the local copy of the document has the actual revision_id value:

s = await Sample.find_one(Sample.name="TestName")
s.num = 10

# If a concurrent process already changed the doc, 
# the next operation will raise an error
await s.replace()

If you want to ignore revision and apply all the changes even if the local copy is outdated, you can use the parameter ignore_revision

await s.replace(ignore_revision=True)

Other

There is a bunch of smaller features, presented in this release. I would like to mention a couple of them here.

Save changes

Beanie can keep the document state, that synced with the database, to find local changes and save only them.

This feature must be turned on in the Settings inner class explicitly.

class Sample(Document):
    num: int
    name: str

    class Settings:
        use_state_management = True

To save only changed values the save_changes() method should be used.

s = await Sample.find_one(Sample.name == "Test")
s.num = 100
await s.save_changes()

The save_changes() method can be used only with already inserted documents.

On save validation

Pydantic has very useful config to validate values on assignment - validate_assignment = True. But unfortunately, this is a heavy operation and doesn't fit some use cases.

You can validate all the values before saving the document (insert, replace, save, save_changes) with beanie config validate_on_save instead.

This feature must be turned on in the Settings inner class explicitly.

class Sample(Document):
    num: int
    name: str

    class Settings:
        validate_on_save = True

If any field has a wrong value, it will raise an error on write operations (insert, replace, save, save_changes)

sample = await Sample.find_one(Sample.name == "Test")
sample.num = "wrong value type"

# Next call will raise an error
await sample.replace()

Conclusion

Thank you for reading. I hope you'll find these features useful.

If you would like to help with development - there are some issues at the GitHub page of the project - https://github.com/roman-right/beanie

Links

• Beanie Project
• Documentation
• Discord Server