FastAPI, Uvicorn, Pydantic 및 Starlette 활용에 대한 유용한 팁

FastAPI, Uvicorn, Pydantic, 그리고 Starlette은 현대적인 Python 웹 개발에서 매우 인기 있는 도구들입니다. 이들 각각은 웹 애플리케이션을 구축, 배포 및 관리하는 데 있어 특정 기능을 제공합니다. 여기에 각각의 도구가 어떻게 상호작용하며, 그들을 활용하는 데 있어 유용한 팁을 요약해 보겠습니다.

FastAPI

소개: FastAPI는 현대적이고 빠른(고성능) 웹 프레임워크로, Python 3.6 이상에서 비동기 프로그래밍을 사용합니다. API 개발에 최적화되어 있으며, Pydantic을 사용한 데이터 검증과 자동 문서 생성 기능을 제공합니다.
활용 팁:
- 타입 힌트와 Pydantic 모델: FastAPI는 타입 힌트와 Pydantic 모델을 사용하여 데이터 검증과 직렬화를 자동으로 처리합니다. 이를 통해 개발 속도를 높이고, 오류를 줄일 수 있습니다.
- 자동 문서 생성: FastAPI는 Swagger와 ReDoc을 통해 API 문서를 자동으로 생성합니다. 이 기능을 활용하여 개발자와 사용자가 API를 쉽게 이해하고 사용할 수 있도록 합니다.
- 비동기 지원: FastAPI는 비동기 Python 코드를 네이티브로 지원합니다. async와 await 키워드를 사용하여 I/O 바운드 작업을 최적화하고, 애플리케이션의 성능을 향상시킬 수 있습니다.

Uvicorn

소개: Uvicorn은 FastAPI와 같은 ASGI(Asynchronous Server Gateway Interface) 애플리케이션을 위한 경량, 고성능 ASGI 서버입니다.
활용 팁:
- 개발 서버로 사용: 개발 중에는 Uvicorn을 직접 실행하여 빠른 반복 개발과 테스트를 할 수 있습니다.
- 배포 시 Gunicorn과 함께 사용: 배포 환경에서는 Uvicorn을 Gunicorn의 워커 클래스로 사용하여 복수의 프로세스를 관리하고, 더 안정적인 서비스를 제공할 수 있습니다.

Pydantic

소개: Pydantic은 데이터 검증과 설정 관리를 위한 Python 라이브러리입니다. 타입 힌트를 사용하여 데이터의 구조를 정의하고, 자동으로 데이터 타입을 검증합니다.
활용 팁:
- 모델 정의: API 요청과 응답에 사용될 데이터 모델을 Pydantic을 사용하여 정의합니다. 이는 데이터 검증, 직렬화, 문서화를 자동화하는 데 도움이 됩니다.
- 환경 설정 관리: Pydantic은 환경 변수를 통한 애플리케이션 설정 관리를 쉽게 할 수 있게 해줍니다. BaseSettings 클래스를 사용하여 설정 값을 모델로 정의하고, 자동으로 환경 변수에서 값을 로드할 수 있습니다.

Starlette

소개: Starlette은 FastAPI의 기반이 되는 경량 ASGI 프레임워크입니다. 비동기 웹 애플리케이션 개발을 위한 기본적인 도구를 제공합니다.
활용 팁:
- 경량 서비스 구축: 복잡한 기능이 필요 없는 간단한 비동기 서비스를 구축할 때 Starlette을 직접 사용 할 수 있습니다.
- 미들웨어와 이벤트 핸들러: Starlette은 요청 전후에 실행할 수 있는 미들웨어와 이벤트 핸들러를 지원합니다. 이를 통해 요청 처리 과정을 유연하게 제어할 수 있습니다.

이러한 도구들은 함께 사용될 때 강력한 현대적 웹 애플리케이션 개발 환경을 제공합니다. FastAPI로 빠르고 효율적인 API를 구축하고, Uvicorn으로 비동기 서비스를 호스팅하며, Pydantic으로 데이터를 안전하게 처리하고, 필요에 따라 Starlette의 추가적인 기능을 활용할 수 있습니다.

