FastAPI 서버개발, SQLAlchemy의 DB 모델 정의

FastAPI는 Python으로 웹 애플리케이션을 개발하기 위해 설계된 모던 웹 프레임워크로, 매우 빠르며 비동기 프로그래밍을 지원합니다. 이번 포스트에서는 FastAPI의 백엔드 개발 환경에서 SQLAlchemy를 사용하여 데이터베이스 모델을 정의하는 방법에 대해 자세히 설명하겠습니다. SQLAlchemy는 Python의 ORM(Object Relational Mapping) 라이브러리로, 데이터베이스와의 상호작용을 더 간편하게 만들어줍니다.

1. FastAPI와 SQLAlchemy 소개

FastAPI는 다음과 같은 몇 가지 주요 이점을 가지고 있습니다:

  • 비동기 지원: Python의 asyncawait 구문을 사용하여 비동기 프로그래밍을 간편하게 구현할 수 있습니다.
  • 자동화된 문서화: FastAPI는 OpenAPI 스펙을 기반으로 클라이언트를 위한 API 문서를 자동으로 생성합니다.
  • 고성능: Starlette 위에 구축되어 있어 높은 성능을 자랑합니다.

SQLAlchemy는 관계형 데이터베이스와 상호작용하기 위한 ORM으로 다음과 같은 이점이 있습니다:

  • 데이터베이스 독립성: 여러 종류의 데이터베이스와 호환됩니다.
  • 복잡한 쿼리를 쉽게 작성: SQLAlchemy의 쿼리 빌더를 사용하면 복잡한 SQL 쿼리를 쉽게 작성할 수 있습니다.
  • 모델 정의: Python 클래스 정의를 통해 데이터베이스 모델을 직관적으로 생성할 수 있습니다.

2. 개발 환경 설정

FastAPI와 SQLAlchemy를 사용하기 위해서는 우선 Python이 설치되어 있어야 합니다. 각 패키지를 설치하는 과정은 다음과 같습니다.

pip install fastapi[all] sqlalchemy

참고: 위의 명령어에서 fastapi[all]는 FastAPI와 함께 uvicorn과 같은 모든 의존성을 설치합니다.

3. SQLAlchemy DB 모델 정의하기

SQLAlchemy를 사용하여 데이터베이스 모델을 정의하려면 몇 가지 단계를 거쳐야 합니다:

  1. SQLAlchemy 엔진 생성
  2. 세션 클래스 정의
  3. 데이터베이스 모델 정의
  4. 테이블 생성

3.1 SQLAlchemy 엔진 생성

데이터베이스 연결을 위해 SQLAlchemy 엔진을 생성해야 합니다. 이는 데이터베이스 URL을 통해 이루어집니다. 예를 들어, SQLite 데이터베이스를 사용할 경우 아래와 같이 설정할 수 있습니다.

from sqlalchemy import create_engine

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

3.2 세션 클래스 정의

SQLAlchemy의 세션은 데이터베이스에 대한 트랜잭션을 관리합니다. 아래 코드를 통해 세션 클래스를 정의할 수 있습니다.

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

Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

3.3 데이터베이스 모델 정의

이제 SQLAlchemy의 Base 클래스를 상속받아 원하는 데이터베이스 모델을 정의할 수 있습니다. 예를 들어, 사용자 정보를 저장하기 위한 User 모델은 다음과 같이 정의할 수 있습니다.

from sqlalchemy import Column, Integer, String, Boolean

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)
    is_active = Column(Boolean, default=True)

3.4 테이블 생성하기

모델 정의가 끝났다면 실제 데이터베이스에 테이블을 생성해야 합니다. 다음과 같은 코드를 사용하여 테이블을 생성할 수 있습니다:

Base.metadata.create_all(bind=engine)

4. FastAPI와 SQLAlchemy 통합하기

