[Python 강의] 28강 - Flask를 이용한 비동기 작업 처리

안녕하세요! 그레이 해커 월횽입니다. 오늘은 Flask에서 비동기 작업 처리에 대해 알아보겠습니다. 비동기 처리는 시간이 오래 걸리는 작업을 요청과 분리하여 처리 속도를 높이고 서버의 응답성을 개선하는 데 매우 유용합니다. 이번 강의에서는 Flask에서 Celery를 사용하여 비동기 작업을 구현하는 방법을 다루어 보겠습니다.

 

 

1. 비동기 작업이 필요한 상황

비동기 작업이 필요한 대표적인 예로는 다음과 같은 경우가 있습니다.

· 이메일 전송
· 이미지나 파일 처리
· 대규모 데이터 처리
· 외부 API 호출

 

 

2. Celery 설치 및 Redis 설정

비동기 처리를 위해 Celery와 Redis(메시지 브로커 역할)를 설정하겠습니다.

반응형

Celery 및 Redis 설치

pip install celery redis

 

 

 

3. Flask와 Celery 연동

이제 Flask와 Celery를 연동하여 비동기 작업을 처리하는 기본 구조를 만들어 보겠습니다.

from flask import Flask, request, jsonify
from celery import Celery

app = Flask(__name__)

# Flask와 Celery 설정
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    return celery

celery = make_celery(app)

 

 

 

4. 비동기 작업 함수 정의

이제 Flask에서 비동기 작업을 정의해 보겠습니다. 예를 들어, 사용자가 요청한 데이터를 처리하는 작업을 비동기로 처리할 수 있습니다.

728x90

@celery.task
def background_task(data):
    # 데이터 처리 작업 (예: 파일 변환, 외부 API 호출)
    return {"status": "complete", "data": data}

 

 

 

5. 비동기 작업 호출

Flask 라우트에서 비동기 작업을 호출하는 방법입니다. delay 메서드를 사용하여 작업을 비동기로 실행합니다.

@app.route('/process', methods=['POST'])
def process_data():
    data = request.get_json()
    task = background_task.delay(data)  # 비동기 작업 호출
    return jsonify({"task_id": task.id, "status": "작업이 시작되었습니다."}), 202

 

 

 

6. 작업 상태 확인

작업이 완료되었는지 상태를 확인하는 방법입니다.

SMALL

@app.route('/task_status/<task_id>')
def task_status(task_id):
    task = background_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        response = {"status": "작업이 대기 중입니다."}
    elif task.state == 'SUCCESS':
        response = {"status": "작업이 완료되었습니다.", "result": task.result}
    else:
        response = {"status": "작업이 진행 중입니다."}
    return jsonify(response)

 

 

 

7. 이메일 전송 비동기 작업 예시

이제 비동기 작업을 활용하여 이메일을 전송하는 예제를 만들어 보겠습니다.

@celery.task
def send_email_async(email_data):
    # 이메일 전송 로직 (예: SMTP 또는 외부 서비스 API 사용)
    return "이메일이 성공적으로 전송되었습니다."

@app.route('/send_email', methods=['POST'])
def send_email():
    email_data = request.get_json()
    task = send_email_async.delay(email_data)
    return jsonify({"task_id": task.id, "status": "이메일 전송이 시작되었습니다."}), 202

 

 

8. Celery 워커 실행

비동기 작업을 실행하기 위해 Celery 워커를 실행해야 합니다. 다음 명령어로 Celery 워커를 실행합니다.

celery -A app.celery worker --loglevel=info

 

 

다음 시간에는 Flask와 비동기 작업을 활용한 실시간 알림 시스템 구축에 대해 알아보겠습니다!

 

 

 

 

 

- 이전 수업 목록