FastAPI 최신 버전 사용: Pydantic의 Rust 구현을 통한 성능 향상

설명: FastAPI의 최신 버전을 사용하면 Pydantic이 Rust로 구현되어 성능이 크게 향상됩니다.
예시: FastAPI 및 Pydantic을 최신 버전으로 업그레이드합니다.

간단한 비동기 애플리케이션 작성: 스레드 관리 오버헤드 감소를 통한 성능 향상

설명: 단순한 비동기 애플리케이션을 작성하면 스레드 관리 오버헤드가 줄어들어 성능이 향상될 수 있습니다.

예시:

from fastapi import FastAPI

# FastAPI 앱 인스턴스 생성
app = FastAPI()

# 루트 경로('/')에 대한 GET 요청을 처리하는 비동기 엔드포인트
@app.get("/")
async def read_root():
    # JSON 형태의 응답 반환
    return {"Hello": "World"}

uvloop 사용하기: asyncio 이벤트 루프 최적화

설명: uvloop을 사용하여 asyncio 이벤트 루프를 최적화할 수 있습니다. 이는 비동기 프로그래밍을 빠르고 효율적으로 만들어줍니다.

예시:

import uvloop
import asyncio

# uvloop를 기본 이벤트 루프로 설치
uvloop.install()

# 메인 비동기 함수 정의
async def main():
    # 비동기 코드 작성 부분
    pass  # 비동기 코드를 여기에 작성

# 메인 함수 실행
asyncio.run(main())

httptools 사용하기: 빠른 HTTP 파싱

설명: httptools는 HTTP 파싱을 더 빠르게 수행할 수 있도록 도와주는 라이브러리입니다.

예시:

from httptools import HttpRequestParser

# HttpRequestParser 인스턴스 생성
parser = HttpRequestParser()

# 요청 데이터를 처리하는 함수
def process_request(data):
    # 데이터를 파싱하고 요청 처리하는 로직을 여기에 구현
    pass  # 실제 요청 처리 로직

# 테스트를 위한 HTTP 요청 데이터
data = b'GET /path HTTP/1.1\r\nHost: example.com\r\n\r\n'

# parser에 데이터를 공급하여 파싱
parser.feed_data(data)

# 파싱된 요청을 처리
# 주의: 이 예제에서는 parser.get_request() 사용이 잘못되었습니다.
# HttpRequestParser에는 get_request() 메서드가 없습니다.
# 올바른 사용 방법은 HttpRequestParser를 상속받는 클래스를 정의하고,
# on_url(), on_header()와 같은 콜백 메서드를 구현하는 것입니다.
process_request(parser.get_request())  # 이 부분은 수정이 필요합니다.

더 큰 스레드풀 사용하기: 동시 요청 처리 능력 향상

설명: Uvicorn은 기본적으로 스레드 풀을 사용하므로, 풀의 크기를 늘리면 동시에 더 많은 요청을 처리할 수 있습니다.
예시: Uvicorn 실행 명령어에 --workers 플래그를 사용하여 스레드 풀 크기를 지정합니다.
```
uvicorn your_app:app --workers 4
```

중복 유효성 검사 제거하기: FastAPI에 의한 유효성 검사로 성능 향상

설명: 중복된 유효성 검사를 제거하면 처리 시간이 단축되어 성능이 향상됩니다.

예시:

from pydantic import BaseModel  # Pydantic을 임포트하여 데이터 모델링 도구를 사용
from fastapi import FastAPI  # FastAPI 프레임워크 임포트

# Pydantic 모델 정의
class Item(BaseModel):
    name: str  # Item 모델에는 'name' 필드가 있으며, 타입은 문자열입니다.

# FastAPI 앱 인스턴스 생성
app = FastAPI()

