FastAPI Users Beanie DB Adapter

Warnings

• breaking Version 5.0.0 drops support for Python 3.9 and Beanie versions older than 2.0. Ensure your environment meets these requirements.
  Fix: Upgrade Python to 3.10+ and Beanie to 2.0+ before upgrading this package.
• breaking Versions 4.0.0 and 3.0.0 dropped support for Python 3.8 and 3.7 respectively. Older Python versions are no longer supported.
  Fix: Upgrade Python to 3.10+ for compatibility with the latest versions.
• breaking In version 2.0.0, `BeanieBaseUser` and `BeanieBaseAccessToken` became pure mixins and no longer automatically inherit from `beanie.Document`. You must explicitly inherit from `Document` when defining your user model. Additionally, `BeanieBaseUser` now only supports `PydanticObjectId` as the ID type.
  Fix: Update your User model definition to `class User(BeanieBaseUser[PydanticObjectId], Document): ...`
• gotcha Beanie database initialization (`init_beanie`) must occur before any database operations are attempted. This is typically done in FastAPI's `startup_event` handler or using an `async_context_manager` for the app's `lifespan`.
  Fix: Ensure `await init_beanie(database=db, document_models=[YourUserModel])` is called during application startup.
• deprecated The documentation currently instructs users to use `motor.motor_asyncio.AsyncIOMotorClient` for MongoDB connections. However, `pymongo 4.16+` now includes a native asynchronous client, suggesting `motor` might eventually be replaced or become less recommended.
  Fix: While `motor` still works, be aware of potential future changes in recommended connection methods as PyMongo's native async client matures. Monitor Beanie and FastAPI Users documentation for updates.

Install

• pip install fastapi-users-db-beanie Pip

Imports

• BeanieBaseUser

  from fastapi_users.db import BeanieBaseUser

• BeanieUserDatabase

  from fastapi_users.db import BeanieUserDatabase

• Document

  from beanie import Document

  Since v2.0.0, BeanieBaseUser is a mixin and must explicitly inherit from beanie.Document.
• PydanticObjectId

  from beanie import PydanticObjectId

  BeanieBaseUser only supports PydanticObjectId as the ID type since v2.0.0.

Quickstart

This quickstart demonstrates how to integrate `fastapi-users-db-beanie` with FastAPI Users and Beanie. It sets up a MongoDB connection, defines a `User` model inheriting from `BeanieBaseUser` and `beanie.Document`, and provides a dependency for `BeanieUserDatabase`. The FastAPI app includes authentication and user management routers, ensuring Beanie is initialized on startup.

import os
from typing import AsyncGenerator

import motor.motor_asyncio
from beanie import Document, init_beanie, PydanticObjectId
from fastapi import Depends, FastAPI
from fastapi_users import FastAPIUsers, schemas
from fastapi_users.authentication import BearerTransport, AuthenticationBackend
from fastapi_users.db import BeanieBaseUser, BeanieUserDatabase

# --- Beanie/MongoDB Setup ---
DATABASE_URL = os.environ.get('MONGODB_URL', 'mongodb://localhost:27017')
client = motor.motor_asyncio.AsyncIOMotorClient(
    DATABASE_URL,
    uuidRepresentation="standard"
)
db = client[os.environ.get('MONGODB_DB_NAME', 'database_name')]

class User(BeanieBaseUser[PydanticObjectId], Document):
    # You can add custom fields here
    pass

class UserRead(schemas.BaseUser[PydanticObjectId]):
    pass

class UserCreate(schemas.BaseUserCreate):
    pass

class UserUpdate(schemas.BaseUserUpdate):
    pass

async def get_user_db() -> AsyncGenerator[BeanieUserDatabase, None]:
    yield BeanieUserDatabase(User)

# --- FastAPI Users Setup ---
bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")

def get_jwt_strategy():
    SECRET = os.environ.get('AUTH_SECRET', 'YOUR_SUPER_SECRET_KEY') # CHANGE THIS IN PRODUCTION!
    return AuthenticationBackend(
        name="jwt",
        transport=bearer_transport,
        get_user_token_data=UserRead, # Or your custom token data schema
        secret=SECRET,
        lifetime_seconds=3600
    )

fastapi_users = FastAPIUsers(
    get_user_db,
    [get_jwt_strategy()],
    UserRead,
    UserCreate,
    UserUpdate
)

# --- FastAPI App ---
app = FastAPI()

@app.on_event("startup")
async def on_startup():
    await init_beanie(
        database=db,
        document_models=[User]
    )

app.include_router(
    fastapi_users.get_auth_router(get_jwt_strategy()),
    prefix="/auth/jwt",
    tags=["auth"]
)

app.include_router(
    fastapi_users.get_register_router(UserRead, UserCreate),
    prefix="/auth",
    tags=["auth"]
)

app.include_router(
    fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix="/users",
    tags=["users"]
)