42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
from typing import Optional, TypeVar
|
|
from datetime import datetime
|
|
from pydantic import Field
|
|
from beanie import Document
|
|
|
|
from fastapi_users.db import BeanieBaseUser, BeanieUserDatabase
|
|
from fastapi_users_db_beanie.access_token import BeanieAccessTokenDatabase, BeanieBaseAccessToken
|
|
|
|
from pymongo import IndexModel
|
|
|
|
|
|
class AccessToken(BeanieBaseAccessToken, Document):
|
|
pass
|
|
|
|
|
|
class User(BeanieBaseUser, Document):
|
|
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)
|