import uuid from typing import Any, Dict, Generic, Optional from fastapi import Depends from fastapi_users import BaseUserManager, UUIDIDMixin, models, exceptions, schemas from .models import User, get_user_db SECRET = "SECRET" class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]): reset_password_token_secret = SECRET verification_token_secret = SECRET async def get_by_email(self, user_email: str) -> models.UP: return await self.get_by_login(user_email) async def get_by_login(self, user_login: str) -> models.UP: """ Get a user by user_login. :param user_login: E-mail of the user to retrieve. :raises UserNotExists: The user does not exist. :return: A user. """ user = await self.user_db.get_by_login(user_login) if user is None: raise exceptions.UserNotExists() return user async def create( self, user_create: schemas.UC, safe: bool = False ) -> models.UP: """ Create a user in database. Triggers the on_after_register handler on success. :param user_create: The UserCreate model to create. :param safe: If True, sensitive values like is_superuser or is_verified will be ignored during the creation, defaults to False. :raises UserAlreadyExists: A user already exists with the same e-mail. :return: A new user. """ await self.validate_password(user_create.password, user_create) existing_user = await self.user_db.get_by_login(user_create.login) if existing_user is not None: raise exceptions.UserAlreadyExists() user_dict = ( user_create.create_update_dict() if safe else user_create.create_update_dict_superuser() ) password = user_dict.pop("password") user_dict["hashed_password"] = self.password_helper.hash(password) created_user = await self.user_db.create(user_dict) await self.on_after_register(created_user) return created_user async def get_user_manager(user_db=Depends(get_user_db)): yield UserManager(user_db)