FastAPI 서버개발, 간단한 CRUD 애플리케이션 개발

본 포스트에서는 FastAPI를 활용하여 간단한 CRUD 애플리케이션을 개발하는 방법을 다룹니다. FastAPI는 최신 Python 웹 프레임워크 중 하나로, 비동기 프로그래밍을 지원하며, 자동으로 API 문서를 생성하는 기능이 포함되어 있어 개발자들에게 좋은 선택입니다. 이번 강좌에서는 FastAPI의 기본 개념을 설명하고, 실제로 쉽게 사용할 수 있는 CRUD(Create, Read, Update, Delete) 애플리케이션을 만들어 보겠습니다.

FastAPI 소개

FastAPI는 Python 3.6 이상에서 사용할 수 있는 웹 프레임워크입니다. 여기서 “Fast”라는 이름은 성능이 매우 뛰어나기 때문에 붙여졌습니다. FastAPI는 Pydantic을 통해 데이터 검증을 수행하고, Starlette를 기반으로 하여 비동기 기능을 제공합니다. API 문서는 Swagger UI와 ReDoc을 통해 자동으로 생성되며, 이는 개발자와 사용자가 API를 쉽게 이해하고 사용할 수 있도록 돕습니다.

특징

  • 빠른 성능: FastAPI는 우수한 성능을 제공하며, 이로 인해 높은 처리량을 요구하는 애플리케이션에 적합합니다.
  • 데이터 검증: Pydantic을 활용한 강력한 데이터 검증이 가능합니다.
  • 비동기 지원: 비동기식 프로그래밍이 가능하여, 높은 효율성을 보장합니다.
  • 자동 문서화: Swagger UI와 ReDoc을 통해 API 문서가 자동으로 생성됩니다.

프로젝트 설정

FastAPI를 사용하기 위해 먼저 환경을 설정해야 합니다. 아래의 단계에 따라 FastAPI 애플리케이션을 만들 준비를 하겠습니다.

1. Python 설치

FastAPI는 Python을 기반으로 하므로, 우선 Python 3.6 이상이 설치되어 있어야 합니다. Python을 설치한 후, 아래의 명령어로 FastAPI와 Uvicorn(ASGI 서버)를 설치합니다:

pip install fastapi uvicorn

2. 프로젝트 디렉토리 생성

다음으로, 프로젝트를 위한 디렉토리를 생성합니다. 예를 들어, fastapi_crud라는 이름으로 디렉토리를 만들고 이동합니다:

mkdir fastapi_crud
cd fastapi_crud

3. 애플리케이션 파일 생성

애플리케이션의 메인 파일인 main.py를 생성합니다:

touch main.py

간단한 CRUD 애플리케이션 구현

이제 FastAPI를 사용하여 간단한 CRUD 애플리케이션을 구현해 보겠습니다. 이 예제에서는 사용자 정보를 관리하는 API를 만들 것입니다. 사용자 정보는 이름과 나이로 구성되어 있습니다. 아래에 구체적인 코드를 설명합니다.

1. FastAPI 앱 설정

아래의 코드는 FastAPI 애플리케이션을 설정하고 시작하는 방법을 보여줍니다:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    age: int

2. 데이터 저장소

이 예제에서는 메모리에 사용자 데이터를 저장하는 단순한 방식으로 리스트를 사용합니다:

users = []

3. Create: 사용자 생성 API

사용자를 생성하는 API를 정의합니다. 이 API는 JSON 형식의 사용자 정보를 POST 방식으로 받아 새로운 사용자를 리스트에 추가합니다:

@app.post("/users/")
async def create_user(user: User):
    users.append(user)
    return user

4. Read: 모든 사용자 조회 API

모든 사용자 정보를 반환하는 API를 정의합니다:

@app.get("/users/")
async def read_users():
    return users

5. Read: 특정 사용자 조회 API

특정 사용자의 정보를 조회하는 API를 아래와 같이 정의할 수 있습니다:

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    for user in users:
        if user.id == user_id:
            return user
    return {"error": "User not found"}

6. Update: 사용자 정보 수정 API

