[Python] FastAPI 비동기 처리(3) 비동기 백그라운드 작업 구현하기

이전 글

2025.05.15 - [Python] - [Python] FastAPI 비동기 처리(1) 비동기 엔드포인트 만들기

[Python] FastAPI 비동기 처리(1) 비동기 엔드포인트 만들기

2025.05.22 - [Python] - [Python] FastAPI 비동기 처리(2) 비동기 데이터베이스 연결 구현하기

[Python] FastAPI 비동기 처리(2) 비동기 데이터베이스 연결 구현하기

---

 

안녕하세요.

이전 글에서는 FastAPI에서 비동기 데이터베이스 연결과 ORM을 활용한 CRUD 예제를 알아보았습니다.

이번 글에서는 FastAPI의 비동기 백그라운드 작업(Background Task) 기능과 실제 활용 예제에 대해 다뤄보겠습니다.

 

1. FastAPI의 BackgroundTasks

API 서버를 개발하다 보면, 이메일 발송, 대용량 파일 처리, 알림 전송 처럼, 응답과는 별개로 오래 걸리는 작업을 백그라운드에서 처리하고 싶을 때가 많습니다.

이럴 때 FastAPI의 BackgroundTask 기능을 활용하면 클라이언트에게 빠르게 응답을 반환하면서 필요한 작업을 비동기로 처리할 수 있습니다.

기본 사용법

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log_file:
        log_file.write(f"{message}\n")

@app.post("/send/")
async def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "알림 전송 완료")
    return {"message": "알림 요청이 접수되었습니다."}

• background_tasks.add_task(함수, 인자...) 형태로 등록하면, API 응답 후에 해당 함수가 비동기로 실행됩니다.

비동기 함수(Async)

FastAPI의 BackgroundTasks는 동기 함수뿐 아니라 async def로 정의한 비동기 함수도 지원합니다.

import asyncio

async def async_write_log(message: str):
    await asyncio.sleep(1)  # 예시: 실제로는 I/O 작업 등
    with open("log.txt", mode="a") as log_file:
        log_file.write(f"{message}\n")

@app.post("/send-async/")
async def send_async_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(async_write_log, "비동기 알림 전송 완료")
    return {"message": "비동기 알림 요청이 접수되었습니다."}

 

2. 실전 예제: FastAPI 백그라운드 이메일 인증 발송 예제

실제 서비스에서는 이메일 발송, 슬랙 메시지 전송 등 시간이 오래 걸리는 작업을 백그라운드로 처리할 수 있습니다.

아래는 회원가입 시 이메일 인증을 발송하는 상황에 대한 FastAPI 예제입니다.

 

참고:

• 실제 DB 대신 fake_users_db를 사용하여 예시를 구성했습니다.
• aiosmtplib :  비동기 SMTP 클라이언트 라이브러리
• pydantic : Python에서 데이터 유효성 검사와 설정을 쉽게 해주는 라이브러리, 요청(Request)이나 응답(Response) 데이터를 자동으로 검사하고 타입을 보장할 때 주로 사용
• uuid(Universally Unique Identifier) : 전 세계에서 중복되지 않는 고유한 값을 만들어주는 Python 표준 라이브러리, 주로 고유한 식별자(예: 인증 토큰, 사용자 ID 등)를 만들 때 사용

from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, EmailStr
import uuid
from email.message import EmailMessage
import aiosmtplib

app = FastAPI()

# (실습용) 메모리 내 임시 사용자 DB
fake_users_db = {}

# SMTP 서버 정보(Gmail 기준, 네이버 등은 해당 서비스에 맞게 변경)
SMTP_HOST = "smtp.gmail.com"
SMTP_PORT = 587
SMTP_USER = "your_gmail@gmail.com"      # 본인 Gmail 주소로 변경
SMTP_PASSWORD = "your_app_password"     # 앱 비밀번호로 변경

# 회원가입 요청에서 받을 데이터 모델 (이메일, 이름)
class UserCreate(BaseModel):
    email: EmailStr
    name: str

# 실제 이메일을 비동기로 발송하는 함수
async def send_verification_email(email: str, token: str):
    # 인증 링크 생성 (실제 서비스에서는 도메인에 맞게 수정)
    verification_link = f"http://localhost:8000/verify?token={token}"

    # 이메일 메시지 객체 생성
    message = EmailMessage()
    message["From"] = SMTP_USER       # 발신자
    message["To"] = email             # 수신자
    message["Subject"] = "이메일 인증 요청"  # 제목
    # 본문 작성
    message.set_content(
        f"""안녕하세요,
아래 링크를 클릭해 이메일 인증을 완료하세요.

인증 링크: {verification_link}
"""
    )

    # aiosmtplib로 비동기 SMTP 메일 발송
    await aiosmtplib.send(
        message,                       # 이메일 메시지 객체
        hostname=SMTP_HOST,            # SMTP 서버 주소
        port=SMTP_PORT,                # SMTP 포트
        username=SMTP_USER,            # 로그인 계정
        password=SMTP_PASSWORD,        # 앱 비밀번호
        start_tls=True,                # TLS 보안 연결 사용
    )

# 회원가입 API 엔드포인트
@app.post("/register/")
async def register_user(user: UserCreate, background_tasks: BackgroundTasks):
    # 이미 가입된 이메일인지 확인
    if user.email in fake_users_db:
        raise HTTPException(status_code=400, detail="이미 등록된 이메일입니다.")

    # 인증 토큰 생성 (uuid 사용)
    token = str(uuid.uuid4())

    # 사용자 정보를 메모리 DB에 저장 (실제 서비스는 DB 사용)
    fake_users_db[user.email] = {
        "name": user.name,
        "token": token,
        "verified": False
    }

    # 이메일 발송 작업을 백그라운드로 등록 (비동기 실행)
    background_tasks.add_task(send_verification_email, user.email, token)

    # 사용자에게 응답
    return {"message": "회원가입이 완료되었습니다. 이메일을 확인해 인증을 진행하세요."}

# 이메일 인증 링크 클릭 시 호출되는 엔드포인트
@app.get("/verify")
async def verify_email(token: str):
    # 모든 사용자 정보를 순회하며 토큰 일치 확인
    for email, info in fake_users_db.items():
        if info["token"] == token:
            info["verified"] = True    # 인증 완료 처리
            return {"message": f"{email}님의 이메일 인증이 완료되었습니다."}
    # 일치하는 토큰이 없으면 오류 반환
    raise HTTPException(status_code=400, detail="유효하지 않은 인증 토큰입니다.")