편집
1,408
번
바뀜
→기초 사용법
|-
|-
|task 등록
|task 등록
task 작성
|일을 등록한다.
|일을 등록한다.
위에서 app.autodiscover_tasks()를 사용했다면 각 앱의 tasks.py 안에 넣어주면 작동한다.
위에서 app.autodiscover_tasks()를 사용했다면 각 앱의 tasks.py 안에 넣어주면 작동한다.
(우측은 예시로, 자세한 작성은 아래 '뷰에서의 작업'을 참고하자)
데코레이터가 붙으면 이것이 task로 인식된다.
|사용하고자 하는 앱 안에서 tasks.py로 등록한다.<syntaxhighlight lang="python">
|사용하고자 하는 앱 안에서 tasks.py로 등록한다.<syntaxhighlight lang="python">
from __future__ import absolute_import, unicode_literals
from __future__ import absolute_import, unicode_literals
bind=True 옵션을 사용하면 task 인스턴스를 참조해 자신의 request를 쉽게 가져올 수 있다는데... 찾아볼 필요가 있겠다.
bind=True 옵션을 사용하면 task 인스턴스를 참조해 자신의 request를 쉽게 가져올 수 있다는데... 찾아볼 필요가 있겠다.
bind=True 옵션을 쓰면 무조건 첫번째 인수는 self로 받아야 한다.
|-
|-
|샐러리 실행
|샐러리 실행
|[데몬으로 시작하는 법은 따로 있나보다.]
|[데몬으로 시작하는 법은 따로 있나보다.]
|celery -A settings.py가있는폴더.celery worker --loglevel=info
|celery -A settings.py가있는폴더.celery worker --loglevel=info
|}
== 뷰에서의 작업 ==
{| class="wikitable"
!과정
!설명
!방법
|-
|모델 작성
|[https://levelup.gitconnected.com/django-celery-going-deeper-with-background-tasks-in-python-fa44958cf7cd 링크]의 코드를 참조하여 작성하였다.
|<syntaxhighlight lang="python">
from django.db import models
class 모델(models.Model):
STATUS_PENDING = 'PENDING'
STATUS_ERROR = 'ERROR'
STATUS_SUCCESS = 'SUCCESS'
STATUSES = ( # 상태를 기입하기 위한 사전작업.
(STATUS_PENDING, 'Pending'),
(STATUS_ERROR, 'Error'),
(STATUS_SUCCESS, 'Success'),
)
input = models.IntegerField()
output = models.IntegerField(blank=True, null=True)
created_at = models.DateTimeField(auto_now_add=True)
modified_at = models.DateTimeField(auto_now=True)
status = models.CharField(max_length=8, choices=STATUSES)
message = models.CharField(max_length=110, blank=True)
</syntaxhighlight>
|-
|뷰
|delay()는 celery의 업무를 수행하는 apply_async()의 간소판.
카운트다운이나 재실행 등의 자세한 옵션은 apply_async()에 더 자세한 변수를 넣음으로써 가능해진다.
|<syntaxhighlight lang="python">
from django.shortcuts import render, redirect
from django.views import View
from .models import 모델 # 모델을 활용한다면.
from .tasks import 필요한일
class CalculationView(View):
def get(self, request):
"""시작하는 템플릿."""
return render(request, 'app/start.html')
def post(self, request):
"""업무 수행."""
n = request.POST['폼에서받을값']
calculation = 모델.objects.create(status=Calculation.STATUS_PENDING, 기타 필요한 변수 기입)
필요한일.delay(calculation.id) # delay는 백그라운드에서 업무를 수행하라는 의미.
return redirect('fibonacci_list')
</syntaxhighlight>
|-
|task 작성
|
|<syntaxhighlight lang="python">
from celery_tutorial.celery import app # celery를 등록한 경로에서.
from .models import Calculation
@app.task(bind=True)
def 필요한일(self, calculation_id):
"""Perform a calculation & update the status"""
calculation = Calculation.objects.get(id=calculation_id)
try:
calculation.output = fib(calculation.input) # 함수를 위에서 정의하여 업무 실행.
calculation.status = Calculation.STATUS_SUCCESS # 업무가 실행되면 수행할 작업.
except Exception as e:
calculation.status = Calculation.STATUS_ERROR # 에러로 끝났음을 알린다.
calculation.message = str(e)[:110] # 에러메시지를 담는다.
calculation.save()
</syntaxhighlight>
|}
|}