특정 사용자의 정보를 수정하는 API를 만듭니다:

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: User):
    for idx, existing_user in enumerate(users):
        if existing_user.id == user_id:
            users[idx] = user
            return user
    return {"error": "User not found"}

7. Delete: 사용자 삭제 API

특정 사용자를 삭제하는 API를 정의합니다:

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    for idx, existing_user in enumerate(users):
        if existing_user.id == user_id:
            del users[idx]
            return {"message": "User removed"}
    return {"error": "User not found"}

전체 코드

위에서 설명한 모든 코드를 하나의 main.py 파일에 합쳐보겠습니다:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    age: int

users = []

@app.post("/users/")
async def create_user(user: User):
    users.append(user)
    return user

@app.get("/users/")
async def read_users():
    return users

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    for user in users:
        if user.id == user_id:
            return user
    return {"error": "User not found"}

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: User):
    for idx, existing_user in enumerate(users):
        if existing_user.id == user_id:
            users[idx] = user
            return user
    return {"error": "User not found"}

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    for idx, existing_user in enumerate(users):
        if existing_user.id == user_id:
            del users[idx]
            return {"message": "User removed"}
    return {"error": "User not found"}

애플리케이션 실행

이제 애플리케이션을 실행할 시간입니다. 아래의 명령어를 사용하여 Uvicorn 서버를 시작합니다:

uvicorn main:app --reload

서버가 성공적으로 실행되면, 웹 브라우저에서 http://127.0.0.1:8000/docs로 이동하여 Swagger UI를 통해 API를 테스트할 수 있습니다.

테스트

Swagger UI를 사용하여 상세하게 API를 테스트하는 방법에 대해 설명합니다. 각각의 엔드포인트에서 POST, GET, PUT, DELETE 요청을 통해 사용자 정보를 추가, 조회, 수정 및 삭제할 수 있습니다.

사용자 생성

POST 요청을 통해 사용자를 생성할 수 있습니다. JSON 형식으로 아래와 같은 데이터를 입력합니다:

{
    "id": 1,
    "name": "홍길동",
    "age": 30
}

모든 사용자 조회

GET 요청을 통해 모든 사용자의 리스트를 볼 수 있습니다.

특정 사용자 조회

GET 요청을 통해 특정 사용자의 ID로 조회할 수 있습니다. 예: /users/1

사용자 수정

PUT 요청을 통해 특정 사용자의 정보를 수정할 수 있습니다. JSON 형식으로 업데이트된 데이터를 입력합니다:

{
    "id": 1,
    "name": "홍길동 업데이트",
    "age": 31
}

사용자 삭제

DELETE 요청을 통해 특정 사용자를 삭제할 수 있습니다. 예: /users/1

결론

이번 포스트에서는 FastAPI를 사용하여 간단한 CRUD 애플리케이션을 개발하는 방법을 배웠습니다. FastAPI는 매우 직관적이고 사용하기 쉬우며, 이를 통해 효율적인 웹 API를 쉽게 구축할 수 있습니다. 또한, Swagger UI와 같은 도구를 통해 API 문서를 자동으로 생성하고 테스트할 수 있는 점은 개발 과정을 더욱 효율적으로 만들어 줍니다.

이제 여러분도 FastAPI를 사용하여 더 복잡한 애플리케이션을 만들어 볼 수 있는 기초를 쌓았길 바랍니다. FastAPI가 제공하는 다양한 기능들을 탐험해 보세요!

FastAPI 서버개발, 앱에서 DB에 접속하기 위한 준비

FastAPI는 최신 Python 웹 프레임워크 중 하나로, 고성능과 빠른 개발 속도를 자랑하며 비동기 기능을 지원하여 효율적인 서버 애플리케이션을 만들 수 있게 해줍니다. 본 강좌에서는 FastAPI를 사용하여 데이터베이스와 연동하는 방법에 대해 자세히 설명하겠습니다. 올바른 방법으로 DB와 연결하기 위해 필요한 준비 단계와 예제 코드를 통해 실습할 수 있도록 준비했습니다.

1. FastAPI 소개

