홈>2020-02-01 00:38

amazon web services - django 함수를 주기적으로 호출

웹 개발에 대한 신인 질문입니다. 개발자가 django 응용 프로그램에서 전자 메일을 보내는 기능을 호출하고 실행하여 수동으로 재정의하고 수동으로 보낼 수있을뿐만 아니라 시간을 활성화하여 정해진 시간에 주기적으로 전자 메일을 보내는 기능을 호출하고 실행하는 방법을 찾고 있습니다 주.

장고에서 주기적으로 기능을 실행하는 셀러리 사용을 제안하는 답변을 찾았습니다

그러나 애플리케이션을 재배치하지 않고 기간을 변경하고 싶습니다. AWS 도구에 대한 몇 가지 연구를 수행했으며 AWS 게이트웨이와 AWS Lambda 및 AWS Cloudwatch의 조합으로 웹 앱에 URL/엔드 포인트를 보내거나 기능을 활성화하기 위해 요청을 얻습니다.

현재 아래와 같은 것이 있습니다.

views.py

def api_send_email(request):
    #insert send email function
    print ("sending")
    return redirect('/')

urls.py

urlpatterns = [
    url(r'^send_email$', api_send_email, name="api_send_email"),
]

따라서 위의 URL은 https // xxx/send_email url로 이동하거나 aws에서 주기적으로 해당 URL에 get 요청을 보내서 수동으로 트리거 할 수 있습니다. 대신 게시 요청을 수행하는 것이 더 안전 할 것이라고 생각했지만 내 앱 자체에 csrf 토큰이 필요하기 때문에 AWS 도구가 그렇게 할 수 있는지 확실하지 않습니다.

이 작업을 수행하는 가장 좋은 방법에 대한 제안은 환영합니다.

감사합니다

django amazon-web-services

답변 # 1

셀러리로도 가능하다고 생각합니다. 이를 위해 주기적 작업을 추가 할 수 있습니다. 5 분마다 시작되는 주기적 작업이 있다고 가정하겠습니다.

그런 다음 모델에 로직을 설정하여 당시 이메일을 보내야하는지 결정할 수 있습니다. 예를 들면 다음과 같습니다.

class YourTaskConfig(models.Model):
    SEND_CHOICES = (
        ('minute': 'minute'),
        ('hour': 'hour'),
        ('day': 'day'),
    )
    send_every = models.CharField(max_length=25, choices=SEND_CHOICES)
    interval_amount = models.IntegerField()
    last_executed = models.DateTimeField(auto_add_now=True)
    is_active = models.BooleanField(default=True)
    def should_run(self):
       now = timezone.now()
       if self.send_every == 'minute':
           td = datetime.timedelta(seconds=self.interval_amount*60)
       elif self.send_every == 'day':
           td = datetime.timedelta(days=self.interval_amount)
       ... # rest of the logic on time delta
       if now - self.last_executed >= td:
           self.save()  # Updates current execution time
           return True
       return False

이메일모델은이 구성에 대해 FK를 가질 수 있습니다 (있는 경우).

class Email(models):
    config = models.ForeignKey(YourTaskConfig, on_delete=models.DO_NOTHING)

정기적 인 작업을 사용하십시오 :

from celery.task.schedules import crontab
from celery.decorators import periodic_task

@periodic_task(run_every=(crontab(minute='*/5')), name="some_task", ignore_result=True)  # Runs every 5 minute
def some_task():
    for i in YourTaskConfig.objects.filter(is_active=True):  # run only active tasks
        should_send_email = i.should_run()
        if should_send_email:
            i.email_set.all()  # Here you go, you have your emails which you want send

참고 : 테스트되지 않은 코드이지만이 솔루션의 일반적인 아이디어를 얻을 수 있습니다. 도움이 되길 바랍니다 !!

Python2.net

amazon web services - django 함수를 주기적으로 호출

관련 자료