71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
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) |