FastAPI는 Python 3.6+ 버전에서 사용할 수 있으며, 이 프레임워크는 ASGI(Asynchronous Server Gateway Interface)를 기반으로 하고 있습니다. FastAPI의 주요 장점 중 일부는 다음과 같습니다.

  • 빠른 성능: FastAPI는 Starlette와 Pydantic을 기반으로 하여 높은 성능을 자랑합니다.
  • 비동기 지원: 비동기 프로그래밍을 지원하여 I/O 작업을 효율적으로 처리합니다.
  • 자동 문서화: OpenAPI 스펙을 사용하여 자동으로 API 문서를 생성합니다.

2. 데이터베이스와의 연결 준비

FastAPI를 통해 데이터베이스와 연결하기 위해, 다음과 같은 준비 작업이 필요합니다.

2.1. 데이터베이스 선택

일반적으로 사용하는 데이터베이스로는 MySQL, PostgreSQL, SQLite 등이 있습니다. 이 강좌에서는 SQLite를 사용하여 데이터베이스를 설정하겠습니다. SQLite는 파일 기반의 경량 데이터베이스로, 개발 및 테스트 용도로 적합합니다.

2.2. 필요한 패키지 설치

FastAPI와 데이터베이스 연결을 위해 몇 가지 패키지를 설치해야 합니다. 아래의 명령어를 사용하여 필요한 패키지를 설치합니다.

pip install fastapi uvicorn sqlalchemy databases

2.3. 데이터베이스 모델 정의

SQLAlchemy를 사용하여 데이터베이스 모델을 정의합니다. 모델은 데이터베이스의 테이블과 매핑되는 클래스입니다. 아래는 사용자 정보를 저장하기 위한 User 모델을 정의한 예제입니다.


from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(bind=engine)
    

3. FastAPI 기본 서버 설정

FastAPI 애플리케이션을 구성하고, 데이터베이스와의 연결 설정을 진행합니다. FastAPI 인스턴스를 생성하고, 필요한 라우터를 추가하는 방법에 대해 살펴보겠습니다.


from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = sessionmaker(bind=engine)()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
def create_user(user: User, db: Session = Depends(get_db)):
    db.add(user)
    db.commit()
    db.refresh(user)
    return user
    

4. FastAPI와의 데이터베이스 상호작용

이제 FastAPI와 데이터베이스 상호작용을 위한 API 엔드포인트를 정의할 시간입니다. 사용자 생성, 조회, 조회 및 삭제와 같은 기본 CRUD 기능을 다음과 같이 구현할 수 있습니다.


@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

@app.get("/users/")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
        return {"message": "User deleted"}
    return {"message": "User not found"}
    

5. 데이터베이스 연결 테스트

모든 구성이 완료되었으면 이제 FastAPI 서버를 실행하고 데이터베이스 연결이 제대로 작동하는지 테스트해보는 것이 중요합니다. 아래 명령어로 FastAPI 서버를 실행할 수 있습니다.

uvicorn main:app --reload

서버가 실행된 후, Swagger 문서를 통해 API 엔드포인트를 테스트해 볼 수 있습니다. 사용자를 생성하고, 조회하고, 삭제하는 테스트를 통해 데이터베이스 연결이 정상임을 확인할 수 있습니다.

6. FastAPI의 비동기 데이터베이스 연동

FastAPI는 비동기 기능을 제공하므로, 비동기 데이터베이스 프레임워크를 사용하여 효율적으로 데이터베이스와 상호작용할 수 있습니다. 아래는 async/await 패턴을 사용하여 비동기적으로 데이터베이스 쿼리를 수행하는 방법에 대한 간단한 예제입니다.


from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
async_engine = create_async_engine(DATABASE_URL, echo=True)
Base = declarative_base()

async def get_db() -> AsyncSession:
    async_session = sessionmaker(async_engine, expire_on_commit=False, class_=AsyncSession)
    async with async_session() as session:
        yield session

@app.post("/users/", response_model=User)
async def create_user(user: User, db: AsyncSession = Depends(get_db)):
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user
    

비동기 데이터베이스 연동을 통해 성능을 더욱 최적화할 수 있습니다. 이러한 방법을 통해 더 많은 클라이언트 요청을 처리하면서 서버 자원을 효율적으로 사용할 수 있습니다.

