Как правильно переопределить функцию декоратора в Luigi для AWS Batch? - PullRequest
Новая
       36

Как правильно переопределить функцию декоратора в Luigi для AWS Batch?

0 голосов
/ 02 марта 2020

Я использую Луиджи, чтобы запустить AWS Пакетное задание. Я хочу создать подкласс luigi.contrib.batch.BatchTask (документацию для Luigi для AWS Batch (зависит от boto3) можно найти здесь: https://luigi.readthedocs.io/en/stable/_modules/luigi/contrib/batch.html#BatchTask)

Класс BatchTask: 

class BatchTask(luigi.Task):

    """
    Base class for an Amazon Batch job

    Amazon Batch requires you to register "job definitions", which are JSON
    descriptions for how to issue the ``docker run`` command. This Luigi Task
    requires a pre-registered Batch jobDefinition name passed as a Parameter

    :param job_definition (str): name of pre-registered jobDefinition
    :param job_name: name of specific job, for tracking in the queue and logs.
    :param job_queue: name of job queue where job is going to be submitted.

    """
    job_definition = luigi.Parameter()
    job_name = luigi.OptionalParameter(default=None)
    job_queue = luigi.OptionalParameter(default=None)
    poll_time = luigi.IntParameter(default=POLL_TIME)

    def run(self):
        bc = BatchClient(self.poll_time)
        job_id = bc.submit_job(
            self.job_definition,
            self.parameters,
            job_name=self.job_name,
            queue=self.job_queue)
        bc.wait_on_job(job_id)


    @property
    def parameters(self):
        """Override to return a dict of parameters for the Batch Task"""
        return {}

С моим подклассом я хочу переопределить функцию параметров, чтобы я мог проходить через свои собственные параметры с помощью кода, а не полагаться на веб-интерфейс AWS для заполнения параметров вручную. Эта функция параметров имеет тег @property над функцией, который показывает мне, что это декоратор - что-то, с чем я не очень хорошо знаком.

Я думаю, что мой вопрос - это вопрос агностии c выше, хотя и :

Как переопределить функцию параметров, чтобы она не возвращала пустой словарь, а определенный мной словарь?

Письменный подкласс: 

import luigi.contrib.batch as batch

batch_job_revision_number=1

class SubclassLuigiBatchTask(batch.BatchTask):

    job_definition='arn:aws:batch:{0}:job-definition/{1}:{2}'.format(
        'aws-credentials-that-i-cannot-share',
        'aws_pre-registered-job-description-name',
        batch_job_revision_number)
    job_name='my_example_job'
    job_queue='my_example_queue'
    poll_time = 10

    task = batch.BatchTask(
        job_definition='arn:aws:batch:{0}:job-definition/{1}:{2}'.format(
            'aws-credentials-that-i-cannot-share',
            'aws_pre-registered-job_description',
            batch_job_revision_number),
        job_name='my_example_job',
        job_queue='my_example_queue',
        poll_time=10
    )

    @task.parameters
    def parameters(self):
        return {
            "job_definition": "df -h"
        }

    # This run function runs the predefined job (job definition)
    def run(self):
        self.task.run()

    # function only runs if output doesn't exist yet
    def output(self):
        return LocalTarget('/home/some_user/task_completed.txt')

Приведенный выше пример возвращает ошибку: 

TypeError: 'dict' object is not callable

Как получить доступ к функции сеттера для переопределения этого значения по умолчанию return {}? Я пытался в командной строке, чтобы увидеть, какие функции доступны для task.parameters, чтобы увидеть, какие методы были доступны: 

>>> dir(task.parameters)
['__class__', '__contains__', '__delattr__', '__delitem__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__getitem__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__iter__', '__le__', '__len__', '__lt__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__setitem__', '__sizeof__', '__str__', '__subclasshook__', 'clear', 'copy', 'fromkeys', 'get', 'items', 'keys', 'pop', 'popitem', 'setdefault', 'update', 'values']

1 Ответ

0 голосов
/ 04 марта 2020

Вместо 

@task.parameters
def parameters(self):
    ...

вам нужно сделать: 

@property
def parameters(self):
    ...

Переопределение свойств с помощью декораторов не так уж плохо. Просто взгляните на следующий пример: 

class A:
  def __init__(self):
    self._a = "3"

  @property
  def a(self):
    return self._a

class B(A):
  def __init__(self):
    super(B, self).__init__()
    self._b = "4"

  @property
  def a(self):
    return self._b

a = A()  
b = B()

print('a', a.a)
print('b', b.a)

Это распечатает: 

a 3
b 4

Я думаю, что добавление этого странного декоратора @task.parameters может быть тем, что все испортило. Может быть, на самом деле пытаться исправить себя? Очень странные вещи.

...