We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Celerybeat Cleanup Error
```
worker_1         | Traceback (most recent call last):
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/celery/beat.py", line 320, in apply_async
worker_1         |     **entry.options)
worker_1         |   File "/opt/warehouse/src/warehouse/tasks.py", line 95, in apply_async
worker_1         |     return super().apply_async(*args, **kwargs)
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/celery/app/task.py", line 518, in apply_async
worker_1         |     check_arguments(*(args or ()), **(kwargs or {}))
worker_1         | TypeError: backend_cleanup() takes 0 positional arguments but 1 was given
worker_1         |
worker_1         | During handling of the above exception, another exception occurred:
worker_1         |
worker_1         | Traceback (most recent call last):
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/celery/beat.py", line 222, in apply_entry
worker_1         |     result = self.apply_async(entry, producer=producer, advance=False)
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/celery/beat.py", line 328, in apply_async
worker_1         |     entry, exc=exc)), sys.exc_info()[2])
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/vine/five.py", line 178, in reraise
worker_1         |     raise value.with_traceback(tb)
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/celery/beat.py", line 320, in apply_async
worker_1         |     **entry.options)
worker_1         |   File "/opt/warehouse/src/warehouse/tasks.py", line 95, in apply_async
worker_1         |     return super().apply_async(*args, **kwargs)
worker_1         |   File "/opt/warehouse/lib/python3.6/site-packages/celery/app/task.py", line 518, in apply_async
worker_1         |     check_arguments(*(args or ()), **(kwargs or {}))
worker_1         | celery.beat.SchedulingError: Couldn't apply scheduled task celery.backend_cleanup: backend_cleanup() takes 0 positional arguments but 1 was given
```

