TokyoAJ

도쿄아재

PYTHON 2025.07.11

[Cloudflare] FastAPI와 Cloudflare Images를 연동한 이미지 업로드 서비스 구축하기

프로젝트 개요

이 프로젝트는 FastAPI 프레임워크를 사용하여 Cloudflare Images API와 연동한 이미지 업로드 서비스를 구축하는 방법을 다룹니다. 클라이언트가 이미지를 업로드하면 Cloudflare Images에 저장하고, 저장된 이미지를 조회하거나 삭제할 수 있는 RESTful API를 제공합니다.


주요 기능

  1. 이미지 업로드: 클라이언트로부터 이미지 파일을 받아 Cloudflare Images에 저장
  2. 이미지 목록 조회: 저장된 모든 이미지 목록을 조회
  3. 이미지 삭제: 특정 이미지를 삭제
  4. 헬스체크: API 서버 상태 확인


기술 스택

  1. FastAPI: 현대적이고 빠른 Python 웹 프레임워크
  2. Cloudflare Images: 이미지 저장 및 최적화 서비스
  3. aiohttp: 비동기 HTTP 클라이언트 (파일 업로드용)
  4. httpx: 비동기 HTTP 클라이언트 (API 요청용)
  5. uvicorn: ASGI 서버


환경 설정

필요한 패키지 설치

pip install fastapi uvicorn python-dotenv httpx aiohttp

환경 변수 설정

.env 파일을 생성하여 Cloudflare 인증 정보를 설정합니다:

CLOUDFLARE_ACCOUNT_ID=your_account_id
CLOUDFLARE_API_TOKEN=your_api_token


애플리케이션 구조

1. FastAPI 애플리케이션 초기화

from fastapi import FastAPI, HTTPException, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

app = FastAPI(
title="FastAPI Cloudflare Images API",
description="FastAPI 애플리케이션과 Cloudflare Images를 연동한 이미지 업로드 서비스",
version="1.0.0"
)

FastAPI 애플리케이션을 생성할 때 메타데이터를 설정하여 Swagger UI에서 API 문서를 보기 좋게 표시할 수 있습니다.

2. CORS 설정

app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

개발 환경에서는 모든 도메인에서의 접근을 허용하지만, 프로덕션 환경에서는 특정 도메인만 허용하도록 수정해야 합니다.

3. Cloudflare 인증 설정

CLOUDFLARE_ACCOUNT_ID = os.getenv("CLOUDFLARE_ACCOUNT_ID")
CLOUDFLARE_API_TOKEN = os.getenv("CLOUDFLARE_API_TOKEN")

if not CLOUDFLARE_ACCOUNT_ID or not CLOUDFLARE_API_TOKEN:
raise ValueError("CLOUDFLARE_ACCOUNT_ID와 CLOUDFLARE_API_TOKEN을 설정해주세요.")

환경 변수에서 Cloudflare 인증 정보를 로드하고, 필수 값이 없으면 에러를 발생시킵니다.


API 엔드포인트 구현

1. 헬스체크 엔드포인트

@app.get("/")
async def root():
return {
"message": "FastAPI Cloudflare Images API",
"version": "1.0.0",
"status": "running"
}

@app.get("/health")
async def health_check():
return {"status": "healthy"}

기본적인 서버 상태 확인을 위한 엔드포인트입니다.

2. 이미지 업로드 엔드포인트

