(self, message, on_ack=noop,
hostname=None, eventer=None, app=None,
connection_errors=None, request_dict=None,
task=None, on_reject=noop, body=None,
headers=None, decoded=False, utc=True,
maybe_make_aware=maybe_make_aware,
maybe_iso8601=maybe_iso8601, **opts)
| 88 | ) |
| 89 | |
| 90 | def __init__(self, message, on_ack=noop, |
| 91 | hostname=None, eventer=None, app=None, |
| 92 | connection_errors=None, request_dict=None, |
| 93 | task=None, on_reject=noop, body=None, |
| 94 | headers=None, decoded=False, utc=True, |
| 95 | maybe_make_aware=maybe_make_aware, |
| 96 | maybe_iso8601=maybe_iso8601, **opts): |
| 97 | self._message = message |
| 98 | self._request_dict = (message.headers.copy() if headers is None |
| 99 | else headers.copy()) |
| 100 | self._body = message.body if body is None else body |
| 101 | self._app = app |
| 102 | self._utc = utc |
| 103 | self._decoded = decoded |
| 104 | if decoded: |
| 105 | self._content_type = self._content_encoding = None |
| 106 | else: |
| 107 | self._content_type, self._content_encoding = ( |
| 108 | message.content_type, message.content_encoding, |
| 109 | ) |
| 110 | self.__payload = self._body if self._decoded else message.payload |
| 111 | self.id = self._request_dict['id'] |
| 112 | self._type = self.name = self._request_dict['task'] |
| 113 | if 'shadow' in self._request_dict: |
| 114 | self.name = self._request_dict['shadow'] or self.name |
| 115 | self._root_id = self._request_dict.get('root_id') |
| 116 | self._parent_id = self._request_dict.get('parent_id') |
| 117 | timelimit = self._request_dict.get('timelimit', None) |
| 118 | if timelimit: |
| 119 | self.time_limits = timelimit |
| 120 | self._argsrepr = self._request_dict.get('argsrepr', '') |
| 121 | self._kwargsrepr = self._request_dict.get('kwargsrepr', '') |
| 122 | self._on_ack = on_ack |
| 123 | self._on_reject = on_reject |
| 124 | self._hostname = hostname or gethostname() |
| 125 | self._eventer = eventer |
| 126 | self._connection_errors = connection_errors or () |
| 127 | self._task = task or self._app.tasks[self._type] |
| 128 | ignore_result = self._request_dict.get('ignore_result', None) |
| 129 | if ignore_result is None: |
| 130 | ignore_result = self._task.ignore_result |
| 131 | self._ignore_result = ignore_result |
| 132 | |
| 133 | # timezone means the message is timezone-aware, and the only timezone |
| 134 | # supported at this point is UTC. |
| 135 | eta = self._request_dict.get('eta') |
| 136 | if eta is not None: |
| 137 | try: |
| 138 | eta = maybe_iso8601(eta) |
| 139 | except (AttributeError, ValueError, TypeError) as exc: |
| 140 | raise InvalidTaskError( |
| 141 | f'invalid ETA value {eta!r}: {exc}') |
| 142 | self._eta = maybe_make_aware(eta, self.tzlocal) |
| 143 | else: |
| 144 | self._eta = None |
| 145 | |
| 146 | expires = self._request_dict.get('expires') |
| 147 | if expires is not None: |
nothing calls this directly
no test coverage detected