[가계부 만들기] 카테고리 설정 - 구조/리스트조회

가계부에서 수입/지출 항목을 입력하는 기능을 만들기 전에 먼저 카테고리 설정에 관련된 기능을 구현하고자 한다.

모든 수입/지출 내역은 카테고리로 정리할 수 있을 것이다. 물론 카테고리 항목이 비어있을 수도 있겠지만 카테고리를 정리해두면 추후에 통계 자료를 만들 때 활용할 수 있는 여지가 많다.

다년간 개발을 하다보니 원 데이터가 상세하면 이를 가공해서 만들 수 있는 통계도 훨씬 다양하게 만들 수 있다.

이전 내용에서 회원 가입시 기본 카테고리를 자동으로 생성해주는 코드를 붙여두었다.

기본적인 카테고리 정보가 있으니 먼저 카테고리 리스틑 가져오는 것부터 시작해서 카테고리를 새로 생성하고, 수정하고, 삭제하는 기능을 구현하도록 할 것이다.

카테고리는 하나의 카테고리 정보가 때어내어서 관리하는 일보다는 리스트로 전체를 조회해서 수정하고 삭제하는 기능을 구현하게 될 것이다.

728x90

Category 기본 구조

먼저 카테고리 관련 기능을 구현하기 전에 필요한 파일을 만들 것이다.

이전 포스팅에서 services/category_service.py 파일을 만들어 두었으니 router 파일을 생성하면 된다.

routers/category.py 파일을 생성하고 다음과 같이 입력한다.

from fastapi import APIRouter, Depends
from ..libraries.auth import decode_access_token

router = APIRouter(
    prefix="/category",
    tags=["category"],
    dependencies=[Depends(decode_access_token)],
    responses={
        404: {"description": "Not Found"},
    }
)

기본적인 router 정보를 입력하였다. 이 구조는 이전 routers/member.py과 유사하다.

다만 member.py에서는 로그인을 하지 않은 상태에서 접근하는 경우가 있기 때문에 dependencies를 router를 선언할 때 지정하지 않았다. 하지만 그 외의 router는 기본적으로 로그인을 했다는 것을 가정하고 개발하기 때문에 router를 선언할 때 dependencies를 지정하였다.

선언한 category.py의 라우터를 main에 등록해주어야 한다. main.py 파일을 열어서 아래와 같이 작성한다.

반응형

...생략...
from .routers import members, category


app = FastAPI(title="account-book-api")

app.include_router(members.router)
app.include_router(category.router)

...생략...

라우터를 등록하였으니 category.py에서 정의하는 router 정보를 이제 사용할 수 있게 되었다.

Schema 정의

이전 포스팅에서 models.py에서 category 테이블을 아래와 같이 정의하였다.

반응형

class Category(Base):
    __tablename__ = "tb_category"

    category_no = Column(BIGINT(unsigned=True), nullable=False, autoincrement=True, comment="카테고리 번호")
    member_no = Column(SMALLINT(unsigned=True), nullable=True, comment="카테고리 생성자 번호(기본 카테고리일 경우 null)")
    category_name = Column(VARCHAR(length=50), nullable=False, comment="카테고리 이름")
    has_children = Column(CHAR(length=1), nullable=False, default="F", comment="자식을 가지고 있는지 여부(T|F)")
    inout_type = Column(CHAR(length=1), nullable=False, default="O", comment="수입/지출 구분(I: incomes, O: outgoings)")
    parent_no = Column(BIGINT(unsigned=True), nullable=True, comment="부모 카테고리 번호")
    class_name = Column(VARCHAR(length=30), nullable=True, comment="아이콘 클래스")
    sort_order = Column(TINYINT(unsigned=True), default=1, comment="정렬순서")
    reg_dt = Column(DATETIME(timezone=False), nullable=False, server_default=func.now(), comment="생성일시")
    upd_dt = Column(DATETIME(timezone=False), nullable=False, server_default=func.now(), onupdate=func.now(), comment="수정일시")
    is_deleted = Column(CHAR(length=1), default="F", server_default="F", nullable=False, comment="삭제여부(T|F)")
    del_dt = Column(DATETIME(timezone=False), nullable=True, comment="삭제일시")

    __table_args__ = (
        PrimaryKeyConstraint(category_no, name="pk_category"),
        ForeignKeyConstraint(
            ["member_no"],
            ["tb_members.member_no"],
            name="fk_category__member_no",
            onupdate="NO ACTION",
            ondelete="NO ACTION",
        ),
        {
            "comment": "카테고리 정보"
        }
    )

이제 여기에 맞는 Schema를 정의할 것이다. 다른 내용은 생략하고 category 관련된 내용만 정의한다.

database/schemas.py 파일을 열어서 다음 내용을 추가한다.

반응형

class CategoryUpsert(BaseModel):
    category_name: str = Field(title="카테고리 이름")
    parent_no: Optional[int] = Field(title="부모 카테고리 번호", default=None)
    class_name: Optional[str] = Field(title="아이콘 정보", default=None)
    inout_type: Literal['I', 'O'] = Field(title="수입/지출구분")