@[app.post](<http://app.post>)("/upload-image")
async def upload_image(file: UploadFile = File(...)):
# 파일 형식 검증
if not file.content_type.startswith('image/'):
raise HTTPException(
status_code=400,
detail="이미지 파일만 업로드 가능합니다."
)
# 파일 내용 읽기
file_content = await [file.read](<http://file.read>)()
# Cloudflare API 설정
url = f"<https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/images/v1>"
headers = get_cloudflare_headers(is_upload=True)
# multipart/form-data 형식으로 파일 업로드
form = aiohttp.FormData()
form.add_field('file', file_content,
filename=file.filename,
content_type=file.content_type)
# Cloudflare API 호출
async with aiohttp.ClientSession() as session:
async with [session.post](<http://session.post>)(url, headers=headers, data=form, timeout=30) as response:
response_text = await response.text()
response_status = response.status

이미지 업로드 기능의 핵심은 다음과 같습니다:

  1. 파일 형식 검증: 업로드된 파일이 이미지 형식인지 확인
  2. 파일 내용 읽기: 업로드된 파일의 바이너리 데이터 읽기
  3. multipart/form-data 구성: Cloudflare API가 요구하는 형식으로 데이터 구성
  4. 비동기 HTTP 요청: aiohttp를 사용하여 파일 업로드 수행

3. 이미지 목록 조회 엔드포인트

@app.get("/images")
async def list_images():
url = f"<https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/images/v1>"
headers = get_cloudflare_headers()
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url, headers=headers)
if response.status_code == 200:
result = response.json()
if result.get("success"):
return {
"success": True,
"images": result["result"]["images"],
"total_count": len(result["result"]["images"])
}

저장된 이미지 목록을 조회하는 간단한 GET 요청입니다. httpx를 사용하여 비동기 HTTP 클라이언트를 구현합니다.

4. 이미지 삭제 엔드포인트

@app.delete("/images/{image_id}")
async def delete_image(image_id: str):
url = f"<https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/images/v1/{image_id}>"
headers = get_cloudflare_headers()
async with httpx.AsyncClient() as client:
response = await client.delete(url, headers=headers)
if response.status_code == 200:
result = response.json()
if result.get("success"):
return {
"success": True,
"message": f"이미지 {image_id}가 성공적으로 삭제되었습니다."
}

경로 매개변수를 사용하여 특정 이미지 ID에 대한 삭제 요청을 처리합니다.

유틸리티 함수

Cloudflare API 헤더 생성

def get_cloudflare_headers(is_upload=False):
headers = {
"Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}"
}
if not is_upload:
headers["Content-Type"] = "application/json"
return headers

파일 업로드와 일반 API 요청에 따라 적절한 헤더를 생성하는 유틸리티 함수입니다.


에러 처리

코드에서는 다양한 에러 상황을 고려하여 적절한 예외 처리를 구현합니다:

  1. 설정 오류: Cloudflare 인증 정보 누락
  2. 파일 형식 오류: 이미지가 아닌 파일 업로드 시도
  3. 네트워크 오류: API 요청 타임아웃
  4. API 오류: Cloudflare API 응답 에러
try:
# API 호출 로직
pass
except httpx.TimeoutException:
raise HTTPException(status_code=500, detail="API 요청 시간 초과")
except Exception as e:
raise HTTPException(status_code=500, detail=f"오류가 발생했습니다: {str(e)}")

서버 실행

if __name__ == "__main__":
[uvicorn.run](<http://uvicorn.run>)(
"main:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info"
)

개발 환경에서는 reload=True 옵션을 사용하여 코드 변경 시 자동으로 서버를 재시작합니다.


보안 고려사항

  1. 환경 변수 관리: API 키와 같은 민감한 정보는 환경 변수로 관리
  2. CORS 설정: 프로덕션 환경에서는 특정 도메인만 허용
  3. 파일 형식 검증: 업로드되는 파일의 형식을 반드시 검증
  4. 파일 크기 제한: 대용량 파일 업로드 방지를 위한 크기 제한 필요


실행 방법

# 가상환경 생성 및 활성화
python -m venv venv
source venv/bin/activate # Windows: venvScriptsactivate

# 의존성 설치
pip install -r requirements.txt

# 환경 변수 설정
cp env.example .env
# .env 파일에 실제 Cloudflare 인증 정보 입력

# 서버 실행
python [main.py](<http://main.py>)

서버 실행 후 http://localhost:8000/docs에서 Swagger UI를 통해 API 문서를 확인할 수 있습니다.


마무리

이 프로젝트는 FastAPI의 강력한 기능과 Cloudflare Images의 이미지 최적화 기능을 결합하여 효율적인 이미지 업로드 서비스를 구축하는 방법을 보여줍니다. 비동기 처리를 통해 높은 성능을 제공하며, 적절한 에러 처리와 보안 고려사항을 포함하여 실제 서비스에 적용할 수 있는 수준의 코드를 작성했습니다.

추가로 고려할 수 있는 개선사항으로는 파일 크기 제한, 이미지 리사이징, 사용자 인증, 로깅 시스템 등이 있습니다.

댓글