예외 처리

main.py

...
...
from fastapi import FastAPI, Depends
from fastapi.security import APIKeyHeader
...

API_KEY_HEADER = API_KEY_HEADER(name='Authorization', auto_error=False)

# 라우터 정의
    app.include_router(user.router, tags=['Users'], prefix='/api', dependencies=[Depends(API_KEY_HEADER)])
 ...
 

Depends, APIKeyHeader를 임포트 할게요.
APIKeyHeader의 매개변수는 키워드 인자로 각각 name, auto_error를 두게 되요. name에는 토큰의 키로 사용하는 Authorizaiton을 문자열로 넣게되요. 그리고 auto_error는 False로 해주지 않으면 오류를 뱉어내기에 False로 해뒀어요.

라우터 정의부분에 추가로 연결했는데요. routes디렉토리에 users.py를 만들고 내부 소스코드는 아래와 같이 만들게요.

users.py

from fastapi import APIRouter
from starlette.requests import Request

from app.database.schema import Users
from app.errors.exceptions import NotFoundUserEx
from app.models import UserMe

router = APIRouter()


@router.get("/me", response_model=UserMe)
async def get_user(request: Request):
    """
    get my info
    :param request:
    :return:
    """
    user = request.state.user
    user_info = Users.get(id=user.id)
    return user_info