이제 FastAPI를 사용해 만든 API와 SQLAlchemy 데이터베이스 모델을 통합할 차례입니다. 이를 위해 FastAPI 경로에서 CRUD(Create, Read, Update, Delete) 작업을 처리하는 엔드포인트를 작성합니다.

4.1 FastAPI 애플리케이션 생성

from fastapi import FastAPI

app = FastAPI()

4.2 CRUD API 엔드포인트 작성

다음으로, CRUD 작업을 위한 엔드포인트를 정의합니다.

4.2.1 Create: 사용자 생성

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

# 의존성 주입을 위한 함수
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

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

4.2.2 Read: 사용자 조회

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

4.2.3 Update: 사용자 업데이트

@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, user: User, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_user.username = user.username
    db_user.email = user.email
    db.commit()
    db.refresh(db_user)
    return db_user

4.2.4 Delete: 사용자 삭제

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"detail": "User deleted"}

5. 결론

이 글에서 우리는 FastAPI와 SQLAlchemy를 사용하여 데이터베이스 모델을 정의하고, 이를 기반으로 CRUD API를 구축하는 방법에 대해 알아보았습니다. FastAPI의 비동기 지원과 SQLAlchemy의 ORM 기능을 결합하면 Python으로 웹 애플리케이션을 효율적으로 개발할 수 있습니다. 이 과정을 통해 더 나은 백엔드 시스템을 구축하고, 확장 가능한 API를 만들 수 있습니다.

추가적으로 FastAPI의 기능을 이용하여 데이터 유효성 검증, 쿼리 파라미터 처리, JWT 인증, CORS 설정 등 다양한 기능을 적용할 수 있습니다. 이러한 기능들은 실제 실무에서 매우 유용하게 사용될 수 있습니다.

FastAPI와 SQLAlchemy의 기본 사용법을 익힌 후에는 복잡한 비즈니스 로직을 추가하여 더욱 정교한 애플리케이션을 만들어 가는 과정을 통해 더 큰 발전을 이룰 수 있을 것입니다.

FastAPI 서버개발, RESTful API 설계 원칙과 FastAPI의 활용

RESTful API 설계 원칙과 FastAPI의 활용

1. FastAPI란?

FastAPI는 Python으로 작성된 현대적인 웹 프레임워크로, 빠르고, 사용하기 쉬우며, RESTful API 서버를 구축하기 위해 최적화되었습니다. 이 프레임워크는 Pydantic과 Starlette를 기반으로 하고 있으며, 비동기 프로그래밍을 지원하여 높은 성능을 자랑합니다. FastAPI의 주요 강점 중 하나는 데이터 유효성 검사와 스키마 생성을 자동으로 수행할 수 있다는 점입니다. 이러한 기능은 개발자 경험을 향상시키고 개발 시간을 단축시킵니다.

2. RESTful API란?

REST(Representational State Transfer)는 웹에서 자원을 정의하고 자원에 대한 상호작용을 정의하는 아키텍처 스타일입니다. RESTful API는 이러한 원칙을 따르는 웹 서비스로, 자원에 HTTP 메서드를 사용하여 접근합니다. REST 아키텍처는 다음과 같은 기본 원칙을 가지고 있습니다:

  • 자원 지향: API는 고유한 URI로 자원을 식별합니다.
  • 무상태: 클라이언트와 서버 간의 상호작용은 독립적이며, 서버는 클라이언트의 상태를 유지하지 않습니다.
  • 표현: 자원은 다양한 표현 형식(JSON, XML 등)으로 제공될 수 있습니다.
  • HTTP 메서드 활용: GET, POST, PUT, DELETE 등의 메서드를 사용하여 자원에 대한 CRUD(Create, Read, Update, Delete) 작업을 수행합니다.

3. FastAPI 설치하기

FastAPI를 사용하기 위해서는 Python과 pip가 필요합니다. 아래는 FastAPI와 함께 Uvicorn 서버를 설치하는 방법입니다:

pip install fastapi uvicorn