7. 결론

FastAPI와 데이터베이스를 함께 사용하는 방법을 살펴보았습니다. FastAPI 서버를 설정하고, 데이터베이스와의 연결을 위한 준비 과정을 통해 기본적인 CRUD 기능을 구현했습니다. 이 과정에서 FastAPI의 비동기 처리 기능을 활용했다면 더욱 효과적인 웹 애플리케이션을 구축할 수 있습니다.

이제 여러분들은 FastAPI와 데이터베이스를 연동한 시스템을 구축하는 기초적인 방법을 배우셨습니다. 이 지식을 바탕으로 더 복잡한 애플리케이션을 설계하고 개발해보세요.

작성자: 조광형

날짜: [작성일자]

FastAPI 서버개발, FastAPI 애플리케이션 구조화, 모델 구현

FastAPI는 현대적인 웹 애플리케이션 개발을 쉽게 하고 효율적으로 할 수 있도록 돕는 python 웹 프레임워크입니다.
이 글에서는 FastAPI를 활용하여 서버를 개발하는 방법, 애플리케이션을 구조화하는 방법, 그리고 모델을 구현하는 방법에 대해 이야기하겠습니다.
이 글은 FastAPI에 대한 기본적인 이해가 있는 독자들을 대상으로 하며,
고급 개발 기법과 모범 사례를 포함하여 실제 애플리케이션 구현 과정 전반을 다룰 것입니다.

FastAPI란?

FastAPI는 Asynchronous Python 웹 프레임워크로, Python 3.6 이상을 지원하며,
고속의 웹 API를 구축하는 데 적합합니다. FastAPI는 다음과 같은 주요 특징을 가지고 있습니다:

  • 자동 문서화: OpenAPI 및 Swagger UI를 사용하여 API를 쉽게 문서화합니다.
  • 비동기 지원: async/await 구문을 통해 비동기 처리를 구현하여 성능을 극대화합니다.
  • 입력 검증: Pydantic을 사용한 요청 및 응답 데이터 검증이 가능합니다.
  • 사용이 용이: 간결한 코드 작성과 쉬운 학습 곡선으로 빠르게 시작할 수 있습니다.

FastAPI 설치

FastAPI를 사용하기 위해서는 우선 FastAPI 라이브러리와 ASGI 서버인 uvicorn을 설치해야 합니다.

bash
pip install fastapi uvicorn

FastAPI 애플리케이션 구조화

FastAPI 애플리케이션을 구조화하는 것은 애플리케이션을 유지보수하고 확장 가능하게 만드는 데 중요한 요소입니다.
일반적인 프로젝트 구조는 다음과 같이 구성됩니다:


my_fastapi_app/
├── app/
│   ├── main.py                   # 애플리케이션의 진입점
│   ├── api/                      # API 라우터
│   │   ├── __init__.py
│   │   └── endpoints/
│   │       ├── __init__.py
│   │       ├── items.py          # 아이템 관련 API
│   │       └── users.py          # 사용자 관련 API
│   ├── models/                   # Pydantic 모델 정의
│   │   ├── __init__.py
│   │   ├── item.py               # 아이템 모델
│   │   └── user.py               # 사용자 모델
│   └── database.py               # 데이터베이스 연결 및 설정
├── requirements.txt              # 필요한 패키지 목록
└── README.md                     # 프로젝트 설명

main.py

FastAPI 애플리케이션의 진입점인 main.py 파일을 작성합니다.
이 파일은 애플리케이션을 시작하고 API 라우터를 포함하는 역할을 합니다.

python
from fastapi import FastAPI
from app.api.endpoints import items, users

app = FastAPI()

# API 라우터 포함
app.include_router(items.router)
app.include_router(users.router)

@app.get("/")
def read_root():
    return {"Hello": "World"}

API 라우터 – items.py

이 파일에서는 아이템에 대한 API 엔드포인트를 정의합니다.

python
from fastapi import APIRouter
from app.models.item import Item

router = APIRouter()

@router.post("/items/")
async def create_item(item: Item):
    return item

@router.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id, "name": "sample item"}

모델 구현 – item.py

