Django - Celery로 비동기 Async Task 구성하기

앞서 진행한 Celery 구성을 통해 간단한 비동기 Task를 예제를 통해 사용해 보고자 한다.

기본적인 Celery를 구성하는 부분은 아래 내용을 참고하기 바란다.

Django - Celery 구성하기 (asecurity.dev)

Django - Celery 구성하기

 

그럼 위 기본 설정이후 Task를 구성해보도록 하겠다.

여기에서는 추가로 생성한 blog 앱에 Task를 구성할 예정이며, 이를 위해 기존 blog/tasks.py 내용을 아래와 같이 수정하였다.

작업은 테스트를 위해서 httping.org로 지연 요청을 보내는 함수 send_httpbin 하나를 만들고 이 작업을 확인하는 get_task_state 를 하나 생성하였다.

blog/tasks.py

import requests
from celery import shared_task
from celery.result import AsyncResult
import json

@shared_task()
def send_httpbin(count):
	for _ in range(0, count):
	    requests.get(f'https://httpbin.org/delay/4')

def get_task_state(taskid):
    if taskid:
        task: AsyncResult = AsyncResult(taskid)

        task_result = {
            'state': task.state,
            'result': task.result
        }
    else:
        task_result = {
            'state': 'error',
            'result': 'taskid empty'
        }
    
    return task_result

이렇게 만든 함수를 blog/views.py 를 이용해서 URL 호출시 동작하도록 아래와 같이 구성하였다.

blog/views.py 

import requests
from .tasks import send_httpbin, get_task_state
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
# Create your views here.

@csrf_exempt
def send(request):
    count = request.POST.get('count')
    task_list = []
    for value in range(0, int(count)):
        c_task = send_httpbin.delay(value)
        task_list.append(c_task.id)
    return JsonResponse({'taskid': task_list})


def status(request):
    taskid = request.GET.get('taskid')
    response = get_task_state(taskid)
    return JsonResponse(response)

위 내용을 보면, send라는 경로에 send_httpbin 를 이용해서 Celery를 Task를 만들고, Task ID를 반환하는 방식이다.

이후 status를 통해 결과를 확인하는 구조라고 할 수 있다. 

csrf_exempt를 사용하는 이유는 로컬에서 테스트시 발생하는 CORS 오류를 회피하기 위해 지정한 것이기 때문에 실제 호출 하는 위치가 다른 경우 이를 사용하지 않아도 된다. 자세한 내용은 아래를 참고 하기 바란다.

 

Python - FastApi Cross-Origin Resource Sharing (CORS) 이해/해결

마지막으로 기본 프로젝트 파일에 URL을 등록하도록하자.

base/urls.py

"""base URL Configuration

The `urlpatterns` list routes URLs to views. For more information please see:
    https://docs.djangoproject.com/en/4.1/topics/http/urls/
Examples:
Function views
    1. Add an import:  from my_app import views
    2. Add a URL to urlpatterns:  path('', views.home, name='home')
Class-based views
    1. Add an import:  from other_app.views import Home
    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
Including another URLconf
    1. Import the include() function: from django.urls import include, path
    2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
"""
from django.contrib import admin
from django.urls import path
import blog.views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('send', blog.views.send),
    path('status', blog.views.status)
]

 

 

그럼 간단히 테스트를 POSTMan을 이용해서 진행해보자.

아래와 같이 Body에 count 갯수를 지정하여 호출을 진행해보도록 하자.

실행 결과

{
    "taskid": [
        "0e3cdb8d-9899-454a-a1b1-d2825f2a3922",
        "64a15794-4aec-4335-bbf4-a6402a97175e",
        "4cf08e2d-6c91-4d56-ba39-dd032dbae15c",
        "5568532b-6ae8-4196-be99-1b7282fc982b",
        "4eaae5c4-0de0-4ed9-86f9-9e0760f68a44"
    ]
}

정상적으로 5개의 작업이 비동기로 처리된 것을 알 수 있다.

이제 저 Task Id를 이용해서 각각의 작업 상태를 확인 할 수 있다.

 

마치며

Celery를 통해 Django에서 비동기 작업을 관리할 수 있는 방안을 간단히 확인해 보았다.

Django와 비동기 구현시 복잡한 작업이 필요하지 않고 쉽게 사용할 수 있으므로, 비동기 작업을 개발한다면 적극적으로 추천한다.