Uvicorn은 FastAPI 애플리케이션을 실행하는 ASGI 서버입니다.

4. FastAPI로 간단한 RESTful API 구축하기

4.1. 기본 예제

FastAPI로 간단한 RESTful API를 만들어보겠습니다. 다음은 사용자 정보를 처리하는 API입니다:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

users = []

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

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

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

@app.delete("/users/{user_id}")
def delete_user(user_id: int):
    for user in users:
        if user.id == user_id:
            users.remove(user)
            return {"message": "User deleted successfully"}
    return {"error": "User not found"}

위의 코드는 FastAPI를 사용하여 사용자 정보를 추가, 조회, 삭제할 수 있는 간단한 API를 생성합니다.

5. RESTful API 설계 원칙

5.1. URI 설계

RESTful API에서는 자원의 URI를 명확하게 설계하는 것이 중요합니다. URI는 자원을 표현하고, 클라이언트가 REST API를 통해 자원에 접근할 수 있게 합니다. 다음은 좋은 URI 설계의 예입니다:

  • GET /users: 모든 사용자 목록 조회
  • POST /users: 새 사용자 추가
  • GET /users/{id}: 특정 사용자 조회
  • DELETE /users/{id}: 특정 사용자 삭제

URI를 설계할 때는 명사 형태를 사용하는 것이 좋습니다. 동사는 HTTP 메서드를 통해 전달됩니다.

5.2. HTTP 메서드 사용

RESTful API에서는 자원에 대한 행위를 표현하기 위해 HTTP 메서드를 사용합니다. 각 메서드의 용도는 다음과 같습니다:

  • GET: 서버로부터 자원 조회
  • POST: 서버에 새로운 자원 생성
  • PUT: 서버의 자원 수정
  • DELETE: 서버의 자원 삭제

5.3. 상태 코드 사용

RESTful API에서는 적절한 HTTP 상태 코드를 사용하여 클라이언트에게 요청의 결과를 전달해야 합니다. 주요 상태 코드는 다음과 같습니다:

  • 200 OK: 요청 성공
  • 201 Created: 자원 생성 성공
  • 204 No Content: 자원 삭제 성공
  • 404 Not Found: 요청한 자원 없음
  • 400 Bad Request: 잘못된 요청
  • 500 Internal Server Error: 서버 오류

6. FastAPI의 고급 기능

6.1. 데이터 검증 및 스키마

FastAPI는 Pydantic을 사용하여 데이터 검증과 스키마 관리를 제공합니다. 요청의 본문에서 전달되는 데이터에 대해 자동으로 검증을 수행하고, 잘못된 데이터가 들어오면 에러를 반환합니다. 예를 들어, 다음과 같이 데이터 검증을 추가할 수 있습니다:

from pydantic import EmailStr

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

여기서 email 필드는 이메일 형식의 문자열만 허용됩니다.

6.2. 의존성 주입

FastAPI는 의존성 주입을 지원하여 코드의 재사용성을 높이고, 테스트를 쉽게 할 수 있습니다. 의존성은 FastAPI의 @Depends를 사용하여 정의할 수 있습니다. 예를 들어:

from fastapi import Depends

