REAL Python – FastAPI] – “풀스택 블로그 개발하기 – 회원목록 조회 API 구현하기”
REAL Python – FastAPI] – “풀스택 블로그 개발하기 – 회원목록 조회 API 구현하기”
요구사항 정리
이번에는 회원 목록 조회 API를 구현하며, FastAPI 에서 제공하는 기능들을 하나하나 맛봅니다. 생각해 보면 “회원” 에는 여러 종류가 있을 수 있죠. 소셜 회원가입, 이메일로 회원가입, 닉네임으로 회원가입 등이 있죠. 회원가입 하고 나서도 그 뿐인가요, 블로그에 글을 작성할 수 있는 등급, 읽기만 할 수 있는 등급, 블로그를 수정할 수 있는 등급 등 등급 관련해서도 복잡합니다.
코드 작성 전 이렇게 “정확히 무엇을 해야 하는가?” 를 정해두고 시작하는 것이 좋겠습니다.
- 먼저 회원가입은 “닉네임, 이메일, 비밀번호, 비밀번호 재입력” 을 입력함으로서 진행하는 것으로 정합시다.
- 닉네임과 이메일은 중복이 없어야 합니다.
- 닉네임은 특수문자가 없는 영어&숫자로만 이루어진 25자 이하의 문자열이어야 합니다.
- 이메일은 실제 존재하는 이메일이어야 합니다.
- 비밀번호는 입력 시 여덟 자 이상이어야 합니다.
- 비밀번호는 암호화되어 저장되어야 합니다.
- 회원 등급은 “에디터”, “관리자”, “독자” 세 가지가 있는 것으로 합시다.
- “에디터”는 블로그에 새로운 글을 발행하거나, 수 있습니다.
- “관리자” 는 저자의 역할을 포함해 회원 관리, 블로그 메타 정보를 수정하는 등의 작업을 수행할 수 있습니다.
- “독자” 는 댓글을 달거나, 우리 블로그를 구독하는 등의 작업을 수행할 수 있습니다.
좋아요, 이제 뭘 해야 하는지 명확해졌습니다.
User 모델 정의하기
회원 목록
이전 시간, 우리는 데이터베이스를 다루기 위해 SQL 코드를 직접 다루지 않고, Python 의 클래스 문법을 통해 객체로서 다루는 방법을 알아봤죠? 실제 “회원” 이라는 모델을 Python 코드로 구현할 시간입니다. users/models.py 에 아래의 코드를 입력합니다.
from datetime import datetime
from enum import Enum as PythonEnum
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Enum
from config.db import Model
class UserRole(str, PythonEnum):
ADMIN = "admin"
EDITOR = "editor"
READER = "reader"
class User(Model):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
username = Column(String(25), nullable=False, unique=True)
email = Column(String(255), nullable=False, unique=True)
password = Column(String(255), nullable=False)
created_at = Column(DateTime, default=datetime.now)
is_active = Column(Boolean(), default=False)
role = Column(Enum(UserRole), default=UserRole.READER)
이는 우리가 만들고자 하는 User 테이블을 Python 코드로 나타낸 것입니다.
- 테이블 이름을
user로 지정했고, username, email, password와 같은 필드들이 존재합니다.- 그 중,
role필드는 “관리자, 에디터” 만 존재해야 하므로Enum으로 관리하고 있는 걸 확인할 수 있습니다.
이제 모델이 정의되었으니, 이를 실제로 테이블로 만들 시간입니다. 이 파트에서 우리는 이전 시간에 설정해 뒀던 alembic 을 활용할 수 있습니다. alembic 이 “어떤 모델을 관리해야 하는지” 알 수 있게, env.py 에 아래의 코드를 추가합니다.
그리고, 터미널에 아래의 명령어를 입력해 주세요.
alembic revision --autogenerate -m "create user table"
그러면 migrations 폴더 아래에 새로운 파일이 만들어졌음을 확인할 수 있을 겁니다.
실제 위의 파일 내용을 읽어보면.. 아래와 같습니다. 이건 뭘 의미하는 걸까요?
테이블을 생성하라고 말하는 것만 같이 이름붙여진 create_table 메서드, 우리가 정의한 필드들이 나열되어 있습니다. 뭔가 우리의 모델을 기반으로 생성된 코드임은 분명합니다. 이것이 실제로 수행하는 것은 아래의 명령어를 터미널에 입력함으로서 확인해볼 수 있습니다.
alembic upgrade head --sql
이제 조금 익숙한 구문이 등장합니다.
- 위의 첫 번째
CREATE TABLE alembic_version ...은 데이터베이스 버전 관리를 위해alembic이 만들어내는 테이블입니다. - 두 번째
CREATE TABLE user ...은 우리의SQLAlchemy클래스가 실제로SQL로 변환된 것입니다. 클래스 정의와 위의SQL을 확인해 보세요. 중요한 것은, 작성한 파이썬 코드가 SQL로 바뀌었다는 것입니다. - 세 번째
INSERT에서는 우리가 만들어낸 변경의 버전을 기록하고 있네요.
이제, 위의 SQL 을 실제로 수행하려면 아래의 명령어를 입력하면 됩니다.
alembic upgrade head
이제 https://sqlitebrowser.org/dl/ 와 같은 도구를 활용해 app.db 데이터베이스를 열어 보세요. 아래와 같은 – 무미건조하지만 좋은 시작을 알리는 테이블 하나가 생겼을 겁니다.
사용법은 https://datacarpentry.org/sql-socialsci/02-db-browser.html 를 참고하세요!
필자는
PyCharm의DB연결 툴을 그대로 사용합니다.
그러면 실제로 데이터베이스에 첫 번째 회원을 가입시켜 볼까요? 터미널에서 poetry add ipython -G development 로 ipython 을 설치하고, ipython 을 입력하세요.
그리고, 아래의 코드를 입력해 보세요.
from config.db import AsyncSessionLocal
from api.users.models import User
db = AsyncSessionLocal()
user = User(
username="test",
email="test@test.com",
password="test",
is_active=True,
role="admin",
)
db.add(user)
await db.commit()
그러면, 위와 같은 INSERT 문이 제대로 동작하고 새로운 회원이 저장된 것을 확인할 수 있을 겁니다.
좋아요. 이제 클라이언트의 HTTP 요청에 따라, ORM 을 사용하여, 회원 정보를 조회하거나 생성하는 등의 작업을 하면 되겠네요.
FastAPI 의 Route
보통, 웹 API 는 URL 또는 URI 를 분석하여 어떤 응답을 할 지 결정하게 됩니다. POST: api/users/ 와 같이 HTTP method, URL 등을 분석하고 “새로운 회원을 생성해야겠구나!” 등과 같은 응답을 결정하는 과정을 라우팅이라고 합니다. 사실 우리는 맨 처음부터 FastAPI 가 제공하는 라우팅을 이용하고 있었습니다!
이게 어떤 말인지, 잠시 app.py 로 이동해 보겠습니다.
이전 시간 우리가 어떻게 기본적인 API 를 정의했는지 기억하시나요? 아래와 같은 형태였습니다:
app.get의("/")은HTTP API의 경로를 나타냅니다.http:127.0.0.1:8000/경로에 대한 처리를 담당한다는 의미입니다.- 그 밑,
async def로 나타내어지는함수는 딕셔너리 자료형을 리턴합니다.FastAPI는 이런 자료형을json으로 변환해HTTP응답을 만들어내는 역할을 하죠.
이런 경우는 동작하긴 하지만, 프로젝트의 규모가 커짐에 따라 엄청난 길이가 생기게 됩니다. 아마 여러분은 이런 파일을 건들고 싶지 않을 겁니다. 유지보수를 하려면 ctrl+F 신공과 함께 다른 사람들이 작업한 것을 건들지 않도록 충돌까지 조심하며 작성해야겠죠.
FastAPI 에서는 이러한 경우를 위해, 파일과 관심사를 분리하여 작성할 수 있도록 도와주는 내장 클래스, APIRouter 를 제공합니다. 그게 대체 뭔지, 코드로 알아봅시다.
api/routers.py 에 아래의 코드를 작성합니다.
from fastapi import APIRouter, Depends
users_router = APIRouter()
@users_router.get("/")
async def get_users():
return [
{"id": 1, "name": "tgoddessana"},
{"id": 2, "name": "smith"},
]
3줄에 users_router 를 작성하고, 이전에 경로를 등록했던 것과 똑같이 get() 매핑을 한 후 함수를 작성했습니다.
from fastapi import FastAPI
from api.users.routers import users_router
from config.settings import settings
app = FastAPI(
title=settings.TITLE,
description=settings.DESCRIPTION,
)
app.include_router(users_router)
그리고, app.py 에서 위처럼 라우터를 등록하면 우리가 작성한 라우터 경로들은 앱에 잘 등록되어 있을 겁니다.
맞아요, 이렇게 FastAPI 는 APIRouter 객체를 통해 라우팅을 등록하고 앱을 작성할 수 있게 해 줍니다.
SQLAlchemy 로 실제 회원 목록 불러오기
위의 API 구현은 하드코딩된 딕셔너리를 리턴할 뿐, 실제 데이터베이스에 존재하는 회원의 정보를 반환하는 것이 아닙니다. 이제 실제 데이터베이스에서 회원 정보를 가져와 조회하는 API 를 작성해봅시다.
SQLAlchemy 의 Session
이를 수행하기 위해서는 SQLAlchemy 의 Session 객체가 필요합니다. Session 이 대체 뭔지, 왜 필요한지에 대해서 잠깐 짚어보겠습니다. (실습하지 않아도 되는 코드입니다)
new_user = User(
username="tgoddessana",
email="twicegoddessana1229@gmail.com",
password="password",
is_active=True,
role="admin",
)
위는 우리가 작성했던 User 클래스의 인스턴스를 생성하는 코드입니다. 그리고 User 클래스는 SQLAlchemy 의 일부분이었죠. 여러분의 컴퓨터에 설치되어 있는 파이썬 인터프리터가 위의 코드를 실행하면 어떤 일이 일어날까요? 테이블과 매핑되는 클래스의 인스턴스를 만들었으니, 위의 정보를 가지는 새로운 회원을 데이터베이스에 저장할까요?
어림도 없습니다. 왜냐면 위에서 파이썬 인터프리터는 단지 User 인스턴스를 메모리에 올리기만 하기 때문입니다. 데이터베이스와의 접속은 이뤄지지 않죠. 위의 변경사항을 실제로 데이터베이스에 커밋하려면 Session 이라는 것을 사용해야 합니다.
..위처럼 AsyncSession 객체를 이용해야 하죠. 어쩌면 위와 같은 세션 객체를 이용하는 것이 Django 를 사용해보셨던 분들에게는 익숙하지 않을 수 있습니다. Django 가 데이터베이스의 연결을 암시적으로 처리하는 반면, SQLAlchemy 는 그렇지 않기 때문이죠. 공식 문서의 “세션이 하는 일은 무엇인가?” 를 보면 더 확연히 그것이 어떤 것을 수행하는지를 알 수 있습니다.
In the most general sense, the
Sessionestablishes all conversations with the database and represents a “holding zone” for all the objects which you’ve loaded or associated with it during its lifespan. It provides the interface where SELECT and other queries are made that will return and modify ORM-mapped objects. The ORM objects themselves are maintained inside theSession, inside a structure called the identity map – a data structure that maintains unique copies of each object, where “unique” means “only one object with a particular primary key”.가장 일반적인 의미에서 세션은 데이터베이스와의 모든 대화를 설정하고 수명 기간 동안 로드하거나 연결된 모든 객체에 대한 “보유 영역”을 나타냅니다. 이 영역은 ORM 매핑된 객체를 반환하고 수정하는 SELECT 및 기타 쿼리가 수행되는 인터페이스를 제공합니다. ORM 객체 자체는 각 객체의 고유한 복사본을 유지하는 데이터 구조인 ID 맵이라는 구조 내부의 세션 내부에서 유지되며, 여기서 “고유”는 “특정 기본 키를 가진 객체가 하나뿐”이라는 의미입니다.
https://docs.sqlalchemy.org/en/20/orm/session_basics.html
Session 사용하기
이제 실제로 Session 을 사용할 차례입니다. 우리가 두 번째 강좌에서 그냥 Session 객체가 아니라 AsyncSession 객체를 만들었던 것을 기억하시나요? SQLAlchemy 는 버전 1.4 부터 Asynchronous 지원을 추가했습니다. 이번 프로젝트에서는 이를 적극 활용하여 데이터베이스 작업을 수행하겠습니다. routers.py 에 아래의 코드를 작성합니다.
from fastapi import APIRouter
from sqlalchemy import select
from config.db import AsyncSessionLocal
from api.users.models import User
users_router = APIRouter()
@users_router.get("/")
async def get_users():
# AsyncSession 객체 생성
db = AsyncSessionLocal()
# AsyncSession 객체를 통해 DB 연결 후 쿼리 실행
users = await db.execute(select(User))
# 쿼리 실행 후 DB 연결 종료
await db.close()
# 쿼리 결과 반환
return {"data": users.scalars().all()}
- 위의 코드에서는 경로 요청마다
AsyncSession객체를 만든 다음, 세션을 통해 데이터를 쿼리하고, 세션을 닫습니다. - 그리고, 쿼리한 결과를 딕셔너리에 담아 리턴합니다.
FastAPI는 위의 딕셔너리를Json응답으로 잘 변환해 줄 겁니다.
..그리고 Swagger 문서에서 API 호출을 해 보면..
(데이터베이스에 회원이 존재한다면) 위와 같이 응답을 해 줄 겁니다. 하지만 우리가 작성한 코드에서는 몇 가지 개선할 점도 보입니다.
- 세션 객체를 만들고, 닫는 작업은 반복되는 작업입니다. 일반적으로, 중복되는 코드는 개발자를 불편하게 합니다.
- 불필요한 속성을 노출하고 있습니다. 예컨대
pasword와 같은 정보는 노출되면 안 되는 정보입니다.
이 문제들을 하나하나 해결해 가 봅시다.
FastAPI 의 Depends 로 의존성 주입하기
이 섹션에서는, 세션 객체를 만들고, 닫는 반복되는 코드 를 자동화합니다. 이를 수행하기 위해서, 우리는 먼저 FastAPI 에서 제공하는 기능인 의존성 주입 에 대해 알아볼 필요가 있습니다.
routers.py 에 아래와 같은 간단한 함수를 정의해 봅시다.
..아마 Python 문법에 전문가가 아닌 사람이어도 위의 함수가 호출된다면 어떤 일이 일어날 지 예측할 수 있을 겁니다. 콘솔 어딘가에 전문가스러워 보이는 로그가 기록되겠죠?
# from typing import Callable, Annotated 에서 import!
# from fastapi import APIRouter, Depends 에서 import!
@users_router.get("/")
async def get_users(logger: Annotated[Callable, Depends(log_request)]):
# AsyncSession 객체 생성
db = AsyncSessionLocal()
# AsyncSession 객체를 통해 DB 연결 후 쿼리 실행
users = await db.execute(select(User))
# 쿼리 실행 후 DB 연결 종료
await db.close()
# 쿼리 결과 반환
return {"data": users.scalars().all()}
이제 작성해뒀던 코드를 위처럼 바꿉니다. 우리는 2번째 줄을 주목할 필요가 있습니다:
async def get_users(logger: Annotated[Callable, Depends(log_request)]):
typing.Annotated를 통해서 타입 힌트와 함께 메타데이터를 추가했습니다. 첫 번째Callable으로 “logger는 호출 가능한 객체이고,Depends(log_request)라는 메타데이터를 가지고 있어!” 를 알려주고 있네요.Depends는FastAPI에서 의존성 주입을 가능케끔 하는 함수입니다. “이 함수(get_users) 는log_request함수가 있어야 합니다! 를FastAPI애플리케이션에 알려주는 역할을 하죠.
위의 코드를 저장하고, 브라우저에서 새로운 요청을 보내 보세요. 콘솔에 아래와 같은 내용이 출력될 겁니다.
여기서 알 수 있는 사실은 우리가 Depends 를 사용함으로서, Depends 안에 들어 있는 호출 가능한 객체인 log_request 가 FastAPI 애플리케이션에 의해 요청 전 호출되었다는 것입니다. 무언가 공통적으로 필요한 기능이 있거나, 반복되는 코드가 있으면 위처럼 주입한 다음 사용하면 됩니다.
그런 의미에서, 우리는 “새로운 세션 객체를 만들고, 최종적으로 세션을 닫는 것” 을 의존성 주입을 통해 해결할 수 있습니다. 데이터베이스와의 연결이 필요하면 필히 있어야 하는 반복되는 코드죠. config/db.py 에 아래의 내용을 추가로 작성합니다.
from typing import AsyncGenerator
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import (
create_async_engine,
AsyncSession,
async_sessionmaker,
)
from sqlalchemy.orm import declarative_base
from config.settings import settings
engine = create_async_engine(
settings.DB_URL, echo=True, connect_args={"check_same_thread": False}
)
AsyncSessionLocal = async_sessionmaker(engine, autocommit=False, class_=AsyncSession)
naming_convention = {
"ix": "%(column_0_label)s_idx",
"uq": "%(table_name)s_%(column_0_name)s_key",
"ck": "%(table_name)s_%(constraint_name)s_check",
"fk": "%(table_name)s_%(column_0_name)s_fkey",
"pk": "%(table_name)s_pkey",
}
Model = declarative_base(metadata=MetaData(naming_convention=naming_convention))
#######
# 추가 #
#######
async def get_session() -> AsyncGenerator[AsyncSession, None]:
session = AsyncSessionLocal()
try:
yield session
finally:
await session.close()
위의 get_session 함수는 대체 뭘까요? 조금은 생소해 보이는 yield 라는 키워드가 보입니다. 타입 힌트를 보아하니, AsyncGenerator 타입의 인스턴스를 반환하고, 그 AsyncGenerator 는 AsyncSession 객체를 반환하는 것으로 보이네요.
우리가 위에서 session.close 를 수행했던 것을 기억하시나요? 이는 의존성으로 “무언가를 얻는 것(Session 객체)” 도 중요하지만, “무언가를 수행하는 것(Session 을 닫는 것)” 도 필요하다는 것을 의미합니다. 다행히 FastAPI 는 요청이 완료된 후 몇 가지 동작을 수행할 수 있게 해 주는 종속성도 제공하죠. 이를 위해서 return 대신, yield 가 필요합니다.
결국, 아래와 같은 모양새인 거죠.
그러면 실제로 이를 사용해 볼까요? router.py 의 get_users 함수를 아래와 같이 수정합니다. (이전의 logger 함수는 없애도 괜찮습니다.)
@users_router.get("/")
async def get_users(session: Annotated[AsyncSessionLocal, Depends(get_session)]):
results = await session.execute(select(User))
users = results.scalars().all()
return {"data": users}
그리고 실제 요청을 보내 보면, 동작이 제대로 되고 있음을 확인할 수 있습니다. 조금 더 명확히 해 보자면,
async def get_session() -> AsyncGenerator[AsyncSession, None]:
session = AsyncSessionLocal()
try:
print("session created")
yield session
finally:
print("session closed")
await session.close()
아래와 같이 제대로 session.close() 가 동작한 거죠.
좋아요! “중복되는 코드가 있다” 는 해결했습니다.
Schema 로 응답 스키마 정의하기
여전히 우리의 API 에 남아있는 문제는 의도치 않게 모든 회원 정보를 노출한다는 것입니다. 우리가 만들고자 하는 사이트의 정의에 따라 노출해야 하는 정보와 노출하면 안 되는 정보는 바뀔 겁니다. 이러한 문제를 해결하기 위해서, 우리는 Pydantic 을 사용할 수 있습니다.
Pydantic은Python에서 데이터 검증을 위해 자주 사용되는 빠른 라이브러리입니다.
api/users/schemas.py 에 아래의 내용을 입력합니다.
from datetime import datetime
from typing import Sequence
from pydantic import BaseModel, EmailStr
class UserListResponse(BaseModel):
class _User(BaseModel):
id: int
username: str
email: EmailStr
is_active: bool
created_at: datetime
role: str
data: Sequence[_User]
위의 조그마한 클래스는 보기에 굉장히 편안하죠? 타입 힌트 문법으로 “어떤 필드는 어떤 형식이어야 해!” 를 알려주고 있습니다.
UserListResponse클래스 안에_User클래스를 정의했습니다. 클래스 내부에서 사용하기 위함입니다.data라는 클래스 변수에Sequence[_User]를 입력함으로서, “data라는 키에Sequence타입의_User들이 들어올 것이다” 라고 선언하고 있습니다._User클래스 안에는 한 명의 회원이 어떻게 표현될 것인가를 정의했는데, 불필요한 필드인password데이터는 직렬화하는 필드에서 빠지도록 정의했습니다.
응답 형식을 정의했으니 이를 사용해 봐야겠죠? 라우터의 코드를 아래와 같이 변경합니다.
from typing import Annotated, Any
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from api.users.schemas import UserListResponse
from config.db import get_session
from api.users.models import User
users_router = APIRouter()
@users_router.get("/", response_model=UserListResponse)
async def get_users(session: Annotated[AsyncSession, Depends(get_session)]) -> Any:
results = await session.execute(select(User))
users = results.scalars().all()
return {"data": users}
위처럼 14줄에서 response_model 을 정의하면, get_users 가 반환하는 값을 UserListResponse 가 직렬화한 후 응답하도록 할 수 있습니다.
- 실제 호출을 해 보면, 위에서 정의하지 않은 필드인
password는 잘 숨겨져 있네요.
보너스 섹션: SQL echo 옵션 설정하기
하나의 API 호출이 있을 때마다 콘솔에 SQL 이 프린트되는 이유는 아래의 설정 때문입니다.
이를 환경 변수로 관리해 봅시다.
config/settings.py 에 아래의 코드를 추가합니다.
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env")
TITLE: str = "Shovelog Backend API"
DESCRIPTION: str = "Backend API for Shovelog."
DB_URL: str
ECHO_SQL: bool = False # 추가!
settings = Settings()
그리고, .env 에 해당 키에 대한 값을 작성합니다.
db.py 에는 아래처럼 settings 에서 값을 읽어오면 됩니다.
그러면 요청이 있더라도 SQL 이 콘솔에 표시되지 않습니다.
요약
- 이번 시간에는 회원 목록 조회를 구현했습니다.
- 이전에 소개했던
Python ORM인SQLAlchemy로 회원 테이블을 위한 클래스를 정의하고, 그 클래스가 어떤SQL로 변환되는지를 확인하고, 테이블을 만들었습니다. SQLAlchemy의AsyncSession객체를 이용하여 데이터베이스의 연결을 관리하고, 접속하여 조회 쿼리를 실행했습니다.FastAPI가 제공하는 의존성 주입 기능인Depends()함수를 이용하여, 세션을 열고 세션을 닫는 코드의 중복을 제거했습니다.Pydantic을 이용해, 우리가 원하는 필드만 노출할 수 있도록 스키마를 작성하고 이용했습니다.
이제 기본적인 흐름과 조회 방법을 알았으니, 다음 시간에는 더욱 복잡한 기능인 회원가입을 구현해 보겠습니다.