# '/items/' 경로에 POST 요청을 받는 엔드포인트 정의
@app.post("/items/")
async def create_item(item: Item):  # 비동기 함수로 정의하고, 요청 바디는 Item 모델로 자동 검증됩니다.
    # FastAPI가 Pydantic 모델을 사용하여 자동으로 요청 데이터의 유효성 검사를 수행합니다.
    # 따라서 별도의 유효성 검사 코드를 작성할 필요가 없습니다.
    return {"item_name": item.name}  # 검증된 데이터의 'name' 필드를 응답에 포함하여 반환합니다.

ORJSON 사용하기: JSON 처리 성능 향상

설명: ORJSON은 JSON 처리를 효율적으로 수행하여 애플리케이션 성능을 향상시킵니다.

예시:

from fastapi.responses import JSONResponse  # FastAPI에서 JSONResponse 임포트
import orjson  # orjson 라이브러리 임포트

# 직렬화할 데이터 정의
data = {"key": "value"}

# ORJSON을 사용하여 JSON 직렬화
# 이 단계는 JSONResponse 객체를 생성할 때 직접적으로 필요하지 않지만, orjson의 사용법을 보여줍니다.
json_str = orjson.dumps(data)

# FastAPI의 JSONResponse에 ORJSON 사용하여 응답 객체 생성
# content에는 응답으로 보낼 데이터를 직접 전달합니다.
# dumps 매개변수에는 orjson.dumps 함수를 할당하여, FastAPI가 응답 데이터를 직렬화할 때 ORJSON을 사용하도록 설정합니다.
response = JSONResponse(content=data, dumps=orjson.dumps)

유효성 검사 및 로깅 생략 옵션 고려하기: 성능 최적화를 위한 고려 사항

설명: 유효성 검사나 로깅을 생략하면 성능은 향상되지만, 이는 일반적으로 권장되는 방법은 아닙니다.

예시:

from fastapi import FastAPI  # FastAPI 프레임워크 임포트

# FastAPI 앱 인스턴스 생성
app = FastAPI()

# '/items/' 경로에 대한 GET 요청을 처리하는 엔드포인트 정의
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):  # 비동기 함수로 정의, skip과 limit은 쿼리 파라미터로, 기본값을 갖습니다.
    # 이 함수는 유효성 검사나 로깅을 추가로 수행하지 않고,
    # 단순히 요청에서 받은 skip과 limit 값을 그대로 응답에 포함하여 반환합니다.
    return {"skip": skip, "limit": limit}

ASGI 미들웨어 사용하기: 애플리케이션 기능 확장

설명: ASGI 미들웨어를 사용하여 애플리케이션에 다양한 확장 기능을 추가할 수 있습니다.
예시: Starlette 미들웨어를 사용하여 애플리케이션에 인증, 로깅 또는 사용자 정의 기능을 추가합니다.

응답 압축하기: 대역폭 절약 및 응답 속도 향상

설명: 응답을 압축하여 전송하면 대역폭을 절약하고 응답 시간을 단축할 수 있습니다.
예시: FastAPI에서는 미들웨어를 사용하여 응답을 압축할 수 있습니다. Starlette의 GZipMiddleware를 사용하는 방법이 있습니다. 자세한 내용은 Starlette 미들웨어 문서를 참조하세요.

asyncpg를 이용한 데이터베이스 작업 최적화: 비동기 PostgreSQL 드라이버 활용

설명: asyncpg는 비동기 PostgreSQL 드라이버로, 데이터베이스 작업에 효율적으로 사용됩니다.

예시: FastAPI에서 databases 및 asyncpg를 사용하여 PostgreSQL과 통합할 수 있습니다.이 예시에서는 asyncpg를 사용하여 PostgreSQL에 연결하고, databases 모듈을 사용하여 FastAPI 애플리케이션에서 비동기적으로 데이터베이스 작업을 수행합니다. 이를 통해 데이터베이스 작업이 성능 향상을 가져올 수 있습니다.

import databases  # 비동기 데이터베이스 지원을 위한 라이브러리
import sqlalchemy  # SQL 쿼리 생성 및 데이터베이스 스키마 관리를 위한 라이브러리
from fastapi import FastAPI, Depends  # FastAPI 웹 프레임워크 및 의존성 주입 기능

# 데이터베이스 연결 URL 정의
DATABASE_URL = "postgresql://user:password@localhost/db"

# 비동기 데이터베이스 객체 생성
database = databases.Database(DATABASE_URL)

# SQLAlchemy 메타데이터 객체 생성
metadata = sqlalchemy.MetaData()

# 'items' 테이블 정의
items = sqlalchemy.Table(
    "items",  # 테이블 이름
    metadata,  # 메타데이터 객체
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, index=True),  # 'id' 컬럼 정의
    sqlalchemy.Column("name", sqlalchemy.String, index=True),  # 'name' 컬럼 정의
)

# SQLAlchemy 엔진 객체 생성
engine = sqlalchemy.create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite를 사용하는 경우 필요한 옵션
)

# 데이터베이스 스키마 생성 (테이블 생성)
metadata.create_all(engine)

# FastAPI 앱 인스턴스 생성
app = FastAPI()

# '/items/' 경로에 대한 POST 요청을 처리하는 엔드포인트
@app.post("/items/")
async def create_item(name: str, db: database = Depends(database)):
    # 'items' 테이블에 새로운 아이템을 삽입하는 쿼리 생성
    query = items.insert().values({"name": name})
    # 쿼리 실행 및 결과 반환
    return await db.execute(query)

Redis를 이용한 캐싱: 데이터 검색 속도 및 성능 향상

설명: Redis를 사용하여 캐싱을 구현하면 중복된 계산을 피하고 데이터를 더 빠르게 검색할 수 있습니다.

예시: FastAPI에서 aioredis를 사용하여 Redis와 통합할 수 있습니다.

import aioredis
from fastapi import FastAPI, Depends

app = FastAPI()

# Redis 연결 초기화를 위한 비동기 함수 정의
async def get_redis():
    redis = await aioredis.create_redis_pool('redis://localhost')
    try:
        yield redis
    finally:
        await redis.close()

# 복잡한 데이터 계산을 수행하는 함수 (예시로 추가)
async def compute_expensive_data():
    # 복잡한 계산을 수행하는 가정 (실제 구현 필요)
    return "expensive_data"

# '/cached_data/' 경로에 대한 GET 요청 처리
@app.get("/cached_data/")
async def get_cached_data(key: str, redis: aioredis.Redis = Depends(get_redis)):
    cached_data = await redis.get(key)
    if cached_data:
        # 캐시된 데이터가 있으면 반환
        return {"data": cached_data.decode('utf-8')}
    else:
        # 캐시되지 않은 경우, 복잡한 데이터 계산 수행
        data = await compute_expensive_data()
        # 계산된 데이터를 Redis에 저장하고, 1시간 후 만료되도록 설정
        await redis.set(key, data, expire=3600)
        return {"data": data}

Pydantic의 conlist 사용하기: 대량 리스트 데이터 처리 최적화

설명: Pydantic에서 conlist를 사용하여 대량의 데이터를 빠르게 처리할 수 있습니다.

예시:

from typing import List
from pydantic import BaseModel, conlist  # Pydantic 모델링 도구 임포트
from fastapi import FastAPI  # FastAPI 프레임워크 임포트

# Pydantic을 사용한 데이터 모델 정의
class Item(BaseModel):
    name: str  # 아이템 이름
    tags: conlist(str, max_items=10)  # 최대 10개의 문자열 태그를 가질 수 있는 리스트

# FastAPI 앱 인스턴스 생성
app = FastAPI()

# '/items/' 경로에 대한 POST 요청을 처리하는 엔드포인트 정의
@app.post("/items/")
async def create_item(item: Item):  # 비동기 함수로 정의, 요청 바디는 Item 모델로 자동 검증됩니다.
    # FastAPI와 Pydantic을 사용하여 요청 바디의 유효성 검증 및 데이터 추출을 자동으로 처리합니다.
    # 이 함수는 검증된 아이템 데이터를 반환합니다.
    return {"item_name": item.name, "tags": item.tags}

Pydantic의 parse_obj_as로 데이터 변환 최적화: 효율적인 데이터 변환 처리

설명: Pydantic의 parse_obj_as를 사용하여 데이터 변환을 효율적으로 처리할 수 있습니다.

예시:

from pydantic import BaseModel, parse_obj_as  # Pydantic의 기본 모델 및 객체 파싱 함수 임포트
from typing import List  # 타입 힌팅을 위한 List 임포트
from fastapi import FastAPI  # FastAPI 웹 프레임워크 임포트

# 입력 데이터를 위한 Pydantic 모델 정의
class ItemIn(BaseModel):
    name: str  # 아이템 이름

# 출력 데이터를 위한 Pydantic 모델 정의
class ItemOut(BaseModel):
    name: str  # 변환된 아이템 이름

# FastAPI 앱 인스턴스 생성
app = FastAPI()

# '/items/' 경로에 대한 POST 요청을 처리하는 엔드포인트 정의
@app.post("/items/", response_model=List[ItemOut])
async def create_items(items: List[ItemIn]):
    # 요청받은 아이템 리스트를 ItemOut 리스트로 변환
    # parse_obj_as 함수를 사용하여 ItemIn 객체 리스트를 ItemOut 객체 리스트로 효율적으로 변환
    items_out = parse_obj_as(List[ItemOut], items)
    return items_out

FastAPI의 백그라운드 작업 사용하기: 비동기 작업 처리를 통한 응답 시간 단축

설명: 백그라운드 작업을 사용하여 요청-응답 주기를 단축하고 비동기적으로 처리 가능한 작업을 처리할 수 있습니다.

예시:

from fastapi import BackgroundTasks, FastAPI  # FastAPI와 BackgroundTasks 임포트

app = FastAPI()  # FastAPI 앱 인스턴스 생성

# 로그 파일에 메시지를 기록하는 함수
def write_log(message: str):
    with open("logs.txt", mode="a") as log:  # 'logs.txt' 파일을 추가 모드로 열기
        log.write(message + "\n")  # 메시지를 파일에 쓰고 개행 추가

# '/send-notification/{email}' 경로에 대한 POST 요청 처리
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # 로그 메시지 생성
    message = f"message to {email}"
    # BackgroundTasks를 사용하여 write_log 함수를 백그라운드 작업으로 추가
    background_tasks.add_task(write_log, message)
    # 응답 반환
    return {"message": "Message sent"}

Uvicorn 설정 조정으로 동시성 향상하기: 워커 및 동시성 한계 조정

설명: Uvicorn의 설정을 조정하여 동시성을 향상시킬 수 있습니다.
예시: Uvicorn 실행 명령어에 --workers 및 --limit-concurrency 플래그를 사용하여 동시성을 조절합니다.
```
uvicorn your_app:app --workers 4 --limit-concurrency 1000
```

이러한 팁들을 활용하면 FastAPI 애플리케이션의 성능을 더욱 향상시킬 수 있습니다.

728x90

저작자표시 비영리 동일조건

pages.kr 날으는물고기 <º))))><

FastAPI, Uvicorn, Pydantic 및 Starlette 활용에 대한 유용한 팁

FastAPI, Uvicorn, Pydantic 및 Starlette 활용에 대한 유용한 팁

FastAPI

Uvicorn

Pydantic

Starlette

댓글

티스토리툴바

FastAPI, Uvicorn, Pydantic 및 Starlette 활용에 대한 유용한 팁

FastAPI, Uvicorn, Pydantic 및 Starlette 활용에 대한 유용한 팁

FastAPI

Uvicorn

Pydantic

Starlette

관련글

댓글

티스토리툴바