fastapi - G6 간단하게 훓어보기 - 2) API 호출

1. 쪽지 보내기

   - 쪽지 보내기는 "/memo_form_update" 엔드포인트로 동작을 한다.

   - memo_service: Annotated[MemoService, Depends()] 파라미터로 생성된 MemoService 클래스를 이용해 쪽지와 관련한 처리를 수행한다. 다른 메모 관련 엔드포인트들도 마찬가지이다.

   - point_service: Annotated[PointService, Depends()], 파라미터로 생성된 PointService 클래스를 이용해 포인트 관련한 처리를 수행한다.

   - member: Annotated[Member, Depends(get_login_member)] 는 쪽지를 보내는 사람에 대한 사용자 정보를 DB 테이블에서 가져온다.

   - 테스트 결과 쪽지를 받는 사람은 로그인이 되어 있거나, 로그인을 바로 한 경우 어떤 경우에도 도착한 쪽지에 대한 이벤트가 동작하는 기능은 없다.

      - 코드를 보면 사용자의 mb_memo_call 필드를 가지고 실시간 쪽지 알림을 관리한다. 실제 동작 여부는 front 코드와 함께 확인이 필요할 것 같다.

   - 모든 쪽지는 class Memo(Base) 테이블에 저장된다. 

   - me_send_id는 me_id와 동일한 값이다. 보낸 사람의 id이다. 받는 사람의 메모만 me_send_id를 입력한다. me_send_id 는 쪽지 읽음 처리를 할때 사용된다. 

      - 왜 굳이 중복으로 처리할까 생각했는데 코드를 보면 me_type='send'로 보내는 사람의 쪽지를 저장하고, me_type='recv'로 받는 사람의 쪽지를 저장한다. 보내는 사람의 쪽지가 제대로 저장되었다면을 가정하고 반환값을 받는 memo_send를 이용해 me_id를 저장한다.

      - 이렇게 사용하려면 refresh를 사용해야 안전하다. 그리고 굳이 이렇게 사용하는 이유가 공감이 안된다.

def send_memo(self, member: Member, target: Member, memo: str) -> Memo:
        """쪽지를 전송합니다."""
        try:
            memo_dict = {
                "me_send_mb_id": member.mb_id,
                "me_recv_mb_id": target.mb_id,
                "me_memo": memo,
                "me_send_ip": get_client_ip(self.request),
            }
            memo_send = Memo(me_type='send', **memo_dict)
            self.db.add(memo_send)
            self.db.commit()
            memo_recv = Memo(me_type='recv', me_send_id=memo_send.me_id, **memo_dict)
            self.db.add(memo_recv)
            self.db.commit()
            return memo_send
        except SQLAlchemyError as e:
            self.db.rollback()
            self.raise_exception(500, str(e))

 

2. 현재 접속자

   - 현재 접속자에 대한 처리가 궁금해서 분석하기로 했다.

 

   1) 라우터 정리

      - /api/v1/의 모든 라우터는 api/v1/routers/__init__.py에 정의되어 있다.

      - /api/a1로 접근하는 모든 url은 두개의 의존성을 필수로 수행한다.

         - check_use_api는 api 사용 여부를 확인한다. .env에 정의되어 있다.

         - set_current_connect는 현재 접속자에 대해 다음과 같은 작업을 수행한다.

            - 현재 사용자 정보를 갱신한다.

            - 사용자 정보가 없다면 새로 생성한다.

             - .env에 정해진 시간을 기준으로 시간을 넘어간 사용자의 정보는 삭제된다. 기본은 10분이다.

# api/v1/routers/__init__.py
router = APIRouter(prefix="/api/v1",
                   dependencies=[Depends(check_use_api),
                                 Depends(set_current_connect)])

 

   2) 엔드포인트

      - 현재 접속자 목록 조회는 "/members/current-connect" 엔드포인트로 정의되어 있다.

      - CurrentConnectListRequest이름의 pydantic을 의존서으로 주입받는다. 

data: Annotated[CurrentConnectListRequest, Depends()]

 

      - PagenationRequest 부모 pydantic을 상속받고 only_member 필드를 추가한다.

class CurrentConnectListRequest(PagenationRequest):
    """현재 접속중인 회원 목록 조회 요청 모델"""
    only_member: str = Field(
        Query(default="N",
             title="접속 회원만 표시",
             description="Y: 접속 회원만 표시, N: 전체 표시",
             example="N"))

 

      - 페이지네이션과 관련한 기본 pydantic 모델이 정의된 것을 확인할 수 있다.

class PagenationRequest(BaseModel):
    """페이징 요청 모델"""
    page: int = Field(
        Query(default=1,
              ge=1,
              title="페이지 번호",
              description="가져올 결과의 페이지 번호")
    )
    per_page: int = Field(
        Query(default=10,
              ge=1,
              le=100,
              title="출력 수",
              description="페이지 당 결과 수(최대 100)")
    )

    @property
    def offset(self) -> int:
        """페이지 번호에 따른 offset을 계산합니다."""
        return (self.page - 1) * self.per_page

 

      - property를 제외한 모든 필드는 url의 쿼리를 검증하기 위한 정의가 되어있다.

      - 클래스명, 함수명, 변수명 등에 사용된 단어들이 직관적이지 못한것 같다. 이름만 보아서는 이해가 어렵다.

 

      - CurrentConnectService 클래스는 재 접속자 관련 서비스를 제공하는 종속성 주입 클래스이다.

      - fetch_total_records에 대해서 캐시를 적용하고 있다.

      - 24초 동안 이 코드는 **LRU 캐시(LRU Cache)**를 사용하여 특정 함수의 결과를 캐싱하는 데 사용된다.

@cached(LRUCache(maxsize=1),
            key=lambda self, only_member=False: hashkey("connects_count", only_member))
    def fetch_total_records(self, only_member: bool = False) -> int:
        """현재 접속중인 회원의 총 수를 반환합니다."""
        query = self._base_query(only_member)

        return self.db.scalar(query.add_columns(func.count(Login.mb_id)))