Olá Habr! Vou lhe contar a história do meu esgotamento profissional.

Aconteceu que eu odeio esteiras rotineiras. Tenho vários projetos usando o aipo . Cada vez que uma tarefa se torna mais complicada do que 2 + 2 = 5 , o modelo de solução é reduzido à criação de uma classe que executa a tarefa e a uma função inicial com a qual o Celery pode trabalhar - um clichê. Neste artigo, vou contar como lutei com um clichê e o que veio dele.

Ponto de partida

Considere a tarefa comum do aipo. Há uma classe que executa a tarefa e uma função inicial que instancia a classe e inicia um de seus métodos, nos quais toda a lógica da tarefa é implementada e o tratamento de erros é herdado:

 class MyTask( FirstMixin, SecondMixin, ThirdMixin, ): def main(self): data = self.do_something() response = self.remote_call(data) parsed = self.parser(response) return self.process(parsed) @app.task(bind=True) def my_task(self, arg1, arg2): instance = MyTask( celery_task=self, arg1=arg1, arg2=arg2, ) return instance.full_task() 

Ao mesmo tempo, o método full_task inclui uma chamada para main , no entanto, também lida com tratamento de erros, log e outras bobagens que não estão diretamente relacionadas à tarefa principal.

Ideia da classe de tarefas

Na raiz da classe de tarefas, encontra-se uma idéia simples: na classe base, você pode definir um método da classe de task , implementar o comportamento da função inicial nela e herdar:

 class BaseTask: def __init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) def full_task(self): try: return self.main() except: self.celery_task.retry(countdown=30) @classmethod def task(cls, task, **kwargs): self = cls( celery_task=celery_task, **kwargs, ) return self.full_task() 

Todo o tédio auxiliar coletado na classe base. Nós não vamos voltar para ela novamente. Percebemos a lógica da tarefa:

 @app.taskcls(bind=True) class MyTask( BaseTask, FirstMixin, SecondMixin, ThirdMixin, ): def main(self): data = self.do_something() response = self.remote_call(data) parsed = self.parser(response) return self.process(parsed) 

Sem mais casca, muito melhor. No entanto, e o ponto de entrada?

 MyTask.task.delay(...) 

MyTask.task possui todos os métodos de tarefas usuais: delay , apply_async e, de um modo geral, é.

Agora os argumentos do decorador. É especialmente divertido arrastar bind = True para cada tarefa. É possível passar argumentos padrão por meio de uma classe base?

 class BaseTask: class MetaTask: bind = True def __init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) def full_task(self): try: return self.main() except: self.celery_task.retry(countdown=30) @classmethod def task(cls, task, **kwargs): self = cls( celery_task=celery_task, **kwargs, ) return self.full_task() 

A classe MetaTask aninhada contém argumentos padrão e estará disponível para todas as classes filho. Curiosamente, também pode ser herdado:

 class BaseHasTimeout(BaseTask): class MetaTask(BaseTask.MetaTask): timeout = 42 

Os argumentos passados ​​para o decorador @app.taskcls têm a maior prioridade:

 @app.taskcls(timeout=20) class MyTask( BaseHasTimeout, FirstMixin, SecondMixin, ThirdMixin, ): def main(self): ... 

Como resultado, o tempo limite da tarefa será 20.

Indo além

Em aplicativos da Web, geralmente é necessário iniciar uma tarefa da visualização. No caso de alta adesão, visão e tarefa podem ser combinadas:

 class BaseViewTask: @classmethod def task(cls, **kwargs): # Somehow init View class manually self = cls(...) return self.celery() @app.taskcls class MyView( BaseViewTask, FirstMixin, SecondMixin, ThirdMixin, APIView, ): queryset = MyModel.objects.all() def get_some_data(self, *args, **kwargs): # common methed return self.queryset.filtert(...) def get(self, request): data = self.get_some_data(request.field) # used in request handling return Response(json.dumps(data)) def post(self, request): self.task.delay(...) return Response(status=201) def celery(self): data = self.get_some_data(...) # also used in background task return self.do_something(data) 

A propósito, para evitar a colisão de nomes, a classe aninhada é chamada MetaTask , não Meta , como no django.

Conclusão

Essa funcionalidade é esperada no Aipo 4.5 . No entanto, também preparei um pacote que permite testar o decorador de tarefas hoje. A idéia do pacote é que, ao atualizar o Celery para a versão 4.5, você possa remover sua importação sem alterar uma única linha de código.