FastApi

1
from typing import Optional
2

3
from fastapi import Depends, Request
4
from fastapi_users import (BaseUserManager, IntegerIDMixin, exceptions, models,
5
                           schemas)
6

7
from auth.models import User
8
from auth.utils import get_user_db
9
from config import SECRET_AUTH
10

11

12
class UserManager(IntegerIDMixin, BaseUserManager[User, int]):
13
    reset_password_token_secret = SECRET_AUTH
14
    verification_token_secret = SECRET_AUTH
15

16
    async def on_after_register(self, user: User, request: Optional[Request] = None):
17
        print(f"User {user.id} has registered.")
18

19
    async def create(
20
        self,
21
        user_create: schemas.UC,
22
        safe: bool = False,
23
        request: Optional[Request] = None,
24
    ) -> models.UP:
25
        await self.validate_password(user_create.password, user_create)
26

27
        existing_user = await self.user_db.get_by_email(user_create.email)
28
        if existing_user is not None:
29
            raise exceptions.UserAlreadyExists()
30

31
        user_dict = (
32
            user_create.create_update_dict()
33
            if safe
34
            else user_create.create_update_dict_superuser()
35
        )
36
        password = user_dict.pop("password")
37
        user_dict["hashed_password"] = self.password_helper.hash(password)
38
        user_dict["role_id"] = 1
39

40
        created_user = await self.user_db.create(user_dict)
41

42
        await self.on_after_register(created_user, request)
43

44
        return created_user
45

46

47
async def get_user_manager(user_db=Depends(get_user_db)):
48
    yield UserManager(user_db)
49