class CategoryInfo(CategoryUpsert):
    category_no: int = Field(title="카테고리 번호")
    member_no: Optional[int] = Field(title="소유자 회원 번호")
    sort_order: int = Field(title="정렬순서")

    class Config:
        orm_mode = True

먼저 CategoryUpsert 클래스를 먼저 정의하고 CategoryInfo 테이블은 CategoryUpsert 테이블을 상속받아서 정의하였다.

CategoryUpsert 클래스는 카테고리를 생성하거나 수정할 때 관련 카테고리 정보를 전달받는 request 형태를 정의하였다.

카테고리를 생성하거나 수정할 때 사용자에게 받는 정보는 동일하다.

CategoryInfo 클래스는 사용자가 조희한 Category 관련 정보를 전달하는 형태를 정의한 것이다. 실제로 CategoryInfo 클래스가 Category 테이블과 연결된다.

Category 리스트 조회

가장 먼저 리스트를 조회하는 기능을 구현하도록 하겠다. 우선 service에서 기능을 구현한다.

services/category_service.py 파일에 아래 내용을 구현하였다.

반응형

from typing import Optional, List, Literal
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from ..database import schemas, models

# 카테고리 리스트 조회
async def get_category_list(
    db: AsyncSession,
    member_no: int,
    inout_type: Literal['I', 'O'],
    parent_category_no: Optional[int]
) -> List[schemas.CategoryInfo]:
    stmt = select(models.Category).filter(
        models.Category.member_no == member_no,
        models.Category.inout_type == inout_type,
        models.Category.is_deleted == 'F'
    ).order_by(models.Category.sort_order.asc(), models.Category.category_no.asc())
    if parent_category_no is None:
        stmt = stmt.filter(models.Category.parent_no.is_(None))
    else:
        stmt = stmt.filter(models.Category.parent_no == parent_category_no)

    results = await db.execute(stmt)
    return results.scalars().all()

파라미터를 정의할 때 Literal과 Optional 타입을 사용하였다. Literal은 다음에 오는 리스트 안에 있는 값만을 변수에 담을 수 있으며 그외의 값이 들어오면 Exception이 발생하도록 한다. Optional은 Union[..., None]으로 치환된다. 실제 Optional 내부가 Union으로 치환되는 코드로 구현되어 있다. 지정한 타입으로 넘어오거나 None으로 지정할 수 있는 타입이다.

리스트를 조회할 때는 사용자에게 지정되었거나 사용자가 생성한 카테고리만을 가져오기 위해서 member_no로 조회하며 수입/지출을 구분하여서 가져오도록 하였다. 정렬 순서는 정렬 순서를 저장하는 sort_order 순으로 가져오도록 하였으며, 정렬 순서가 같을 경우는 먼저 등록된 카테고리 정보 먼저 가져오도록 하였다.

만약 부모 카테고리(parent_no)가 지정되었으면 비교해서 찾고 없으면 is null로 조회하도록 처리하였다.

이렇게 조회된 리스트를 반환하도록 한다.

router는 아래와 같이 구현하였다.

반응형

from fastapi import APIRouter, Depends, Query
from typing import List, Optional, Literal
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_400_BAD_REQUEST
from ..database import schemas
from ..database.connection import get_db
from ..services import category_service
from ..libraries.auth import decode_access_token

# 카테고리 리스트 조회
@router.get('/list', description="카테고리 리스트 조회", response_model=List[schemas.CategoryInfo], responses={
    HTTP_400_BAD_REQUEST: {
        "model": schemas.Message
    }
})
async def get_list(
    user_info: schemas.JWTPayload = Depends(decode_access_token),
    inout_type: Literal['I', 'O'] = Query(title="수입/지출 구분"),
    parent_no: Optional[int] = Query(default=None, title="부모카테고리 이름"),
    db: AsyncSession = Depends(get_db)
):
    result = await category_service.get_category_list(db, user_info['member_no'], inout_type, parent_no)
    response = []
    for item in result:
        response.append(schemas.CategoryInfo.from_orm(item).dict())
    return response

router의 리스트 조회 기능은 특별히 설명할 내용이 없다.

Query로 전달된 파라미터를 통해서 category_service.get_category_list를 조회하고 결과를 리스트에 담아서 반환한다.

리스트는 GET으로 지정하였기 때문에 request는 Query에 전달할 값을 담아서 처리하도록 한다.

사용자의 member_no는 JWT에서 꺼내서 사용하기 때문에 따로 Query로 전달받지는 않는다.

여기까지 진행되었으면 리스트 관련 기능은 구현이 끝났다. 이제 테스트를 해볼 차례이다.

docs에서 리스트를 조회해보도록 한다.

반응형

리스트 조회 Reqeust

위의 이미지와 같이 구성하여서 Request를 요청하면 다음 이미지와 같이 Response를 받을 수 있다.

반응형

리스트 조회에 대한 Response

이렇게 조회가 잘 되면 리스트 기능에 대한 구현은 마무리 되었다.

나머지 카테고리 생성/수정/삭제에 대한 기능은 다음 포스팅에서 정리하도록 하겠다.