I assume this is an integration issue with how the warehouse <-> celery integration happens. Not sure what it's actual affect is though.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `warehouse/tasks.py`
Content:
```
1 # Licensed under the Apache License, Version 2.0 (the "License");
2 # you may not use this file except in compliance with the License.
3 # You may obtain a copy of the License at
4 #
5 # http://www.apache.org/licenses/LICENSE-2.0
6 #
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10 # See the License for the specific language governing permissions and
11 # limitations under the License.
12 
13 import functools
14 import logging
15 import urllib.parse
16 
17 import celery
18 import celery.app.backends
19 import celery.backends.redis
20 import pyramid.scripting
21 import pyramid_retry
22 import transaction
23 import venusian
24 
25 from kombu import Queue
26 from pyramid.threadlocal import get_current_request
27 
28 from warehouse.config import Environment
29 from warehouse.metrics import IMetricsService
30 
31 # We need to trick Celery into supporting rediss:// URLs which is how redis-py
32 # signals that you should use Redis with TLS.
33 celery.app.backends.BACKEND_ALIASES[
34     "rediss"
35 ] = "warehouse.tasks:TLSRedisBackend"  # noqa
36 
37 
38 # We need to register that the sqs:// url scheme uses a netloc
39 urllib.parse.uses_netloc.append("sqs")
40 
41 logger = logging.getLogger(__name__)
42 
43 
44 class TLSRedisBackend(celery.backends.redis.RedisBackend):
45     def _params_from_url(self, url, defaults):
46         params = super()._params_from_url(url, defaults)
47         params.update({"connection_class": self.redis.SSLConnection})
48         return params
49 
50 
51 class WarehouseTask(celery.Task):
52     def __new__(cls, *args, **kwargs):
53         obj = super().__new__(cls, *args, **kwargs)
54         if getattr(obj, "__header__", None) is not None:
55             obj.__header__ = functools.partial(obj.__header__, object())
56 
57         # We do this here instead of inside of __call__ so that exceptions
58         # coming from the transaction manager get caught by the autoretry
59         # mechanism.
60         @functools.wraps(obj.run)
61         def run(*args, **kwargs):
62             original_run = obj._wh_original_run
63             request = obj.get_request()
64             metrics = request.find_service(IMetricsService, context=None)
65             metric_tags = [f"task:{obj.name}"]
66 
67             with request.tm, metrics.timed("warehouse.task.run", tags=metric_tags):
68                 try:
69                     result = original_run(*args, **kwargs)
70                     metrics.increment("warehouse.task.complete", tags=metric_tags)
71                     return result
72                 except BaseException as exc:
73                     if isinstance(
74                         exc, pyramid_retry.RetryableException
75                     ) or pyramid_retry.IRetryableError.providedBy(exc):
76                         metrics.increment("warehouse.task.retried", tags=metric_tags)
77                         raise obj.retry(exc=exc)
78                     metrics.increment("warehouse.task.failed", tags=metric_tags)
79                     raise
80 
81         obj._wh_original_run, obj.run = obj.run, run
82 
83         return obj
84 
85     def __call__(self, *args, **kwargs):
86         return super().__call__(*(self.get_request(),) + args, **kwargs)
87 
88     def get_request(self):
89         if not hasattr(self.request, "pyramid_env"):
90             registry = self.app.pyramid_config.registry
91             env = pyramid.scripting.prepare(registry=registry)
92             env["request"].tm = transaction.TransactionManager(explicit=True)
93             self.request.update(pyramid_env=env)
94 
95         return self.request.pyramid_env["request"]
96 
97     def after_return(self, status, retval, task_id, args, kwargs, einfo):
98         if hasattr(self.request, "pyramid_env"):
99             pyramid_env = self.request.pyramid_env
100             pyramid_env["request"]._process_finished_callbacks()
101             pyramid_env["closer"]()
102 
103     def apply_async(self, *args, **kwargs):
104         # The API design of Celery makes this threadlocal pretty impossible to
105         # avoid :(
106         request = get_current_request()
107 
108         # If for whatever reason we were unable to get a request we'll just
109         # skip this and call the original method to send this immediately.
110         if request is None or not hasattr(request, "tm"):
111             return super().apply_async(*args, **kwargs)
112 
113         # This will break things that expect to get an AsyncResult because
114         # we're no longer going to be returning an async result from this when
115         # called from within a request, response cycle. Ideally we shouldn't be
116         # waiting for responses in a request/response cycle anyways though.
117         request.tm.get().addAfterCommitHook(
118             self._after_commit_hook, args=args, kws=kwargs
119         )
120 
121     def _after_commit_hook(self, success, *args, **kwargs):
122         if success:
123             super().apply_async(*args, **kwargs)
124 
125 
126 def task(**kwargs):
127     kwargs.setdefault("shared", False)
128 
129     def deco(wrapped):
130         def callback(scanner, name, wrapped):
131             celery_app = scanner.config.registry["celery.app"]
132             celery_app.task(**kwargs)(wrapped)
133 
134         venusian.attach(wrapped, callback)
135 
136         return wrapped
137 
138     return deco
139 
140 
141 def _get_task(celery_app, task_func):
142     task_name = celery_app.gen_task_name(task_func.__name__, task_func.__module__)
143     return celery_app.tasks[task_name]
144 
145 
146 def _get_task_from_request(request):
147     celery_app = request.registry["celery.app"]
148     return functools.partial(_get_task, celery_app)
149 
150 
151 def _get_task_from_config(config, task):
152     celery_app = config.registry["celery.app"]
153     return _get_task(celery_app, task)
154 
155 
156 def _get_celery_app(config):
157     return config.registry["celery.app"]
158 
159 
160 def _add_periodic_task(config, schedule, func, args=(), kwargs=(), name=None, **opts):
161     def add_task():
162         config.registry["celery.app"].add_periodic_task(
163             schedule, config.task(func).s(), args=args, kwargs=kwargs, name=name, **opts
164         )
165 
166     config.action(None, add_task, order=100)
167 
168 
169 def includeme(config):
170     s = config.registry.settings
171 
172     broker_transport_options = {}
173 
174     broker_url = s["celery.broker_url"]
175     if broker_url.startswith("sqs://"):
176         parsed_url = urllib.parse.urlparse(broker_url)
177         parsed_query = urllib.parse.parse_qs(parsed_url.query)
178         # Celery doesn't handle paths/query arms being passed into the SQS broker,
179         # so we'll just remove them from here.
180         broker_url = urllib.parse.urlunparse(parsed_url[:2] + ("", "", "", ""))
181 
182         if "queue_name_prefix" in parsed_query:
183             broker_transport_options["queue_name_prefix"] = (
184                 parsed_query["queue_name_prefix"][0] + "-"
185             )
186 
187         if "region" in parsed_query:
188             broker_transport_options["region"] = parsed_query["region"][0]
189 
190     config.registry["celery.app"] = celery.Celery(
191         "warehouse", autofinalize=False, set_as_current=False
192     )
193     config.registry["celery.app"].conf.update(
194         accept_content=["json", "msgpack"],
195         broker_url=broker_url,
196         broker_use_ssl=s["warehouse.env"] == Environment.production,
197         broker_transport_options=broker_transport_options,
198         task_default_queue="default",
199         task_default_routing_key="task.default",
200         task_queue_ha_policy="all",
201         task_queues=(
202             Queue("default", routing_key="task.#"),
203             Queue("malware", routing_key="malware.#"),
204         ),
205         task_routes={"warehouse.malware.tasks.*": {"queue": "malware"}},
206         task_serializer="json",
207         worker_disable_rate_limits=True,
208         REDBEAT_REDIS_URL=s["celery.scheduler_url"],
209     )
210     config.registry["celery.app"].Task = WarehouseTask
211     config.registry["celery.app"].pyramid_config = config
212 
213     config.action(("celery", "finalize"), config.registry["celery.app"].finalize)
214 
215     config.add_directive("add_periodic_task", _add_periodic_task, action_wrap=False)
216     config.add_directive("make_celery_app", _get_celery_app, action_wrap=False)
217     config.add_directive("task", _get_task_from_config, action_wrap=False)
218     config.add_request_method(_get_task_from_request, name="task", reify=True)
219
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/warehouse/tasks.py b/warehouse/tasks.py
--- a/warehouse/tasks.py
+++ b/warehouse/tasks.py
@@ -51,8 +51,6 @@
 class WarehouseTask(celery.Task):
     def __new__(cls, *args, **kwargs):
         obj = super().__new__(cls, *args, **kwargs)
-        if getattr(obj, "__header__", None) is not None:
-            obj.__header__ = functools.partial(obj.__header__, object())
 
         # We do this here instead of inside of __call__ so that exceptions
         # coming from the transaction manager get caught by the autoretry

{"golden_diff": "diff --git a/warehouse/tasks.py b/warehouse/tasks.py\n--- a/warehouse/tasks.py\n+++ b/warehouse/tasks.py\n@@ -51,8 +51,6 @@\n class WarehouseTask(celery.Task):\n     def __new__(cls, *args, **kwargs):\n         obj = super().__new__(cls, *args, **kwargs)\n-        if getattr(obj, \"__header__\", None) is not None:\n-            obj.__header__ = functools.partial(obj.__header__, object())\n \n         # We do this here instead of inside of __call__ so that exceptions\n         # coming from the transaction manager get caught by the autoretry\n", "issue": "Celerybeat Cleanup Error\n```\r\nworker_1         | Traceback (most recent call last):\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/celery/beat.py\", line 320, in apply_async\r\nworker_1         |     **entry.options)\r\nworker_1         |   File \"/opt/warehouse/src/warehouse/tasks.py\", line 95, in apply_async\r\nworker_1         |     return super().apply_async(*args, **kwargs)\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/celery/app/task.py\", line 518, in apply_async\r\nworker_1         |     check_arguments(*(args or ()), **(kwargs or {}))\r\nworker_1         | TypeError: backend_cleanup() takes 0 positional arguments but 1 was given\r\nworker_1         |\r\nworker_1         | During handling of the above exception, another exception occurred:\r\nworker_1         |\r\nworker_1         | Traceback (most recent call last):\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/celery/beat.py\", line 222, in apply_entry\r\nworker_1         |     result = self.apply_async(entry, producer=producer, advance=False)\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/celery/beat.py\", line 328, in apply_async\r\nworker_1         |     entry, exc=exc)), sys.exc_info()[2])\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/vine/five.py\", line 178, in reraise\r\nworker_1         |     raise value.with_traceback(tb)\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/celery/beat.py\", line 320, in apply_async\r\nworker_1         |     **entry.options)\r\nworker_1         |   File \"/opt/warehouse/src/warehouse/tasks.py\", line 95, in apply_async\r\nworker_1         |     return super().apply_async(*args, **kwargs)\r\nworker_1         |   File \"/opt/warehouse/lib/python3.6/site-packages/celery/app/task.py\", line 518, in apply_async\r\nworker_1         |     check_arguments(*(args or ()), **(kwargs or {}))\r\nworker_1         | celery.beat.SchedulingError: Couldn't apply scheduled task celery.backend_cleanup: backend_cleanup() takes 0 positional arguments but 1 was given\r\n```\r\n\r\nI assume this is an integration issue with how the warehouse <-> celery integration happens. Not sure what it's actual affect is though.\n", "before_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport functools\nimport logging\nimport urllib.parse\n\nimport celery\nimport celery.app.backends\nimport celery.backends.redis\nimport pyramid.scripting\nimport pyramid_retry\nimport transaction\nimport venusian\n\nfrom kombu import Queue\nfrom pyramid.threadlocal import get_current_request\n\nfrom warehouse.config import Environment\nfrom warehouse.metrics import IMetricsService\n\n# We need to trick Celery into supporting rediss:// URLs which is how redis-py\n# signals that you should use Redis with TLS.\ncelery.app.backends.BACKEND_ALIASES[\n    \"rediss\"\n] = \"warehouse.tasks:TLSRedisBackend\"  # noqa\n\n\n# We need to register that the sqs:// url scheme uses a netloc\nurllib.parse.uses_netloc.append(\"sqs\")\n\nlogger = logging.getLogger(__name__)\n\n\nclass TLSRedisBackend(celery.backends.redis.RedisBackend):\n    def _params_from_url(self, url, defaults):\n        params = super()._params_from_url(url, defaults)\n        params.update({\"connection_class\": self.redis.SSLConnection})\n        return params\n\n\nclass WarehouseTask(celery.Task):\n    def __new__(cls, *args, **kwargs):\n        obj = super().__new__(cls, *args, **kwargs)\n        if getattr(obj, \"__header__\", None) is not None:\n            obj.__header__ = functools.partial(obj.__header__, object())\n\n        # We do this here instead of inside of __call__ so that exceptions\n        # coming from the transaction manager get caught by the autoretry\n        # mechanism.\n        @functools.wraps(obj.run)\n        def run(*args, **kwargs):\n            original_run = obj._wh_original_run\n            request = obj.get_request()\n            metrics = request.find_service(IMetricsService, context=None)\n            metric_tags = [f\"task:{obj.name}\"]\n\n            with request.tm, metrics.timed(\"warehouse.task.run\", tags=metric_tags):\n                try:\n                    result = original_run(*args, **kwargs)\n                    metrics.increment(\"warehouse.task.complete\", tags=metric_tags)\n                    return result\n                except BaseException as exc:\n                    if isinstance(\n                        exc, pyramid_retry.RetryableException\n                    ) or pyramid_retry.IRetryableError.providedBy(exc):\n                        metrics.increment(\"warehouse.task.retried\", tags=metric_tags)\n                        raise obj.retry(exc=exc)\n                    metrics.increment(\"warehouse.task.failed\", tags=metric_tags)\n                    raise\n\n        obj._wh_original_run, obj.run = obj.run, run\n\n        return obj\n\n    def __call__(self, *args, **kwargs):\n        return super().__call__(*(self.get_request(),) + args, **kwargs)\n\n    def get_request(self):\n        if not hasattr(self.request, \"pyramid_env\"):\n            registry = self.app.pyramid_config.registry\n            env = pyramid.scripting.prepare(registry=registry)\n            env[\"request\"].tm = transaction.TransactionManager(explicit=True)\n            self.request.update(pyramid_env=env)\n\n        return self.request.pyramid_env[\"request\"]\n\n    def after_return(self, status, retval, task_id, args, kwargs, einfo):\n        if hasattr(self.request, \"pyramid_env\"):\n            pyramid_env = self.request.pyramid_env\n            pyramid_env[\"request\"]._process_finished_callbacks()\n            pyramid_env[\"closer\"]()\n\n    def apply_async(self, *args, **kwargs):\n        # The API design of Celery makes this threadlocal pretty impossible to\n        # avoid :(\n        request = get_current_request()\n\n        # If for whatever reason we were unable to get a request we'll just\n        # skip this and call the original method to send this immediately.\n        if request is None or not hasattr(request, \"tm\"):\n            return super().apply_async(*args, **kwargs)\n\n        # This will break things that expect to get an AsyncResult because\n        # we're no longer going to be returning an async result from this when\n        # called from within a request, response cycle. Ideally we shouldn't be\n        # waiting for responses in a request/response cycle anyways though.\n        request.tm.get().addAfterCommitHook(\n            self._after_commit_hook, args=args, kws=kwargs\n        )\n\n    def _after_commit_hook(self, success, *args, **kwargs):\n        if success:\n            super().apply_async(*args, **kwargs)\n\n\ndef task(**kwargs):\n    kwargs.setdefault(\"shared\", False)\n\n    def deco(wrapped):\n        def callback(scanner, name, wrapped):\n            celery_app = scanner.config.registry[\"celery.app\"]\n            celery_app.task(**kwargs)(wrapped)\n\n        venusian.attach(wrapped, callback)\n\n        return wrapped\n\n    return deco\n\n\ndef _get_task(celery_app, task_func):\n    task_name = celery_app.gen_task_name(task_func.__name__, task_func.__module__)\n    return celery_app.tasks[task_name]\n\n\ndef _get_task_from_request(request):\n    celery_app = request.registry[\"celery.app\"]\n    return functools.partial(_get_task, celery_app)\n\n\ndef _get_task_from_config(config, task):\n    celery_app = config.registry[\"celery.app\"]\n    return _get_task(celery_app, task)\n\n\ndef _get_celery_app(config):\n    return config.registry[\"celery.app\"]\n\n\ndef _add_periodic_task(config, schedule, func, args=(), kwargs=(), name=None, **opts):\n    def add_task():\n        config.registry[\"celery.app\"].add_periodic_task(\n            schedule, config.task(func).s(), args=args, kwargs=kwargs, name=name, **opts\n        )\n\n    config.action(None, add_task, order=100)\n\n\ndef includeme(config):\n    s = config.registry.settings\n\n    broker_transport_options = {}\n\n    broker_url = s[\"celery.broker_url\"]\n    if broker_url.startswith(\"sqs://\"):\n        parsed_url = urllib.parse.urlparse(broker_url)\n        parsed_query = urllib.parse.parse_qs(parsed_url.query)\n        # Celery doesn't handle paths/query arms being passed into the SQS broker,\n        # so we'll just remove them from here.\n        broker_url = urllib.parse.urlunparse(parsed_url[:2] + (\"\", \"\", \"\", \"\"))\n\n        if \"queue_name_prefix\" in parsed_query:\n            broker_transport_options[\"queue_name_prefix\"] = (\n                parsed_query[\"queue_name_prefix\"][0] + \"-\"\n            )\n\n        if \"region\" in parsed_query:\n            broker_transport_options[\"region\"] = parsed_query[\"region\"][0]\n\n    config.registry[\"celery.app\"] = celery.Celery(\n        \"warehouse\", autofinalize=False, set_as_current=False\n    )\n    config.registry[\"celery.app\"].conf.update(\n        accept_content=[\"json\", \"msgpack\"],\n        broker_url=broker_url,\n        broker_use_ssl=s[\"warehouse.env\"] == Environment.production,\n        broker_transport_options=broker_transport_options,\n        task_default_queue=\"default\",\n        task_default_routing_key=\"task.default\",\n        task_queue_ha_policy=\"all\",\n        task_queues=(\n            Queue(\"default\", routing_key=\"task.#\"),\n            Queue(\"malware\", routing_key=\"malware.#\"),\n        ),\n        task_routes={\"warehouse.malware.tasks.*\": {\"queue\": \"malware\"}},\n        task_serializer=\"json\",\n        worker_disable_rate_limits=True,\n        REDBEAT_REDIS_URL=s[\"celery.scheduler_url\"],\n    )\n    config.registry[\"celery.app\"].Task = WarehouseTask\n    config.registry[\"celery.app\"].pyramid_config = config\n\n    config.action((\"celery\", \"finalize\"), config.registry[\"celery.app\"].finalize)\n\n    config.add_directive(\"add_periodic_task\", _add_periodic_task, action_wrap=False)\n    config.add_directive(\"make_celery_app\", _get_celery_app, action_wrap=False)\n    config.add_directive(\"task\", _get_task_from_config, action_wrap=False)\n    config.add_request_method(_get_task_from_request, name=\"task\", reify=True)\n", "path": "warehouse/tasks.py"}], "after_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport functools\nimport logging\nimport urllib.parse\n\nimport celery\nimport celery.app.backends\nimport celery.backends.redis\nimport pyramid.scripting\nimport pyramid_retry\nimport transaction\nimport venusian\n\nfrom kombu import Queue\nfrom pyramid.threadlocal import get_current_request\n\nfrom warehouse.config import Environment\nfrom warehouse.metrics import IMetricsService\n\n# We need to trick Celery into supporting rediss:// URLs which is how redis-py\n# signals that you should use Redis with TLS.\ncelery.app.backends.BACKEND_ALIASES[\n    \"rediss\"\n] = \"warehouse.tasks:TLSRedisBackend\"  # noqa\n\n\n# We need to register that the sqs:// url scheme uses a netloc\nurllib.parse.uses_netloc.append(\"sqs\")\n\nlogger = logging.getLogger(__name__)\n\n\nclass TLSRedisBackend(celery.backends.redis.RedisBackend):\n    def _params_from_url(self, url, defaults):\n        params = super()._params_from_url(url, defaults)\n        params.update({\"connection_class\": self.redis.SSLConnection})\n        return params\n\n\nclass WarehouseTask(celery.Task):\n    def __new__(cls, *args, **kwargs):\n        obj = super().__new__(cls, *args, **kwargs)\n\n        # We do this here instead of inside of __call__ so that exceptions\n        # coming from the transaction manager get caught by the autoretry\n        # mechanism.\n        @functools.wraps(obj.run)\n        def run(*args, **kwargs):\n            original_run = obj._wh_original_run\n            request = obj.get_request()\n            metrics = request.find_service(IMetricsService, context=None)\n            metric_tags = [f\"task:{obj.name}\"]\n\n            with request.tm, metrics.timed(\"warehouse.task.run\", tags=metric_tags):\n                try:\n                    result = original_run(*args, **kwargs)\n                    metrics.increment(\"warehouse.task.complete\", tags=metric_tags)\n                    return result\n                except BaseException as exc:\n                    if isinstance(\n                        exc, pyramid_retry.RetryableException\n                    ) or pyramid_retry.IRetryableError.providedBy(exc):\n                        metrics.increment(\"warehouse.task.retried\", tags=metric_tags)\n                        raise obj.retry(exc=exc)\n                    metrics.increment(\"warehouse.task.failed\", tags=metric_tags)\n                    raise\n\n        obj._wh_original_run, obj.run = obj.run, run\n\n        return obj\n\n    def __call__(self, *args, **kwargs):\n        return super().__call__(*(self.get_request(),) + args, **kwargs)\n\n    def get_request(self):\n        if not hasattr(self.request, \"pyramid_env\"):\n            registry = self.app.pyramid_config.registry\n            env = pyramid.scripting.prepare(registry=registry)\n            env[\"request\"].tm = transaction.TransactionManager(explicit=True)\n            self.request.update(pyramid_env=env)\n\n        return self.request.pyramid_env[\"request\"]\n\n    def after_return(self, status, retval, task_id, args, kwargs, einfo):\n        if hasattr(self.request, \"pyramid_env\"):\n            pyramid_env = self.request.pyramid_env\n            pyramid_env[\"request\"]._process_finished_callbacks()\n            pyramid_env[\"closer\"]()\n\n    def apply_async(self, *args, **kwargs):\n        # The API design of Celery makes this threadlocal pretty impossible to\n        # avoid :(\n        request = get_current_request()\n\n        # If for whatever reason we were unable to get a request we'll just\n        # skip this and call the original method to send this immediately.\n        if request is None or not hasattr(request, \"tm\"):\n            return super().apply_async(*args, **kwargs)\n\n        # This will break things that expect to get an AsyncResult because\n        # we're no longer going to be returning an async result from this when\n        # called from within a request, response cycle. Ideally we shouldn't be\n        # waiting for responses in a request/response cycle anyways though.\n        request.tm.get().addAfterCommitHook(\n            self._after_commit_hook, args=args, kws=kwargs\n        )\n\n    def _after_commit_hook(self, success, *args, **kwargs):\n        if success:\n            super().apply_async(*args, **kwargs)\n\n\ndef task(**kwargs):\n    kwargs.setdefault(\"shared\", False)\n\n    def deco(wrapped):\n        def callback(scanner, name, wrapped):\n            celery_app = scanner.config.registry[\"celery.app\"]\n            celery_app.task(**kwargs)(wrapped)\n\n        venusian.attach(wrapped, callback)\n\n        return wrapped\n\n    return deco\n\n\ndef _get_task(celery_app, task_func):\n    task_name = celery_app.gen_task_name(task_func.__name__, task_func.__module__)\n    return celery_app.tasks[task_name]\n\n\ndef _get_task_from_request(request):\n    celery_app = request.registry[\"celery.app\"]\n    return functools.partial(_get_task, celery_app)\n\n\ndef _get_task_from_config(config, task):\n    celery_app = config.registry[\"celery.app\"]\n    return _get_task(celery_app, task)\n\n\ndef _get_celery_app(config):\n    return config.registry[\"celery.app\"]\n\n\ndef _add_periodic_task(config, schedule, func, args=(), kwargs=(), name=None, **opts):\n    def add_task():\n        config.registry[\"celery.app\"].add_periodic_task(\n            schedule, config.task(func).s(), args=args, kwargs=kwargs, name=name, **opts\n        )\n\n    config.action(None, add_task, order=100)\n\n\ndef includeme(config):\n    s = config.registry.settings\n\n    broker_transport_options = {}\n\n    broker_url = s[\"celery.broker_url\"]\n    if broker_url.startswith(\"sqs://\"):\n        parsed_url = urllib.parse.urlparse(broker_url)\n        parsed_query = urllib.parse.parse_qs(parsed_url.query)\n        # Celery doesn't handle paths/query arms being passed into the SQS broker,\n        # so we'll just remove them from here.\n        broker_url = urllib.parse.urlunparse(parsed_url[:2] + (\"\", \"\", \"\", \"\"))\n\n        if \"queue_name_prefix\" in parsed_query:\n            broker_transport_options[\"queue_name_prefix\"] = (\n                parsed_query[\"queue_name_prefix\"][0] + \"-\"\n            )\n\n        if \"region\" in parsed_query:\n            broker_transport_options[\"region\"] = parsed_query[\"region\"][0]\n\n    config.registry[\"celery.app\"] = celery.Celery(\n        \"warehouse\", autofinalize=False, set_as_current=False\n    )\n    config.registry[\"celery.app\"].conf.update(\n        accept_content=[\"json\", \"msgpack\"],\n        broker_url=broker_url,\n        broker_use_ssl=s[\"warehouse.env\"] == Environment.production,\n        broker_transport_options=broker_transport_options,\n        task_default_queue=\"default\",\n        task_default_routing_key=\"task.default\",\n        task_queue_ha_policy=\"all\",\n        task_queues=(\n            Queue(\"default\", routing_key=\"task.#\"),\n            Queue(\"malware\", routing_key=\"malware.#\"),\n        ),\n        task_routes={\"warehouse.malware.tasks.*\": {\"queue\": \"malware\"}},\n        task_serializer=\"json\",\n        worker_disable_rate_limits=True,\n        REDBEAT_REDIS_URL=s[\"celery.scheduler_url\"],\n    )\n    config.registry[\"celery.app\"].Task = WarehouseTask\n    config.registry[\"celery.app\"].pyramid_config = config\n\n    config.action((\"celery\", \"finalize\"), config.registry[\"celery.app\"].finalize)\n\n    config.add_directive(\"add_periodic_task\", _add_periodic_task, action_wrap=False)\n    config.add_directive(\"make_celery_app\", _get_celery_app, action_wrap=False)\n    config.add_directive(\"task\", _get_task_from_config, action_wrap=False)\n    config.add_request_method(_get_task_from_request, name=\"task\", reify=True)\n", "path": "warehouse/tasks.py"}]}