from typing import Optional, TypeVar from datetime import datetime from pydantic import Field from beanie import PydanticObjectId from fastapi_users.db import BeanieBaseUser, BeanieUserDatabase from fastapi_users_db_beanie.access_token import BeanieAccessTokenDatabase, BeanieBaseAccessToken from pymongo import IndexModel class AccessToken(BeanieBaseAccessToken[PydanticObjectId]): pass class User(BeanieBaseUser[PydanticObjectId]): login: str entity_id: str created_at: datetime = Field(default=datetime.utcnow(), nullable=False) updated_at: datetime = Field(default_factory=datetime.utcnow, nullable=False) class Settings: indexes = [ IndexModel("login", unique=True), ] class UserDatabase(BeanieUserDatabase): async def get_by_login(self, login: str) -> Optional[TypeVar("UP_BEANIE", bound=BeanieBaseUser)]: """Get a single user by email.""" return await self.user_model.find_one( self.user_model.login == login ) async def get_access_token_db(): yield BeanieAccessTokenDatabase(AccessToken) async def get_user_db(): yield UserDatabase(User)