Beanie is a Python library that provides an asynchronous Object-Document Mapper (ODM) for MongoDB. It integrates seamlessly with FastAPI, making it easy to define, manage, and query MongoDB collections using Python models. Beanie is built on Pydantic and Motor (an async MongoDB driver), which ensures high performance and compatibility with FastAPI's data validation and serialization features.

Beanie is a data modeling library for MongoDB, designed to integrate seamlessly with FastAPI.

https://beanie-odm.dev/

Key Features of Beanie

1. Asynchronous Operations: Built on Motor, all database interactions are asynchronous.
2. Schema Validation: Uses Pydantic models for schema definition and validation.
3. Querying: Provides a Pythonic query language for MongoDB.
4. Relationships: Supports nested documents and references.
5. Migrations: Allows database migrations for schema changes.
6. Aggregation: Supports MongoDB's aggregation framework.

Setting Up Beanie with FastAPI 

1. Installation

pip install beanie motor fastapi

2.  Define a Beanie Document

A Beanie document is defined by inheriting from Document. You can add fields, validators, and default values using Pydantic.

from beanie import Document
from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class Item(Document):
    name: str
    description: Optional[str]
    price: float
    created_at: datetime = datetime.utcnow()

    class Settings:
        collection = "items"  # MongoDB collection name

3.Configure MongoDB Connection

Set up the database connection in your FastAPI app.

from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    database = client.my_database  # Replace with your database name
    await init_beanie(database=database, document_models=[Item])

Here:

• Item is a document model.
• init_beanie connects the models to the MongoDB database.

4. Create API Endpoints

Use the Beanie model to handle CRUD operations in your endpoints.

from fastapi import HTTPException

@app.get("/items/")
async def list_items():
    items=await Item.find_all().to_list()
    return items
    
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    await item.insert()
    return item

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: str):
    item = await Item.get(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, update_data: dict):
    item = await Item.get(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    await item.update({"$set": update_data})
    # await item.set(updated_data)
    return item

@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
    item = await Item.get(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    await item.delete()
    return {"message": "Item deleted successfully"}

Querying Documents

Beanie offers a Pythonic way to query documents using the find method.

• Filter by Conditions

  @app.get("/items/filter/")
  async def filter_items():
      users = await Item.find(Item.price <20).to_list()
      return users

• Aggregation 

  @app.get("/items/average-price/")
  async def average_price():
      pipeline = [{"$group": {"_id": None, "average_price": {"$avg": "$price"}}}]
      result = await Item.aggregate(pipeline).to_list()
      return result

Advantages of Using Beanie in FastAPI

1. Integration with FastAPI: Both use Pydantic, enabling seamless validation and serialization.
2. Async Operations: Makes it ideal for applications requiring high concurrency.
3. Ease of Use: Simplifies the interaction with MongoDB compared to raw queries.
4. Schema Validation: Ensures data integrity with minimal effort.
5. Simplified CRUD: Provides built-in methods for common database operations.

By using Beanie, you can focus more on developing your application rather than worrying about low-level MongoDB operations.

1. Querying in Beanie

Beanie provides a high-level API for querying MongoDB collections. Queries are async and use a clean, Pythonic syntax.

Basic Queries

• Retrieve all documents:

  items = await Item.find_all().to_list()  # Returns a list of all items

• Retrieve a single document by ID:

  item = await Item.get("document_id")
  
  item = await Item.find_one((Item.name == "abc") & (Item.price > 50))
  

• Retrieve documents with a filter:

  items = await Item.find(Item.price > 50).to_list()  # Items priced above 50

• Sorting results:

  items = await Item.find().sort("-price").to_list()  # Descending by price

• Limit and skip:

  items = await Item.find().limit(10).skip(5).to_list()  # Pagination

Advanced Queries

• Compound filters:

  from beanie.operators import And, Or
  
  items = await Item.find(
      And(
          Item.price > 50,
          Item.name == "Special Item"
      )
  ).to_list()

• Text search 

  items = await Item.find({"$text": {"$search": "example"}}).to_list()

• Aggregation

  pipeline = [
      {"$match": {"price": {"$gte": 50}}},
      {"$group": {"_id": None, "average_price": {"$avg": "$price"}}},
  ]
  result = await Item.aggregate(pipeline).to_list()

• Updating documents:

  await Item.find(Item.name == "Old Name").update({"$set": {"name": "New Name"}})

2. Migrations in Beanie

Beanie supports schema migrations, making it easier to handle changes in your document schema over time.

Use Cases for Migrations

• Adding or removing fields in a document.
• Modifying field types or default values.
• Backfilling data for new fields.

Defining a Migration

You define migrations as Python functions.

from beanie.migrations.controllers import Migration

class AddFieldMigration(Migration):
    name = "Add 'stock' field to Item"

    async def upgrade(self):
        await Item.find_all().update({"$set": {"stock": 0}})

    async def downgrade(self):
        await Item.find_all().update({"$unset": {"stock": ""}})

Applying Migrations

Run migrations during the startup event or as a separate script:

from beanie.migrations.utils import apply_migrations

@app.on_event("startup")
async def run_migrations():
    await apply_migrations()

3. References Between Models

Beanie supports both embedded and referenced relationships between documents, making it easy to model complex data.

Embedded Relationships

Use embedded models when the related data is small and doesn't need to be queried independently.

• Define Embedded Models:

  from pydantic import BaseModel
  
  class Tag(BaseModel):
      name: str
      value: int
  
  class Item(Document):
      name: str
      tags: list[Tag]

• Usage:

  item = Item(name="Sample Item", tags=[Tag(name="Sale", value=10)])
  await item.insert()

Referenced Relationships

Use references when related data is stored in separate collections and needs independent querying.

• Define Referenced Models:

  from beanie import Link
  
  class Category(Document):
      name: str
  
  class Item(Document):
      name: str
      category: Link[Category]

• Inserting Data with References:

  category = Category(name="Electronics")
  await category.insert()
  
  item = Item(name="Smartphone", category=category)
  await item.insert()

• Querying Referenced Data:

  item = await Item.find_one(Item.name == "Smartphone").populate("category")
  print(item.category.name)  # Access referenced category

Cascading Operations

You can configure cascading behaviors for referenced models, such as deleting related documents automatically.

class Item(Document):
    name: str
    category: Link[Category]

    class Settings:
        use_state_management = True
        cascade = ["category"]