Open
Description
beanie has some typing limitations (for example, the ID field is always optional, which is frustrating in cases when you know for sure the ID is not supposed to be null)
secondly, the link references are lazily inferred by beanie- which means type information is lost in the IDE.
its actually simpler to write our own pydantic models and return them in the repository layers, using the motor async mongodb python driver in our repository layer.
examples:
from typing import Optional
from pydantic import BaseModel, Field
from bson import ObjectId
# Custom ObjectId validator
class PyObjectId(ObjectId):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, v):
if not ObjectId.is_valid(v):
raise ValueError("Invalid ObjectId")
return ObjectId(v)
@classmethod
def __modify_schema__(cls, field_schema):
field_schema.update(type="string")
# Base schema (shared logic)
class MongoModel(BaseModel):
id: PyObjectId = Field(alias="_id")
class Config:
allow_population_by_field_name = True
json_encoders = {ObjectId: str}
arbitrary_types_allowed = True
# Create and DB variants
class UserCreate(BaseModel):
name: str
email: str
class UserInDB(MongoModel):
name: str
email: strfrom motor.motor_asyncio import AsyncIOMotorClient
from pymongo import ASCENDING
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client["mydb"]
users_collection = db["users"]from typing import List
async def create_user(user: UserCreate) -> UserInDB:
user_dict = user.dict()
result = await users_collection.insert_one(user_dict)
return UserInDB(**user_dict, _id=result.inserted_id)
async def get_user_by_id(user_id: str) -> Optional[UserInDB]:
doc = await users_collection.find_one({"_id": ObjectId(user_id)})
return UserInDB(**doc) if doc else None
async def list_users() -> List[UserInDB]:
cursor = users_collection.find().sort("name", ASCENDING)
return [UserInDB(**doc) async for doc in cursor]References: