본문 바로가기
프로그램 (PHP,Python)

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

by 날으는물고기 2024. 2. 10.

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 코드를 네이티브로 지원합니다. asyncawait 키워드를 사용하여 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의 추가적인 기능을 활용할 수 있습니다.

 

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

 

  1. 간단한 비동기 애플리케이션 작성: 스레드 관리 오버헤드 감소를 통한 성능 향상
  • 설명: 단순한 비동기 애플리케이션을 작성하면 스레드 관리 오버헤드가 줄어들어 성능이 향상될 수 있습니다.
  • 예시:
    from fastapi import FastAPI
    
    # FastAPI 앱 인스턴스 생성
    app = FastAPI()
    
    # 루트 경로('/')에 대한 GET 요청을 처리하는 비동기 엔드포인트
    @app.get("/")
    async def read_root():
        # JSON 형태의 응답 반환
        return {"Hello": "World"}
  1. uvloop 사용하기: asyncio 이벤트 루프 최적화
  • 설명: uvloop을 사용하여 asyncio 이벤트 루프를 최적화할 수 있습니다. 이는 비동기 프로그래밍을 빠르고 효율적으로 만들어줍니다.
  • 예시:
    import uvloop
    import asyncio
    
    # uvloop를 기본 이벤트 루프로 설치
    uvloop.install()
    
    # 메인 비동기 함수 정의
    async def main():
        # 비동기 코드 작성 부분
        pass  # 비동기 코드를 여기에 작성
    
    # 메인 함수 실행
    asyncio.run(main())
  1. 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())  # 이 부분은 수정이 필요합니다.
  1. 더 큰 스레드풀 사용하기: 동시 요청 처리 능력 향상
  • 설명: Uvicorn은 기본적으로 스레드 풀을 사용하므로, 풀의 크기를 늘리면 동시에 더 많은 요청을 처리할 수 있습니다.
  • 예시: Uvicorn 실행 명령어에 --workers 플래그를 사용하여 스레드 풀 크기를 지정합니다.
    uvicorn your_app:app --workers 4
  1. 중복 유효성 검사 제거하기: 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' 필드를 응답에 포함하여 반환합니다.
  1. 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)
  1. 유효성 검사 및 로깅 생략 옵션 고려하기: 성능 최적화를 위한 고려 사항
  • 설명: 유효성 검사나 로깅을 생략하면 성능은 향상되지만, 이는 일반적으로 권장되는 방법은 아닙니다.
  • 예시:
    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}
  1. ASGI 미들웨어 사용하기: 애플리케이션 기능 확장
  • 설명: ASGI 미들웨어를 사용하여 애플리케이션에 다양한 확장 기능을 추가할 수 있습니다.
  • 예시: Starlette 미들웨어를 사용하여 애플리케이션에 인증, 로깅 또는 사용자 정의 기능을 추가합니다.

 

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

 

  1. asyncpg를 이용한 데이터베이스 작업 최적화: 비동기 PostgreSQL 드라이버 활용
  • 설명: asyncpg는 비동기 PostgreSQL 드라이버로, 데이터베이스 작업에 효율적으로 사용됩니다.
  • 예시: FastAPI에서 databasesasyncpg를 사용하여 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)
  1. 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}
  1. 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}
  1. 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
  1. 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"}
  1. Uvicorn 설정 조정으로 동시성 향상하기: 워커 및 동시성 한계 조정
  • 설명: Uvicorn의 설정을 조정하여 동시성을 향상시킬 수 있습니다.
  • 예시: Uvicorn 실행 명령어에 --workers--limit-concurrency 플래그를 사용하여 동시성을 조절합니다.
    uvicorn your_app:app --workers 4 --limit-concurrency 1000

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

728x90

댓글