Pydantic 모델을 정의하여 데이터 유효성을 검사합니다.

python
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

API 라우터 – users.py

사용자 관련 API 엔드포인트를 정의합니다.

python
from fastapi import APIRouter
from app.models.user import User

router = APIRouter()

@router.post("/users/")
async def create_user(user: User):
    return user

@router.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id, "name": "sample user"}

모델 구현 – user.py

사용자 관련 모델을 정의합니다.

python
from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str
    full_name: str = None

데이터베이스 설정

FastAPI에서 SQLAlchemy와 같은 ORM을 사용하여 데이터베이스와의 상호작용을 설정할 수 있습니다.
database.py 파일을 작성하여 데이터베이스 연결을 구성합니다.

python
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # SQLite 데이터베이스 사용
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

FastAPI 애플리케이션 실행

이제 FastAPI 애플리케이션을 실행해 봅시다. 아래와 같은 명령어로 uvicorn을 통해 서버를 시작합니다.

bash
uvicorn app.main:app --reload

서버가 실행되면 http://127.0.0.1:8000/docs에서 자동으로 생성된 Swagger UI를 통해 API를 테스트할 수 있습니다.

결론

FastAPI는 매우 강력하고 유연한 웹 프레임워크로, 데이터 검증 및 비동기 처리를 지원하여 현대적인 웹 API를 구축하는 데 적합합니다.
이 글에서는 FastAPI 서버 개발의 기초와 애플리케이션 구조화, 모델 구현 방법에 대해 알아보았습니다.
FastAPI의 추가 기능과 고급 주제에 대해서는 공식 문서를 참조하여 더욱 깊이 있는 학습을 이어가시기 바랍니다.

성공적인 FastAPI 개발을 기원합니다!

FastAPI 서버개발, FastAPI 테스트 클라이언트로 API 테스트하기

return item

@app.get(“/items/”, response_model=List[Item])
async def read_items():
return list(items.values())

@app.get(“/items/{item_name}”, response_model=Item)
async def read_item(item_name: str):
if item_name in items:
return items[item_name]
return {“error”: “Item not found”}

@app.put(“/items/{item_name}”, response_model=Item)
async def update_item(item_name: str, item: Item):
if item_name in items:
items[item_name] = item
return item
return {“error”: “Item not found”}

@app.delete(“/items/{item_name}”, response_model=Item)
async def delete_item(item_name: str):
if item_name in items:
return items.pop(item_name)
return {“error”: “Item not found”}

코드 설명

위의 코드는 Create, Read, Update, Delete 기능을 갖춘 엔드포인트를 정의합니다. POST 메서드는 새로운 아이템을 생성하며, GET 메서드로 모든 아이템을 조회하거나 특정 아이템을 조회할 수 있습니다. PUT 메서드는 기존 아이템을 수정할 때 사용되고, DELETE 메서드는 아이템을 삭제하는 역할을 합니다.

3. CRUD API 테스트하기

CRUD 기능을 테스트하는 테스트 케이스도 추가해보겠습니다.


def test_create_item():
    response = client.post("/items/", json={"name": "Item1", "price": 10.0})
    assert response.status_code == 200
    assert response.json() == {"name": "Item1", "description": None, "price": 10.0, "tax": None}

