비동기와 백그라운드 작업

@router.get("/sync")
def sync_handler():
    ...

@router.get("/async")
async def async_handler():
    ...

@router.get("/items")
async def list_items():
    response = requests.get("https://api.example.com")   # 동기, 블로킹!
    return response.json()

import httpx

@router.get("/items")
async def list_items():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")
    return response.json()

import asyncio
import requests

@router.get("/items")
async def list_items():
    response = await asyncio.to_thread(requests.get, "https://api.example.com")
    return response.json()

@router.get("/items")
def list_items():
    response = requests.get("https://api.example.com")
    return response.json()

from fastapi import BackgroundTasks

def send_email_sync(to: str, subject: str, body: str):
    # 실제 이메일 전송 (smtplib 등)
    print(f"이메일 보냄: {to}")

@router.post("/users")
async def create_user(
    payload: UserCreate,
    background_tasks: BackgroundTasks,
    db: DBSession,
) -> UserOut:
    user = await user_service.create_user(db, payload.email, payload.password)
    background_tasks.add_task(
        send_email_sync, user.email, "환영", "가입을 환영합니다",
    )
    return user

async def send_email_async(to: str, subject: str, body: str):
    async with httpx.AsyncClient() as client:
        await client.post("https://email-api/", json={...})

background_tasks.add_task(send_email_async, user.email, "환영", "...")

uv add arq

from arq.connections import RedisSettings
from app.core.config import settings

async def send_email(ctx, to: str, subject: str, body: str):
    print(f"[worker] 이메일 보냄: {to}")
    # 실제 전송 코드

class WorkerSettings:
    redis_settings = RedisSettings.from_dsn(settings.redis_url)
    functions = [send_email]
    max_jobs = 10
    job_timeout = 60

from arq import create_pool
from arq.connections import RedisSettings

@router.post("/users")
async def create_user(payload: UserCreate, db: DBSession):
    user = await user_service.create_user(...)
    pool = await create_pool(RedisSettings.from_dsn(settings.redis_url))
    await pool.enqueue_job("send_email", user.email, "환영", "...")
    return user

uv run arq app.tasks.worker.WorkerSettings

from app.db.session import AsyncSessionLocal

async def send_email(ctx, user_id: int):
    async with AsyncSessionLocal() as db:
        user = await db.get(User, user_id)
        # 발송 로직

from fastapi.responses import StreamingResponse

@router.get("/csv")
async def export_csv():
    async def generate():
        yield "id,title\n"
        async for row in stream_rows():
            yield f"{row.id},{row.title}\n"

    return StreamingResponse(generate(), media_type="text/csv")

@router.get("/events")
async def events():
    async def event_stream():
        for i in range(100):
            yield f"data: tick {i}\n\n"
            await asyncio.sleep(1)

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
    )

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws/{room}")
async def websocket_endpoint(websocket: WebSocket, room: str):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[{room}] echo: {data}")
    except WebSocketDisconnect:
        print(f"클라이언트 연결 종료")

import asyncio

_external_sem = asyncio.Semaphore(10)

async def call_external(url: str) -> dict:
    async with _external_sem:
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            return response.json()

async with asyncio.timeout(5):
    response = await client.get(url)

async with httpx.AsyncClient(timeout=10.0) as client:
    ...

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 시작 시
    print("서버 시작")
    redis = await create_redis_pool()
    app.state.redis = redis

    yield

    # 종료 시
    print("서버 종료")
    await redis.close()

app = FastAPI(lifespan=lifespan)

def task_that_might_fail():
    raise SomethingWrong()    # 응답엔 영향 없지만, 로그만 남음

@router.post("/")
async def create(db: DBSession, bg: BackgroundTasks):
    bg.add_task(use_db_later, db)   # ← 위험
    return ...

async def use_db_later(user_id: int):
    async with AsyncSessionLocal() as db:
        ...

bg.add_task(use_db_later, user.id)

@router.get("/")
def handler():
    asyncio.run(something_async())   # 이벤트 루프 중첩 — 에러