오늘 중요한 것중 하나가 include_router()의 매개변수 중 하나인 dependencies=[Depends(API_KEY_HEADER) 부분인데요.
요렇게 일단 해두고 서버 켜서 /docs에서 한번 확인해 볼게요.

특정 endpoint에 자물쇠가 잠겨 있는게 보이조?
결국 로그인하고 토큰값으로 인증해야 해당 엔드포인트에 접근 할 수 있음을 나타내는거조.

로그인하고 토큰을 발급 받은 이후 Authorize버튼을 눌러서 발급 받은Bear fjkasdfjj127y~~~~를 입락할게요. 단 쌍 따옴표는 다 없고 bear부터 시작해야합니다.

정리하자면 include_router()메서드의 dependencies키워드를 통해서 토큰 인증이 필요한 endpoint로 제약을 걸 수 있다는 점!

errors

errors/exceptions.py

오류에 관한 코드와 예외처리를 별도의 모듈로 빼서 정리해 둔거에요.

class StatusCode:
    HTTP_500 = 500
    HTTP_400 = 400
    HTTP_401 = 401
    HTTP_403 = 403
    HTTP_404 = 404
    HTTP_405 = 405


class APIException(Exception):
    status_code: int
    code: str
    msg: str
    detail: str

    def __init__(
        self,
        *,
        status_code: int = StatusCode.HTTP_500,
        code: str = "000000",
        msg: str = None,
        detail: str = None,
        ex: Exception = None,
    ):
        self.status_code = status_code
        self.code = code
        self.msg = msg
        self.detail = detail
        super().__init__(ex)


class NotFoundUserEx(APIException):
    def __init__(self, user_id: int = None, ex: Exception = None):
        super().__init__(
            status_code=StatusCode.HTTP_404,
            msg=f"해당 유저를 찾을 수 없습니다.",
            detail=f"Not Found User ID : {user_id}",
            code=f"{StatusCode.HTTP_400}{'1'.zfill(4)}",
            ex=ex,
        )


class NotAuthorized(APIException):
    def __init__(self, ex: Exception = None):
        super().__init__(
            status_code=StatusCode.HTTP_401,
            msg=f"로그인이 필요한 서비스 입니다.",
            detail="Authorization Required",
            code=f"{StatusCode.HTTP_401}{'1'.zfill(4)}",
            ex=ex,
        )


class TokenExpiredEx(APIException):
    def __init__(self, ex: Exception = None):
        super().__init__(
            status_code=StatusCode.HTTP_400,
            msg=f"세션이 만료되어 로그아웃 되었습니다.",
            detail="Token Expired",
            code=f"{StatusCode.HTTP_400}{'1'.zfill(4)}",
            ex=ex,
        )


class TokenDecodeEx(APIException):
    def __init__(self, ex: Exception = None):
        super().__init__(
            status_code=StatusCode.HTTP_400,
            msg=f"비정상적인 접근입니다.",
            detail="Token has been compromised.",
            code=f"{StatusCode.HTTP_400}{'2'.zfill(4)}",
            ex=ex,
        )

정의된 클래스는 StatusCode, APIException, NotFoundUserEx, NotAuthorized, TokenExpiredEx, TokenDecodeEx 해서 6개가 있어요. StatusCode는 에러코드를 정리한 클래스, APIException은 나머지 4개의 부모클래스로 작동할건데요. Exception클래스를 부모클래스로 상속받아 정의할거에요.

클래스명을 보고도 어떤 에러인지 한눈에 알아 볼수 있는 직관적인 이름으로 정의했어요.

NotFoundUserEx클래스에서 주목할 만 한 점은 super().__init__()부분인데요. 부모클래스에서 상속 받은 속성들을 바꿔 적용시키게 되요.

class NotFoundUserEx(APIException):
    def __init__(self, user_id: int = None, ex: Exception = None):
        super().__init__(
            status_code=StatusCode.HTTP_404,
            msg=f"해당 유저를 찾을 수 없습니다.",
            detail=f"Not Found User ID : {user_id}",
            code=f"{StatusCode.HTTP_400}{'1'.zfill(4)}",
            ex=ex,
        )

사실 Exception은 많아지게 되면 수백개가 되므로 추후 Database에 담아서 관리하게 됩니다.

users.py
해당 소스 코드를 살펴 볼게요.
response_model=UserMe가 reouter.get()데코레이터의 2번째 키워드 인자로 들어 있어요. 만약 저 부분을 누락하게 되면 database/schema.py에 정의된 User클래스의 속성들을 모두 보여주게되는 우를 범하게되요.

from fastapi import APIRouter
from starlette.requests import Request

from app.database.schema import Users
from app.errors.exceptions import NotFoundUserEx
from app.models import UserMe

router = APIRouter()


@router.get('/me', response_model=UserMe)
async def get_user(request: Request):
    '''
    get my info
    :param request:
    :return:
    '''
    user = request.state.user
    user_info = Users.get(id=user.id)
    return user_info

request.state.user부분은 token_validator 미들웨어에서 딕셔너리를 객체화 시킨 이후에 사용가능 하게 바뀐거에요.(바로 아래 소스코드 참조)

token_info = await self.token_decode(access_token=request.headers.get("Authorization"))
                    request.state.user = UserToken(**token_info)

users.py의 소스코드 부분 리뷰를 이어가서, router.get()매개변수의 response_model은 참 중요한데요. pydantic의 도움을 받아서 정의된 models.py의 UserMe클래스가 핵심이되요. 만약 response_model=UserMe가 없으면 pw변수까지 모두 보여주기 때문에 거르는 역할이 아예 사라지게 되요.

[response_model이 있는 경우]

[response_model이 없는 경우]

token_validator.py

...
from app.errors import exceptions as ex
...

	try:
	
    		if request.url.path.startswith("/api"):
                # api 인경우 헤더로 토큰 검사
                	if "authorization" in request.headers.keys():

    			if "Authorization" not in request.headers.keys():
   				raise ex.NotAuthorized()
...

tokne_validator모듈에서 if await self.url_pattern_check()이후 try문 안쪽 else문의 첫 번째 if문 'Authorization' not in request.headers.keys(): 조건에서NotAuthorized 오류를 발생시키는데요. 그리고 이후 해당 로직은 except APIException as e에서 exception_handler(e) 메소드를 비동기로 처리하게되요.

아래 스태틱 메소드는 AccessControl의 staticmethod는에요.
내부 로직은 결국 diction형태로오류 코드

    @staticmethod
    async def exception_handler(error: APIException):
        error_dict = dict(status=error.status_code, msg=error.msg, detail=error.detail, code=error.code)
        res = JSONResponse(status_code=error.status_code, content=error_dict)
        return res

error.status_code의 경우 이미 error패키지의 exceptions모듈에 코드들이 정의되어 있습니다.

또한 추후 진행 될 사항이지만 __call__메서드의 마지막 부분에 loggin 처리를 finally 키워드 안쪽에서 해줄거에요.

        except APIException as e:
            res = await self.exception_handler(e)
            res = await res(scope, receive, send)
        finally:
            # Logging
            ...
        return res

reference - https://www.youtube.com/watch?v=tFyE7kw-AfA&t=134s