← 가이드 목록으로 ← Back to guides
🐍 Python #3

Python 비동기 프로그래밍 & FastAPI 완벽 가이드 Python Async Programming & FastAPI Complete Guide

📖 약 18분 읽기 📖 18 min read 🏷️ asyncio, FastAPI, pytest, aiohttp

Python의 비동기 프로그래밍은 I/O 바운드 작업에서 성능을 극대화합니다. asyncio로 비동기 코드를 작성하고, FastAPI로 현대적인 웹 API를 만들어봅니다.

Python's async programming maximizes performance for I/O-bound operations. Write async code with asyncio and build modern web APIs with FastAPI.

asyncio 기초: async/await asyncio Basics: async/await

async def로 코루틴을 정의하고, await로 비동기 작업을 기다립니다.

Define coroutines with async def and wait for async operations with await.

import asyncio

# 기본 코루틴
async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 코루틴 실행
asyncio.run(hello())

# 값 반환
async def fetch_data() -> str:
    await asyncio.sleep(0.5)
    return "데이터 로드 완료"

result = asyncio.run(fetch_data())
print(result)
import asyncio

# Basic coroutine
async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# Run coroutine
asyncio.run(hello())

# Return value
async def fetch_data() -> str:
    await asyncio.sleep(0.5)
    return "Data loaded"

result = asyncio.run(fetch_data())
print(result)

동시 실행: gather와 TaskGroup Concurrent Execution: gather & TaskGroup

import asyncio

async def task(name: str, delay: float):
    await asyncio.sleep(delay)
    return f"{name} 완료"

# asyncio.gather 사용
async def main_gather():
    results = await asyncio.gather(
        task("A", 1),
        task("B", 2),
        task("C", 1.5)
    )
    print(results) # ['A 완료', 'B 완료', 'C 완료']

# Python 3.11+ TaskGroup (권장)
async def main_taskgroup():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(task("A", 1))
        task2 = tg.create_task(task("B", 2))
    print(task1.result(), task2.result())
import asyncio

async def task(name: str, delay: float):
    await asyncio.sleep(delay)
    return f"{name} done"

# Using asyncio.gather
async def main_gather():
    results = await asyncio.gather(
        task("A", 1),
        task("B", 2),
        task("C", 1.5)
    )
    print(results) # ['A done', 'B done', 'C done']

# Python 3.11+ TaskGroup (recommended)
async def main_taskgroup():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(task("A", 1))
        task2 = tg.create_task(task("B", 2))
    print(task1.result(), task2.result())

aiohttp: 비동기 HTTP 클라이언트 aiohttp: Async HTTP Client

import aiohttp
import asyncio

async def fetch(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# 여러 URL 동시 요청
async def fetch_all(urls: list[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [await r.json() for r in responses]
import aiohttp
import asyncio

async def fetch(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# Concurrent requests to multiple URLs
async def fetch_all(urls: list[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [await r.json() for r in responses]

FastAPI: 현대적인 웹 프레임워크 FastAPI: Modern Web Framework

FastAPI는 Python 3.7+에서 동작하는 고성능 웹 프레임워크입니다. 자동 문서화, 타입 힌트 기반 검증, 비동기 지원이 특징입니다.

FastAPI is a high-performance web framework for Python 3.7+. Features include auto-documentation, type hint-based validation, and async support.

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

# 요청/응답 모델
class User(BaseModel):
    name: str
    email: str
    age: int | None = None

# GET 엔드포인트
@app.get("/")
async def root():
    return {"message": "Hello World"}

# 경로 파라미터
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# POST 요청
@app.post("/users")
async def create_user(user: User):
    return {"created": user.model_dump()}
# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

# Request/Response model
class User(BaseModel):
    name: str
    email: str
    age: int | None = None

# GET endpoint
@app.get("/")
async def root():
    return {"message": "Hello World"}

# Path parameters
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# POST request
@app.post("/users")
async def create_user(user: User):
    return {"created": user.model_dump()}

💡 FastAPI 실행하기 💡 Running FastAPI

# 설치
$ pip install fastapi uvicorn

# 개발 서버 실행
$ uvicorn main:app --reload

# API 문서 확인
# http://localhost:8000/docs (Swagger UI)
# http://localhost:8000/redoc (ReDoc)

pytest: 테스트 작성 pytest: Writing Tests

# test_main.py
import pytest
from httpx import AsyncClient
from main import app

# 비동기 테스트
@pytest.mark.asyncio
async def test_root():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.json() == {"message": "Hello World"}

# 파라미터화된 테스트
@pytest.mark.parametrize("user_id,expected", [
    (1, {"user_id": 1}),
    (42, {"user_id": 42}),
])
@pytest.mark.asyncio
async def test_get_user(user_id, expected):
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get(f"/users/{user_id}")
        assert response.json() == expected
# test_main.py
import pytest
from httpx import AsyncClient
from main import app

# Async test
@pytest.mark.asyncio
async def test_root():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.json() == {"message": "Hello World"}

# Parameterized test
@pytest.mark.parametrize("user_id,expected", [
    (1, {"user_id": 1}),
    (42, {"user_id": 42}),
])
@pytest.mark.asyncio
async def test_get_user(user_id, expected):
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get(f"/users/{user_id}")
        assert response.json() == expected

✨ 비동기 프로그래밍 Best Practices ✨ Async Programming Best Practices

  • await는 코루틴 앞에만 사용
  • 동시 실행에는 gather 또는 TaskGroup 사용
  • CPU 바운드 작업은 run_in_executor 사용
  • 세션/연결은 async with로 관리
  • Use await only before coroutines
  • Use gather or TaskGroup for concurrency
  • Use run_in_executor for CPU-bound tasks
  • Manage sessions/connections with async with

⚠️ 흔한 실수들 ⚠️ Common Mistakes

  • await 없이 코루틴 호출 (실행되지 않음)
  • 동기 함수 안에서 await 사용 불가
  • time.sleep() 대신 asyncio.sleep() 사용
  • Calling coroutine without await (won't execute)
  • Can't use await inside sync functions
  • Use asyncio.sleep() instead of time.sleep()