def test_read_items():
    response = client.get("/items/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

def test_update_item():
    response = client.put("/items/Item1", json={"name": "Item1", "price": 12.0})
    assert response.status_code == 200
    assert response.json() == {"name": "Item1", "description": None, "price": 12.0, "tax": None}

def test_delete_item():
    response = client.delete("/items/Item1")
    assert response.status_code == 200
    assert response.json() == {"name": "Item1", "description": None, "price": 12.0, "tax": None}

코드 설명

각 테스트 함수는 API의 CRUD 기능을 검증합니다. test_create_item 함수는 새로운 아이템을 생성하고, test_read_items 함수는 모든 아이템을 조회합니다. test_update_item은 특정 아이템을 수정하며, test_delete_item은 아이템을 삭제하는 기능을 테스트합니다.

FastAPI의 보안과 인증

FastAPI는 OAuth2와 JWT 토큰 기반 인증을 지원합니다. 보안이 필요한 API 엔드포인트에 대한 인증 방법을 설명하겠습니다.

1. OAuth2PasswordBearer 사용하기


from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 사용자 인증 로직 (여기서는 생략)
    return {"access_token": form_data.username, "token_type": "bearer"}

2. 보호된 엔드포인트 만들기

로그인 후 액세스 토큰을 통해 보호된 엔드포인트를 만들어보겠습니다.


@app.get("/secure-data")
async def read_secure_data(token: str = Depends(oauth2_scheme)):
    return {"data": "This is protected data"}

코드 설명

위 코드는 /token 경로에서 로그인 시 액세스 토큰을 생성하고, 생성된 토큰을 통해 /secure-data 경로에 접근하도록 설정합니다. Depends()를 사용하여 인증이 필요한 기능을 구현했습니다.

3. 보안 테스트

보호된 엔드포인트를 테스트하기 위해 인증 후 토큰을 사용할 수 있습니다.


def test_secure_data():
    response = client.post("/token", data={"username": "testuser", "password": "testpassword"})
    token = response.json().get("access_token")

    response = client.get("/secure-data", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200
    assert response.json() == {"data": "This is protected data"}

결론

이 포스트에서는 FastAPI를 사용하여 간단한 API 서버를 구축하고, FastAPI의 테스트 클라이언트를 활용하여 다양한 API 엔드포인트를 테스트하는 방법에 대해 배웠습니다. FastAPI를 통해 고성능의 API를 쉽게 생성할 수 있으며, TestClient를 이용해 안정적이고 신뢰성 있는 테스트를 진행할 수 있습니다. 다양한 기능과 확장성을 바탕으로 FastAPI를 활용한 프로젝트를 진행해보시길 바랍니다.

FastAPI 서버개발, FastAPI의 비동기 요청 처리 이해하기

FastAPI는 빠르고 현대적인 웹 프레임워크로, Python 3.7 이상에서 사용할 수 있습니다. FastAPI는 ASGI를 기반으로 하여 비동기 프로그래밍을 쉽게 활용할 수 있도록 도와줍니다. 이 글에서는 FastAPI를 사용하여 비동기 요청 처리를 구현하는 방법을 상세히 설명하고, 예제 코드를 통해 이해를 돕겠습니다.

1. FastAPI와 비동기 프로그래밍의 이해

비동기 프로그래밍은 주로 I/O 작업이 많은 작업에서 성능을 크게 개선할 수 있는 기술입니다. 동기식 프로그래밍에서는 요청이 완료될 때까지 해당 요청을 처리하는 스레드가 대기하게 됩니다. 하지만 비동기 프로그래밍에서는 자원이 낭비되지 않도록 다른 작업을 동시에 수행할 수 있습니다.

1.1 FastAPI의 비동기 지원

FastAPI는 비동기 함수(`async def`)를 사용하여 I/O 바운드 작업을 처리할 수 있게 지원합니다. FastAPI의 경로 작업(route handler)에서 비동기 함수를 사용하면, 비동기적으로 요청을 처리할 수 있어 다른 요청이 대기하지 않고 곧바로 처리됩니다.

1.2 비동기와 동기 함수 비교

FastAPI에서 비동기 경로 작업을 구현하는 방법은 동기 함수와 유사하지만, 반드시 `async` 키워드를 사용하여 정의해야 합니다. 아래는 비동기와 동기 함수의 예제 코드입니다.

from fastapi import FastAPI
import time

app = FastAPI()

# 동기 경로 작업
@app.get("/sync")
def read_sync():
    time.sleep(2)  # 2초 대기
    return {"message": "동기 요청 처리 완료"}

# 비동기 경로 작업
@app.get("/async")
async def read_async():
    await asyncio.sleep(2)  # 2초 대기
    return {"message": "비동기 요청 처리 완료"}

위의 코드는 `/sync` 경로를 통해 동기적으로 요청을 처리하고, `/async` 경로를 통해 비동기적으로 요청을 처리합니다. 요청이 들어온 순서에 따라 동기 요청은 처리하는 동안 다른 요청이 대기하게 되지만, 비동기 요청은 대기하는 동안 다른 요청을 처리할 수 있습니다.

2. FastAPI에서 비동기 I/O 작업 처리하기

FastAPI에서 비동기 I/O 작업은 데이터베이스의 CRUD 작업, 웹 API 호출, 파일 업로드 등 다양한 경우에 활용됩니다. 각 I/O 작업이 비동기적으로 수행되면, 애플리케이션의 성능을 높일 수 있습니다.

2.1 비동기 데이터베이스 작업

다양한 비동기 데이터베이스 드라이버가 FastAPI와 함께 사용될 수 있습니다. 예를 들어, databases라는 비동기 ORM 라이브러리와 함께 사용하여 비동기적으로 데이터베이스와 상호작용할 수 있습니다.

import asyncio
from fastapi import FastAPI
import databases

DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/items/")
async def read_items():
    query = "SELECT * FROM items"
    return await database.fetch_all(query)

위의 코드는 `databases` 라이브러리를 사용하여 SQLite 데이터베이스에 비동기적으로 연결하고, 아이템을 읽어오는 예시입니다. FastAPI의 생명주기 이벤트를 통해 서버가 시작할 때와 종료할 때 데이터베이스 연결을 관리할 수 있습니다.

2.2 비동기 HTTP 요청 처리

FastAPI에서는 httpx 같은 라이브러리를 사용하여 비동기 HTTP 요청도 쉽게 처리할 수 있습니다. 비동기 HTTP 요청은 외부 API와의 통신이 필요할 때 매우 유용합니다.

import httpx

@app.get("/external/")
async def read_external():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
    return response.json()

위의 코드는 외부 API에서 데이터를 가져와서 반환하는 비동기 경로 작업입니다. 비동기 클라이언트를 사용함으로써, 데이터 요청 도중 서버가 다른 요청을 처리할 수 있도록 합니다.

3. 비동기 작업에서의 오류 처리

비동기 작업을 구현할 때, 오류 처리 또한 중요합니다. 비동기 작업이 실패할 경우, 적절한 오류 메시지를 반환해야 하며, 이때 HTTPException을 사용할 수 있습니다.

from fastapi import HTTPException

@app.get("/data/")
async def read_data(param: str):
    if param not in ["valid1", "valid2"]:
        raise HTTPException(status_code=404, detail="데이터를 찾을 수 없습니다.")
    return {"message": f"{param}에 대한 데이터"}

위의 코드는 파라미터가 유효하지 않을 때 HTTP 404 오류를 반환하는 예제입니다. FastAPI의 예외 처리 메커니즘은 비동기 방식에서도 동일하게 작동합니다.

4. FastAPI의 비동기 요청 처리 성능 비교

비동기 요청 처리와 동기 요청 처리의 성능 차이를 비교할 수 있는 방법도 중요합니다. 아래는 FastAPI의 비동기 및 동기 경로 작업에 대한 성능 테스트 예제입니다.

import time
import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/sync-test/")
def sync_test():
    time.sleep(1)  # 1초 대기
    return {"message": "비동기 요청이 아닙니다."}

@app.get("/async-test/")
async def async_test():
    await asyncio.sleep(1)  # 1초 대기
    return {"message": "비동기 요청입니다."}

이제 이 두 경로를 동시에 호출하여 성능을 비교할 수 있습니다. 테스트 도구(예: Apache Bench 또는 wrk)를 사용하여 두 경로에서의 응답 시간을 측정해보세요.

5. 결론

FastAPI는 비동기 요청 처리를 지원하여 높은 성능을 기록할 수 있는 프레임워크입니다. I/O 작업을 비동기적으로 처리함으로써, 서버의 자원을 효율적으로 활용할 수 있습니다. 이번 글에서는 FastAPI의 비동기 요청 처리 방식에 대해 설명하고 여러 예제를 통해 이해를 도왔습니다.

FastAPI를 사용하면 비동기적 아키텍처를 쉽게 구현할 수 있으며, 이에 따라 생산성과 성능이 향상됩니다. 비동기 함수를 적절히 활용하여 비즈니스 로직을 구현하고, 다양한 응용 프로그램을 개발해보세요!