def get_query_param(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

@app.get("/items/")
def get_items(queries: dict = Depends(get_query_param)):
    return {"queries": queries}

여기서 get_query_param 함수는 쿼리 파라미터를 처리하는 의존성으로 사용됩니다.

7. FastAPI와 Swagger UI

FastAPI는 Swagger UI를 자동으로 생성하여 API 문서를 쉽게 볼 수 있도록 제공합니다. 서버를 실행한 후, 브라우저에서 http://localhost:8000/docs로 이동하면 API 문서를 확인할 수 있습니다. 이 문서에서는 각 API 엔드포인트의 요청 및 응답 형식, 설명 등을 확인할 수 있습니다.

8. FastAPI 프로젝트 구조

FastAPI 프로젝트의 구조는 의견에 따라 다를 수 있지만, 일반적으로 다음과 같이 구성할 수 있습니다:

my_fastapi_app/
    ├── app/
    │   ├── __init__.py
    │   ├── main.py
    │   ├── models.py
    │   ├── routers/
    │   │   ├── __init__.py
    │   │   ├── user.py
    │   │   └── item.py
    │   └── services/
    │       ├── __init__.py
    │       └── user_service.py
    └── requirements.txt

여기에서 main.py는 FastAPI 애플리케이션을 실행하는 파일이며, models.py는 데이터 모델을 정의하고, routers/에는 각각의 API 엔드포인트를 정의하는 라우터를 구현합니다. services/ 폴더에는 비즈니스 로직을 분리하여 관리할 수 있습니다.

9. FastAPI로 안전한 API 만들기

9.1. 인증 및 인가

FastAPI는 OAuth2와 JWT(Json Web Token)를 사용하여 인증 및 인가 기능을 쉽게 구현할 수 있습니다. 예를 들어, JWT 토큰을 생성하고 검증하는 간단한 방법은 다음과 같습니다:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    # 사용자 인증 로직 구현 (mock)
    if form.username == "user" and form.password == "password":
        return {"access_token": "your_token", "token_type": "bearer"}
    raise HTTPException(status_code=400, detail="Invalid credentials")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    # 토큰 검증 로직 구현

10. FastAPI에서 테스트하기

FastAPI는 테스트를 매우 쉽게 만들어줍니다. Python의 비동기 테스트 기능을 사용하여 API를 테스트할 수 있습니다. 아래는 테스트의 간단한 예입니다:

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_user():
    response = client.post("/users/", json={"id": 1, "name": "John", "age": 30})
    assert response.status_code == 200
    assert response.json() == {"id": 1, "name": "John", "age": 30}

위의 예제에서는 FastAPI의 TestClient를 사용하여 API 호출을 테스트합니다.

11. FastAPI 배포하기

FastAPI 애플리케이션을 배포하기 위해 Uvicorn을 사용하여 서버를 실행할 수 있습니다. 기본적인 실행 명령어는 다음과 같습니다:

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

여기서 --reload 옵션은 개발 모드에서 소스 코드를 수정할 때 자동으로 서버를 다시 시작합니다. 프로덕션 환경에서는 `–host`, `–port`, `–reload` 옵션을 적절히 설정해야 합니다.

12. 마치며

FastAPI는 손쉬운 RESTful API 개발을 위해 필요한 모든 기능을 제공하는 강력한 프레임워크입니다. 이 글에서 다룬 내용을 통해 FastAPI를 사용하여 RESTful API를 설계하고 구현하는 방법을 익혔기를 바랍니다. FastAPI의 다양한 기능을 활용하여 더욱 효율적이고 안전한 API를 개발해보세요.

FastAPI 서버개발, 배포 준비

FastAPI는 Python으로 작성된 현대적인 웹 프레임워크로, 빠른 성능과 직관적인 사용성을 제공하여 RESTful API 서버를 구축하는 데 매우 유용합니다. 이 글에서는 FastAPI 서버를 개발하는 방법뿐만 아니라, 배포 준비 과정에 대해서도 상세히 설명하겠습니다.

FastAPI의 주요 특징

  • 빠른 성능: FastAPI는 Starlette와 Pydantic을 기반으로 하여 매우 빠른 성능을 제공합니다.
  • 자동 문서화: Swagger UI와 ReDoc을 통해 API 문서를 자동으로 생성합니다.
  • 비동기 처리 지원: Python의 비동기 프로그래밍을 쉽게 사용할 수 있습니다.
  • 타입 힌트: Python의 타입 힌트를 사용하여 코드의 가독성과 안정성을 높일 수 있습니다.

FastAPI 서버 개발 시작하기

FastAPI를 사용하여 간단한 웹 서버를 구축해보겠습니다. 이 예제에서는 간단한 사용자 등록 API를 만들어 보겠습니다.

1. FastAPI 설치

FastAPI를 설치하려면 pip를 사용합니다. 또한, Uvicorn을 ASGI 서버로 사용해 실행할 수 있습니다.

pip install fastapi uvicorn

2. FastAPI 앱 만들기

FastAPI 앱을 정의하고, 사용자 등록 API를 구현해보겠습니다.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    email: str

@app.post("/register/")
async def register_user(user: User):
    return {"message": f"User {user.username} registered successfully!", "email": user.email}

3. 서버 실행하기

Uvicorn을 사용하여 서버를 실행합니다. 아래 명령어를 터미널에 입력하면 서버가 시작됩니다.

uvicorn main:app --reload

위 명령으로 `/register/` 엔드포인트가 활성화되며, 이를 통해 사용자 정보를 POST 방식으로 전송할 수 있습니다.

FastAPI 앱의 구조

FastAPI 앱은 보통 다음과 같은 구조로 구성됩니다:

my_fastapi_app/
|-- main.py
|-- models/
|   |-- __init__.py
|   |-- user.py
|-- routers/
|   |-- __init__.py
|   |-- user.py
|-- utils/
|   |-- __init__.py
|   |-- helpers.py

모듈화된 FastAPI 앱 개발

서버가 커질수록 모듈화된 구조가 중요합니다. 각 노력별로 파일을 나누어 관리하면 유지보수와 코드 가독성을 높일 수 있습니다.

1. 모델 정의하기

모델을 정의하는 `user.py` 파일은 아래와 같습니다:

from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str

2. 라우터 정의하기

라우터를 정의하는 `routers/user.py` 파일:

from fastapi import APIRouter
from models.user import User

router = APIRouter()

@router.post("/register/")
async def register_user(user: User):
    return {"message": f"User {user.username} registered successfully!", "email": user.email}

3. 메인 앱에 라우터 추가하기

메인 앱 파일인 `main.py`에서 라우터를 포함하는 방법:

from fastapi import FastAPI
from routers import user

app = FastAPI()
app.include_router(user.router)

환경 설정

FastAPI 애플리케이션의 환경 설정을 위해 `dotenv`를 사용하여 환경 변수를 관리하는 것이 좋습니다. 아래와 같이 설정할 수 있습니다.

1. python-dotenv 설치

pip install python-dotenv

2. .env 파일 생성

DATABASE_URL=mysql://user:password@localhost:3306/mydatabase

3. 환경 변수 사용하기

import os
from dotenv import load_dotenv

load_dotenv()
database_url = os.getenv("DATABASE_URL")

FastAPI 배포 준비

FastAPI 서버를 배포하기 위한 방법을 설명합니다. 아래 단계로 진행할 수 있습니다.

1. 배포 환경 선택

FastAPI 애플리케이션은 여러 클라우드 서비스 제공업체에서 배포할 수 있습니다. 예를 들어, AWS, GCP, 또는 DigitalOcean과 같은 서비스에서 VM을 생성하여 배포할 수 있습니다.

2. Gunicorn 설치

FastAPI 애플리케이션을 프로덕션 환경에서 실행하기 위해 Gunicorn을 설치할 수 있습니다.

pip install gunicorn

3. 배포 스크립트 작성

배포 스크립트를 작성하여 서버를 배포할 수 있습니다. 다음의 예시와 같이 `start.sh` 파일을 생성하세요:

#!/bin/bash
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

이 스크립트로 서버를 실행할 수 있습니다.

4. Reverse Proxy 설정 (선택 사항)

NGINX와 같은 Reverse Proxy 서버를 설정하여 FastAPI 서버의 앞단에 두면 보안과 성능을 향상시킬 수 있습니다.

테스트 및 모니터링

코드 배포 후에는 API의 기능이 올바르게 동작하는지 확인하고, 모니터링 도구를 사용하여 성능을 관찰해야 합니다.

1. 테스트 프레임워크 설정

FastAPI는 Pytest와 함께 사용할 수 있습니다. 아래는 간단한 테스트 예제입니다:

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_register_user():
    response = client.post("/register/", json={"username": "testuser", "email": "test@example.com"})
    assert response.status_code == 200
    assert response.json() == {"message": "User testuser registered successfully!", "email": "test@example.com"}

2. 모니터링 도구

서버의 상태와 성능을 모니터링하기 위해 Prometheus와 Grafana를 사용할 수 있습니다.

결론

이 글에서는 FastAPI를 사용한 서버 개발 과정과 배포 준비 방법에 대해 설명했습니다. FastAPI의 강력한 기능과 유연한 구조 덕분에 현업에서 RESTful API를 효과적으로 구축할 수 있습니다. 배포 후에는 성능 모니터링을 통해 안정적인 서비스를 제공하십시오.

참고 자료

FastAPI 서버개발, DB 조작, Done 리소스

현대의 웹 애플리케이션 개발에서는 효율적이고, 반응성이 뛰어난 API 서버가 필수적입니다. FastAPI는 Python으로 작성된 빠르고 현대적인 웹 프레임워크로, 빠른 개발과 높은 성능을 자랑합니다. 이 글에서는 FastAPI를 사용하여 백엔드 서버를 구축하고, 데이터베이스를 조작하며, Done 리소스를 구현하는 방법에 대해 자세히 설명하겠습니다.

1. FastAPI란?

FastAPI는 Pydantic과 Starlette을 기반으로 한 고성능 웹 프레임워크로, Python에서 API를 구축하기 위한 최적의 선택입니다. FastAPI의 주요 특징은 다음과 같습니다:

  • 자동 문서화: OpenAPI 및 Swagger UI를 사용하여 API 문서화를 자동으로 제공합니다.
  • 타입 검사: Python 타입 힌트를 사용하여 코드의 가독성을 높이고 런타임 오류를 줄입니다.
  • 비동기 지원: 비동기 프로그래밍을 지원하여 높은 성능을 발휘합니다.
  • 유연성: RESTful API, GraphQL 등 다양한 API 디자인을 지원합니다.

2. FastAPI 설치하기

FastAPI를 사용하기 위해 먼저 FastAPI와 서버를 구동할 수 있는 Uvicorn을 설치합니다. 아래 명령어를 사용하여 설치할 수 있습니다:

pip install fastapi uvicorn

3. FastAPI 기본 구조

FastAPI 서버의 기본 구조는 다음과 같습니다. 아래 코드는 간단한 “Hello, World!” API를 구현한 예시입니다.

from fastapi import FastAPI

app = FastAPI()

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

위의 코드를 main.py라는 파일에 저장한 후, 다음 명령어로 서버를 실행할 수 있습니다:

uvicorn main:app --reload

서버가 실행되면 http://127.0.0.1:8000/로 접속하여 결과를 확인할 수 있습니다.

4. 데이터베이스 연결하기

이제 FastAPI와 데이터베이스를 연결해보겠습니다. 이번 예제에서는 SQLite 데이터베이스를 사용하여 Todo 리스트를 관리할 것입니다. 먼저, SQLAlchemydatabases 패키지를 설치합니다.

pip install sqlalchemy databases

4.1 데이터베이스 모델 정의하기

데이터베이스 모델을 정의하기 위해 SQLAlchemy의 ORM을 사용합니다. 아래는 Todo 모델을 정의하는 예시입니다.

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

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class Todo(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    completed = Column(Boolean, default=False)

4.2 데이터베이스 초기화

데이터베이스를 초기화하기 위해 아래와 같은 코드를 추가합니다.

def init_db():
    Base.metadata.create_all(bind=engine)

init_db()

5. CRUD 엔드포인트 구현하기

이제 Todo 리스트를 관리하기 위한 CRUD(Create, Read, Update, Delete) 엔드포인트를 구현하겠습니다.

5.1 Todo 생성하기

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

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/todos/", response_model=Todo)
async def create_todo(todo: Todo, db: Session = Depends(get_db)):
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return todo

5.2 Todo 목록 조회하기

@app.get("/todos/", response_model=List[Todo])
async def read_todos(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    todos = db.query(Todo).offset(skip).limit(limit).all()
    return todos

5.3 Todo 수정하기

@app.put("/todos/{todo_id}", response_model=Todo)
async def update_todo(todo_id: int, todo: Todo, db: Session = Depends(get_db)):
    todo_instance = db.query(Todo).filter(Todo.id == todo_id).first()
    if todo_instance is None:
        raise HTTPException(status_code=404, detail="Todo not found")

    todo_instance.title = todo.title
    todo_instance.completed = todo.completed
    db.commit()
    db.refresh(todo_instance)
    return todo_instance

5.4 Todo 삭제하기

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo_instance = db.query(Todo).filter(Todo.id == todo_id).first()
    if todo_instance is None:
        raise HTTPException(status_code=404, detail="Todo not found")

    db.delete(todo_instance)
    db.commit()
    return {"detail": "Todo deleted"} 

6. Done 리소스 구현하기

이번에는 완료된 Todo를 관리하는 Done 리소스를 구현해보겠습니다. 아래와 같이 새로운 엔드포인트를 추가합니다.

6.1 Done 목록 조회하기

@app.get("/done/", response_model=List[Todo])
async def read_done_todos(db: Session = Depends(get_db)):
    done_todos = db.query(Todo).filter(Todo.completed == True).all()
    return done_todos

7. 클라이언트 테스트하기

이제 Postman 등의 클라이언트를 사용하여 엔드포인트를 테스트해볼 수 있습니다. 아래는 각 엔드포인트에 대한 간단한 테스트 예시입니다.

  • POST /todos/: 새로운 Todo를 생성합니다.
  • GET /todos/: 모든 Todo 목록을 조회합니다.
  • PUT /todos/{todo_id}: 특정 Todo의 내용을 수정합니다.
  • DELETE /todos/{todo_id}: 특정 Todo를 삭제합니다.
  • GET /done/: 완료된 Todo 목록을 조회합니다.

8. 결론

이 글에서는 FastAPI를 사용하여 기본적인 CRUD API를 구축하고, 데이터베이스를 조작하는 방법에 대해 알아보았습니다. FastAPI는 간결한 문법과 뛰어난 성능으로 인해 현대 웹 애플리케이션 개발에 많은 도움이 됩니다. 더 나아가 부가적인 기능(예: JWT 인증, CORS 처리 등)을 추가하여 강력한 API 서버를 만들 수 있습니다.

FastAPI를 통해 여러분의 프로젝트에 더 많은 기능과 서비스를 추가해 보세요. Happy coding!

FastAPI 서버개발, 비동기 대응 CRUDs

최근 몇 년간 FastAPI는 Python에서 비동기 웹 프레임워크로 많은 인기를 얻고 있습니다. FastAPI는 비동기 프로그래밍의 장점을 제공하며, RESTful API를 쉽게 만들 수 있는 강력한 도구입니다. 이 글에서는 FastAPI를 사용하여 비동기적으로 CRUD(생성, 읽기, 수정, 삭제) 작업을 수행하는 방법에 대해 자세히 설명하겠습니다.

1. FastAPI란?

FastAPI는 Python으로 구축된 현대적인 웹 프레임워크로, 고성능 비동기 웹 API를 만드는 데 중점을 두고 설계되었습니다. 주요 특징으로는 다음과 같습니다:

  • 빠른 성능: Starlette와 Pydantic을 기반으로 하므로 속도가 매우 빠릅니다.
  • 자동 문서화: Swagger UI 및 ReDoc과 같은 자동화된 API 문서를 제공합니다.
  • 타입 힌트 지원: Python 타입 힌트를 사용하여 코드의 가독성을 높이고 오류를 줄일 수 있습니다.

2. 환경 설정

FastAPI를 시작하기 위해서는 Python과 pip(패키지 관리자)가 설치되어 있어야 합니다. 아래의 명령어로 FastAPI와 Uvicorn(ASGI 서버)를 설치할 수 있습니다:

pip install fastapi uvicorn

2.1. 프로젝트 구조 설정

간단한 프로젝트 구조를 설정합니다:


my_fastapi_app/
│
├── main.py        # FastAPI 애플리케이션
└── models.py      # Pydantic 모델

3. 기본 CRUD API 구현

이제 기본적인 CRUD 작업을 구현해보겠습니다. 우리는 사용할 데이터 모델을 정의하고, 이를 기반으로 CRUD 엔드포인트를 만들어 보겠습니다.

3.1. 모델 정의

우선 models.py 파일에 Pydantic 모델을 정의합니다:


from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

3.2. CRUD 작업 구현

main.py 파일에 CRUD 작업을 구현합니다:


from fastapi import FastAPI, HTTPException
from typing import List
from models import Item

app = FastAPI()

# 메모리에 데이터를 저장하기 위한 리스트
fake_items_db = []

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    fake_items_db.append(item)
    return item

@app.get("/items/", response_model=List[Item])
async def read_items():
    return fake_items_db

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    item = next((item for item in fake_items_db if item.id == item_id), None)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, updated_item: Item):
    for index, item in enumerate(fake_items_db):
        if item.id == item_id:
            fake_items_db[index] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    global fake_items_db
    fake_items_db = [item for item in fake_items_db if item.id != item_id]
    return {"message": "Item deleted successfully"}

4. 비동기 처리

FastAPI는 비동기 프로그래밍을 지원합니다. 설정한 CRUD 작업은 이미 비동기 함수로 정의되어 있으므로, 비동기적 요청 처리를 자동으로 활용할 수 있습니다. 다수의 사용자가 동시에 요청을 보내더라도, FastAPI는 효율적으로 요청을 처리할 수 있습니다.

4.1. 비동기 데이터베이스 연결

실제로는 인메모리 데이터베이스를 사용하지 않고, 데이터베이스와의 비동기 연결을 사용하는 것이 일반적입니다. 아래는 Async SQLAlchemy를 사용한 예시입니다:


from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

# FastAPI 애플리케이션 생성
app = FastAPI()

# 데이터베이스 세션 종속성 정의
async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

4.2. 비동기 CRUD 작업

이제 데이터베이스를 사용할 수 있으므로 CRUD 작업을 비동기 방식으로 업데이트합니다:


@app.post("/items/", response_model=Item)
async def create_item(item: Item, db: AsyncSession = Depends(get_db)):
    db.add(item)
    await db.commit()
    return item

@app.get("/items/", response_model=List[Item])
async def read_items(db: AsyncSession = Depends(get_db)):
    return await db.execute(select(Item)).scalars().all()

5. 테스트 및 디버깅

FastAPI는 자동으로 Swagger UI를 제공하므로, API를 테스트하는 데 매우 유용합니다. 서버를 실행하면 http://127.0.0.1:8000/docs에서 API 문서를 확인할 수 있습니다.

uvicorn main:app --reload

6. 결론

FastAPI는 비동기 프로그래밍을 통해 효율적인 API 서버를 구축할 수 있는 매우 강력한 도구입니다. 이번 글에서 소개한 CRUD 예제는 기본적인 개념을 이해하는 데 도움이 될 것입니다. 실제 프로젝트에서는 데이터베이스와의 연결, 사용자 인증, 그리고 비즈니스 로직을 추가하여 더욱 발전시킬 수 있습니다.

FastAPI의 장점인 고성능과 자동 문서화 기능을 활용하여 다양한 API를 손쉽게 개발해 보시기 바랍니다.

7. 참고 자료