10. User 계좌 신규 등록 및 오류 설정

1. Introduction

user의 계좌를 만듣기 위해선, 먼저 api호출에 받을 input형식을 만드는 것이 중요합니다. 일단, 우리는 신규가입자에게 아래와 같은 정보를 제공해주기를 요청할 것입니다.

- user_id

- email

- first_name

- last_name

- pasword

이 정보를 통해서 계좌를 등록하는 방법을 이번 lesson을 통해서 알아보겠습니다.

2. Restigser

일단 user에게 요청할 정보를 pydantic model를 통해서 정의하는 것이 정말로 중요합니다.

-auth    ** user계정 관리를 담은 폴더 
  |-schema.py
  |-routes.py
  |-service.py

schema.py에 pydantic model를 이용해서 아래의 코드를 작성합니다.

from datetime import datetime
from pydantic import BaseModel, EmailStr,Field
from typing import Optional,List


class UserInput(BaseModel):
  user_id:str
  email:EmailStr
  first_name:str
  last_name:str
  password_hash:str
  is_verified:bool

그 다음에, routing에서 등록된 email이 이미 존재하는지를 체크하고, 등록하는 함수를 만들필요가 있는데요. 이미, 앞에서 언급한 AuthService에 추가적으로 클래스 함수를 정의내려줍니다.

2.1 get_user_by_email : 현재 이메일이 존재하는지 확인하는 함수

아래의 클래스함수는 mail이 존재하는지 안하는지를 판단해주는 함수입니다.

class AuthService:
     #기존 코드는 생략 

  async def get_user_by_email(self,email:str,session:AsyncSession):
      #check if user exists 
      statement=select(User).where(User.email==email)
      user_exist=await session.exec(statement)
      result=user_exist.first()

      return result

2.2 create_user: 계정을 등록하는 함수

  async def create_user(self,account:UserInput,session:AsyncSession):
      user_dict=account.model_dump()

      email=user_dict['email']
      #check if  the requested email exists in our current database
      result= await self.get_user_by_email(email,session)


      if result:
        # UserExist는 아래의 글 참조 
        raise UserExist()


      #create a string of hex digits in standard form
      new_user=User(
        updated_at=datetime.now(),
        **user_dict
      )

      #hasing the confidential information
      new_user.passowrd_hash=hash(user_dict['password_hash'])
      #set the role as 'user'
      new_user.role="user"
      session.add(new_user)
      await session.commit()

만약, user email이 이미 등록된 상태라면, error 메세지를 전달해야 합니다. 이를 UserExist()라는 class로 이미 정의를 내렸습니다. 이미,배웠던 것과 동일하게 작성하면 됩니다. "Exception" class를 상속받은 뒤에, 이전에 작성했던 create_exception_handler함수에다가 추가적으로 집어넣으면 됩니다.

-src
 |-auth **user계좌등록 및 관리 폴더
 |-erros.py **여기에다가 집어넣음

from typing import Any, Callable

from fastapi import FastAPI,status
from fastapi.responses import JSONResponse
from h11 import Request


class UserExist(Exception):
  """User already exists"""

class UserNotFound(Exception):
  """Request user not found"""



def create_exception_handler(status_code:int,detail:Any)->Callable[[Request,Exception],JSONResponse]:

  async def exception_handler(request:Request,exc:AuthException):
    return JSONResponse(
      content=detail,
      status_code=status_code
    )

  return exception_handler


def register_errors(app:FastAPI):

  ################ Authorization Error #######################
  app.add_exception_handler(
    UserNotFound,
    create_exception_handler(
      status_code=status.HTTP_404_NOT_FOUND,
      detail={
        "message":"The request email not found"
      }
    )
  )


  app.add_exception_handler(
    UserExist,
    create_exception_handler(
      status_code=status.HTTP_400_BAD_REQUEST,
      detail={
        "message":"The requested user already exists",
        "error_code":"User Registrer Error"
      }
    )

다음은 serviec.py에 대한 full-code입니다.

class AuthService:
  @abstractmethod
  def find_record(self,email=None,user_id=None):

    if email:
      statement=select(User).where(User.email==email)
    elif user_id:
      statement=select(User).where(User.user_id==user_id)
    else:
      raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail="You should give either correct user_name or email")
    return statement

  async def get_users(self,session:AsyncSession):
    statement=select(User)
    users=await session.exec(statement)

    return users.all()



  async def get_user_by_conditioins(self,user_id:str,email:str,session:AsyncSession):

    statement=self.find_record(user_id=user_id,email=email)

    result=await session.exec(statement)
    result=result.first()


    if not result:
      raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail="There is no macthing record")

    return result

  async def delete_user_by_conditions(self,account:UserSearch,session:AsyncSession):
    email,user_id=account.email,account.user_id
    user=await self.get_user_by_conditioins(user_id,email,session)
    await session.delete(user)
    await session.commit()

    return {"msg":"you requst has been successful"}


  async def get_user_by_email(self,email:str,session:AsyncSession):
      #check if user exists 
      statement=select(User).where(User.email==email)
      user_exist=await session.exec(statement)
      result=user_exist.first()

      return result




  async def create_user(self,account:UserInput,session:AsyncSession):
      user_dict=account.model_dump()

      email=user_dict['email']
      #check if  the requested email exists in our current database
      result= await self.get_user_by_email(email,session)


      if result:
        raise UserExist()


      #create a string of hex digits in standard form
      new_user=User(
        updated_at=datetime.now(),
        **user_dict
      )

      #hasing the confidential information
      new_user.passowrd_hash=hash(user_dict['password_hash'])
      #set the role as 'user'
      new_user.role="user"
      session.add(new_user)
      await session.commit()

2.3 router

이제 router.py에서 path지정과 creaste_user함수를 지정해주면 등록이 완료됩니다. 계좌등록을 위한 api-paht를 우리는 sign-up이라고 하겠습니다.

from fastapi import Depends,APIRouter, HTTPException,status
from fastapi.responses import JSONResponse
from sqlmodel.ext.asyncio.session import AsyncSession
from datetime import datetime
from src.auth.dependencies import  AccessTokenBearer, RefreshTokenBearer, RoleChecker,find_current_user
from src.auth.utils import create_access_token, decode_url_safe_token
from src.config import Settings
from src.db.main import get_session
from src.auth import schema,service
from src.auth.schema import UserLogin, UserOutput, UserSearch,UserInput,EmailModel
from src.config import settings

from src.mail import mail,create_message
from src.errors import (
  AccessForbidden,
  InvalidToken,
  UserNotFound
)


token_bearer=AccessTokenBearer()
auth_service=service.AuthService()
role_checker=Depends(RoleChecker(allowed_roles=['admin','user']))

# Any path starting with "/users" will be grouped into this folder
auth_router=APIRouter(
  prefix="/users",
  tags=["Users"]
)

@auth_router.post("/sign-up")
async def create_user(account:UserInput,session:AsyncSession=Depends(get_session)):
  result=await auth_service.create_user(account,session)
  return result