We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Enforce functools.wraps on all monkeypatches
I'm using [`celery-once`](https://github.com/cameronmaske/celery-once) to make sure some tasks are only run one at a time. The lock is acquired in `apply_async` and released in `after_return`. 

When I use the `CeleryIntegration` I'm not seeing any of the locks being released after the task has finished running. I'm not very familiar with the inner workings of celery, but any ideas what's going on?


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sentry_sdk/integrations/celery.py`
Content:
```
1 from __future__ import absolute_import
2 
3 import sys
4 
5 from celery.exceptions import (  # type: ignore
6     SoftTimeLimitExceeded,
7     Retry,
8     Ignore,
9     Reject,
10 )
11 
12 from sentry_sdk.hub import Hub
13 from sentry_sdk.utils import capture_internal_exceptions, event_from_exception
14 from sentry_sdk.tracing import Span
15 from sentry_sdk._compat import reraise
16 from sentry_sdk.integrations import Integration
17 from sentry_sdk.integrations.logging import ignore_logger
18 from sentry_sdk._types import MYPY
19 
20 if MYPY:
21     from typing import Any
22 
23 
24 CELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)
25 
26 
27 class CeleryIntegration(Integration):
28     identifier = "celery"
29 
30     def __init__(self, propagate_traces=True):
31         # type: (bool) -> None
32         self.propagate_traces = propagate_traces
33 
34     @staticmethod
35     def setup_once():
36         # type: () -> None
37         import celery.app.trace as trace  # type: ignore
38 
39         old_build_tracer = trace.build_tracer
40 
41         def sentry_build_tracer(name, task, *args, **kwargs):
42             if not getattr(task, "_sentry_is_patched", False):
43                 # Need to patch both methods because older celery sometimes
44                 # short-circuits to task.run if it thinks it's safe.
45                 task.__call__ = _wrap_task_call(task, task.__call__)
46                 task.run = _wrap_task_call(task, task.run)
47                 task.apply_async = _wrap_apply_async(task, task.apply_async)
48 
49                 # `build_tracer` is apparently called for every task
50                 # invocation. Can't wrap every celery task for every invocation
51                 # or we will get infinitely nested wrapper functions.
52                 task._sentry_is_patched = True
53 
54             return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))
55 
56         trace.build_tracer = sentry_build_tracer
57 
58         _patch_worker_exit()
59 
60         # This logger logs every status of every task that ran on the worker.
61         # Meaning that every task's breadcrumbs are full of stuff like "Task
62         # <foo> raised unexpected <bar>".
63         ignore_logger("celery.worker.job")
64 
65 
66 def _wrap_apply_async(task, f):
67     def apply_async(*args, **kwargs):
68         hub = Hub.current
69         integration = hub.get_integration(CeleryIntegration)
70         if integration is not None and integration.propagate_traces:
71             headers = None
72             for key, value in hub.iter_trace_propagation_headers():
73                 if headers is None:
74                     headers = dict(kwargs.get("headers") or {})
75                 headers[key] = value
76             if headers is not None:
77                 kwargs["headers"] = headers
78         return f(*args, **kwargs)
79 
80     return apply_async
81 
82 
83 def _wrap_tracer(task, f):
84     # Need to wrap tracer for pushing the scope before prerun is sent, and
85     # popping it after postrun is sent.
86     #
87     # This is the reason we don't use signals for hooking in the first place.
88     # Also because in Celery 3, signal dispatch returns early if one handler
89     # crashes.
90     def _inner(*args, **kwargs):
91         hub = Hub.current
92         if hub.get_integration(CeleryIntegration) is None:
93             return f(*args, **kwargs)
94 
95         with hub.push_scope() as scope:
96             scope._name = "celery"
97             scope.clear_breadcrumbs()
98             scope.add_event_processor(_make_event_processor(task, *args, **kwargs))
99 
100             span = Span.continue_from_headers(args[3].get("headers") or {})
101             span.transaction = "unknown celery task"
102 
103             with capture_internal_exceptions():
104                 # Celery task objects are not a thing to be trusted. Even
105                 # something such as attribute access can fail.
106                 span.transaction = task.name
107 
108             with hub.span(span):
109                 return f(*args, **kwargs)
110 
111     return _inner
112 
113 
114 def _wrap_task_call(task, f):
115     # Need to wrap task call because the exception is caught before we get to
116     # see it. Also celery's reported stacktrace is untrustworthy.
117     def _inner(*args, **kwargs):
118         try:
119             return f(*args, **kwargs)
120         except Exception:
121             exc_info = sys.exc_info()
122             with capture_internal_exceptions():
123                 _capture_exception(task, exc_info)
124             reraise(*exc_info)
125 
126     return _inner
127 
128 
129 def _make_event_processor(task, uuid, args, kwargs, request=None):
130     def event_processor(event, hint):
131         with capture_internal_exceptions():
132             extra = event.setdefault("extra", {})
133             extra["celery-job"] = {
134                 "task_name": task.name,
135                 "args": args,
136                 "kwargs": kwargs,
137             }
138 
139         if "exc_info" in hint:
140             with capture_internal_exceptions():
141                 if issubclass(hint["exc_info"][0], SoftTimeLimitExceeded):
142                     event["fingerprint"] = [
143                         "celery",
144                         "SoftTimeLimitExceeded",
145                         getattr(task, "name", task),
146                     ]
147 
148         return event
149 
150     return event_processor
151 
152 
153 def _capture_exception(task, exc_info):
154     hub = Hub.current
155 
156     if hub.get_integration(CeleryIntegration) is None:
157         return
158     if isinstance(exc_info[1], CELERY_CONTROL_FLOW_EXCEPTIONS):
159         return
160     if hasattr(task, "throws") and isinstance(exc_info[1], task.throws):
161         return
162 
163     # If an integration is there, a client has to be there.
164     client = hub.client  # type: Any
165 
166     event, hint = event_from_exception(
167         exc_info,
168         client_options=client.options,
169         mechanism={"type": "celery", "handled": False},
170     )
171 
172     hub.capture_event(event, hint=hint)
173 
174 
175 def _patch_worker_exit():
176     # Need to flush queue before worker shutdown because a crashing worker will
177     # call os._exit
178     from billiard.pool import Worker  # type: ignore
179 
180     old_workloop = Worker.workloop
181 
182     def sentry_workloop(*args, **kwargs):
183         try:
184             return old_workloop(*args, **kwargs)
185         finally:
186             with capture_internal_exceptions():
187                 hub = Hub.current
188                 if hub.get_integration(CeleryIntegration) is not None:
189                     hub.flush()
190 
191     Worker.workloop = sentry_workloop
192
```
--- 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/sentry_sdk/integrations/celery.py b/sentry_sdk/integrations/celery.py
--- a/sentry_sdk/integrations/celery.py
+++ b/sentry_sdk/integrations/celery.py
@@ -1,5 +1,6 @@
 from __future__ import absolute_import
 
+import functools
 import sys
 
 from celery.exceptions import (  # type: ignore
@@ -64,6 +65,7 @@
 
 
 def _wrap_apply_async(task, f):
+    @functools.wraps(f)
     def apply_async(*args, **kwargs):
         hub = Hub.current
         integration = hub.get_integration(CeleryIntegration)
@@ -87,6 +89,7 @@
     # This is the reason we don't use signals for hooking in the first place.
     # Also because in Celery 3, signal dispatch returns early if one handler
     # crashes.
+    @functools.wraps(f)
     def _inner(*args, **kwargs):
         hub = Hub.current
         if hub.get_integration(CeleryIntegration) is None:
@@ -114,6 +117,11 @@
 def _wrap_task_call(task, f):
     # Need to wrap task call because the exception is caught before we get to
     # see it. Also celery's reported stacktrace is untrustworthy.
+
+    # functools.wraps is important here because celery-once looks at this
+    # method's name.
+    # https://github.com/getsentry/sentry-python/issues/421
+    @functools.wraps(f)
     def _inner(*args, **kwargs):
         try:
             return f(*args, **kwargs)

{"golden_diff": "diff --git a/sentry_sdk/integrations/celery.py b/sentry_sdk/integrations/celery.py\n--- a/sentry_sdk/integrations/celery.py\n+++ b/sentry_sdk/integrations/celery.py\n@@ -1,5 +1,6 @@\n from __future__ import absolute_import\n \n+import functools\n import sys\n \n from celery.exceptions import (  # type: ignore\n@@ -64,6 +65,7 @@\n \n \n def _wrap_apply_async(task, f):\n+    @functools.wraps(f)\n     def apply_async(*args, **kwargs):\n         hub = Hub.current\n         integration = hub.get_integration(CeleryIntegration)\n@@ -87,6 +89,7 @@\n     # This is the reason we don't use signals for hooking in the first place.\n     # Also because in Celery 3, signal dispatch returns early if one handler\n     # crashes.\n+    @functools.wraps(f)\n     def _inner(*args, **kwargs):\n         hub = Hub.current\n         if hub.get_integration(CeleryIntegration) is None:\n@@ -114,6 +117,11 @@\n def _wrap_task_call(task, f):\n     # Need to wrap task call because the exception is caught before we get to\n     # see it. Also celery's reported stacktrace is untrustworthy.\n+\n+    # functools.wraps is important here because celery-once looks at this\n+    # method's name.\n+    # https://github.com/getsentry/sentry-python/issues/421\n+    @functools.wraps(f)\n     def _inner(*args, **kwargs):\n         try:\n             return f(*args, **kwargs)\n", "issue": "Enforce functools.wraps on all monkeypatches\nI'm using [`celery-once`](https://github.com/cameronmaske/celery-once) to make sure some tasks are only run one at a time. The lock is acquired in `apply_async` and released in `after_return`. \r\n\r\nWhen I use the `CeleryIntegration` I'm not seeing any of the locks being released after the task has finished running. I'm not very familiar with the inner workings of celery, but any ideas what's going on?\r\n\n", "before_files": [{"content": "from __future__ import absolute_import\n\nimport sys\n\nfrom celery.exceptions import (  # type: ignore\n    SoftTimeLimitExceeded,\n    Retry,\n    Ignore,\n    Reject,\n)\n\nfrom sentry_sdk.hub import Hub\nfrom sentry_sdk.utils import capture_internal_exceptions, event_from_exception\nfrom sentry_sdk.tracing import Span\nfrom sentry_sdk._compat import reraise\nfrom sentry_sdk.integrations import Integration\nfrom sentry_sdk.integrations.logging import ignore_logger\nfrom sentry_sdk._types import MYPY\n\nif MYPY:\n    from typing import Any\n\n\nCELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)\n\n\nclass CeleryIntegration(Integration):\n    identifier = \"celery\"\n\n    def __init__(self, propagate_traces=True):\n        # type: (bool) -> None\n        self.propagate_traces = propagate_traces\n\n    @staticmethod\n    def setup_once():\n        # type: () -> None\n        import celery.app.trace as trace  # type: ignore\n\n        old_build_tracer = trace.build_tracer\n\n        def sentry_build_tracer(name, task, *args, **kwargs):\n            if not getattr(task, \"_sentry_is_patched\", False):\n                # Need to patch both methods because older celery sometimes\n                # short-circuits to task.run if it thinks it's safe.\n                task.__call__ = _wrap_task_call(task, task.__call__)\n                task.run = _wrap_task_call(task, task.run)\n                task.apply_async = _wrap_apply_async(task, task.apply_async)\n\n                # `build_tracer` is apparently called for every task\n                # invocation. Can't wrap every celery task for every invocation\n                # or we will get infinitely nested wrapper functions.\n                task._sentry_is_patched = True\n\n            return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))\n\n        trace.build_tracer = sentry_build_tracer\n\n        _patch_worker_exit()\n\n        # This logger logs every status of every task that ran on the worker.\n        # Meaning that every task's breadcrumbs are full of stuff like \"Task\n        # <foo> raised unexpected <bar>\".\n        ignore_logger(\"celery.worker.job\")\n\n\ndef _wrap_apply_async(task, f):\n    def apply_async(*args, **kwargs):\n        hub = Hub.current\n        integration = hub.get_integration(CeleryIntegration)\n        if integration is not None and integration.propagate_traces:\n            headers = None\n            for key, value in hub.iter_trace_propagation_headers():\n                if headers is None:\n                    headers = dict(kwargs.get(\"headers\") or {})\n                headers[key] = value\n            if headers is not None:\n                kwargs[\"headers\"] = headers\n        return f(*args, **kwargs)\n\n    return apply_async\n\n\ndef _wrap_tracer(task, f):\n    # Need to wrap tracer for pushing the scope before prerun is sent, and\n    # popping it after postrun is sent.\n    #\n    # This is the reason we don't use signals for hooking in the first place.\n    # Also because in Celery 3, signal dispatch returns early if one handler\n    # crashes.\n    def _inner(*args, **kwargs):\n        hub = Hub.current\n        if hub.get_integration(CeleryIntegration) is None:\n            return f(*args, **kwargs)\n\n        with hub.push_scope() as scope:\n            scope._name = \"celery\"\n            scope.clear_breadcrumbs()\n            scope.add_event_processor(_make_event_processor(task, *args, **kwargs))\n\n            span = Span.continue_from_headers(args[3].get(\"headers\") or {})\n            span.transaction = \"unknown celery task\"\n\n            with capture_internal_exceptions():\n                # Celery task objects are not a thing to be trusted. Even\n                # something such as attribute access can fail.\n                span.transaction = task.name\n\n            with hub.span(span):\n                return f(*args, **kwargs)\n\n    return _inner\n\n\ndef _wrap_task_call(task, f):\n    # Need to wrap task call because the exception is caught before we get to\n    # see it. Also celery's reported stacktrace is untrustworthy.\n    def _inner(*args, **kwargs):\n        try:\n            return f(*args, **kwargs)\n        except Exception:\n            exc_info = sys.exc_info()\n            with capture_internal_exceptions():\n                _capture_exception(task, exc_info)\n            reraise(*exc_info)\n\n    return _inner\n\n\ndef _make_event_processor(task, uuid, args, kwargs, request=None):\n    def event_processor(event, hint):\n        with capture_internal_exceptions():\n            extra = event.setdefault(\"extra\", {})\n            extra[\"celery-job\"] = {\n                \"task_name\": task.name,\n                \"args\": args,\n                \"kwargs\": kwargs,\n            }\n\n        if \"exc_info\" in hint:\n            with capture_internal_exceptions():\n                if issubclass(hint[\"exc_info\"][0], SoftTimeLimitExceeded):\n                    event[\"fingerprint\"] = [\n                        \"celery\",\n                        \"SoftTimeLimitExceeded\",\n                        getattr(task, \"name\", task),\n                    ]\n\n        return event\n\n    return event_processor\n\n\ndef _capture_exception(task, exc_info):\n    hub = Hub.current\n\n    if hub.get_integration(CeleryIntegration) is None:\n        return\n    if isinstance(exc_info[1], CELERY_CONTROL_FLOW_EXCEPTIONS):\n        return\n    if hasattr(task, \"throws\") and isinstance(exc_info[1], task.throws):\n        return\n\n    # If an integration is there, a client has to be there.\n    client = hub.client  # type: Any\n\n    event, hint = event_from_exception(\n        exc_info,\n        client_options=client.options,\n        mechanism={\"type\": \"celery\", \"handled\": False},\n    )\n\n    hub.capture_event(event, hint=hint)\n\n\ndef _patch_worker_exit():\n    # Need to flush queue before worker shutdown because a crashing worker will\n    # call os._exit\n    from billiard.pool import Worker  # type: ignore\n\n    old_workloop = Worker.workloop\n\n    def sentry_workloop(*args, **kwargs):\n        try:\n            return old_workloop(*args, **kwargs)\n        finally:\n            with capture_internal_exceptions():\n                hub = Hub.current\n                if hub.get_integration(CeleryIntegration) is not None:\n                    hub.flush()\n\n    Worker.workloop = sentry_workloop\n", "path": "sentry_sdk/integrations/celery.py"}], "after_files": [{"content": "from __future__ import absolute_import\n\nimport functools\nimport sys\n\nfrom celery.exceptions import (  # type: ignore\n    SoftTimeLimitExceeded,\n    Retry,\n    Ignore,\n    Reject,\n)\n\nfrom sentry_sdk.hub import Hub\nfrom sentry_sdk.utils import capture_internal_exceptions, event_from_exception\nfrom sentry_sdk.tracing import Span\nfrom sentry_sdk._compat import reraise\nfrom sentry_sdk.integrations import Integration\nfrom sentry_sdk.integrations.logging import ignore_logger\nfrom sentry_sdk._types import MYPY\n\nif MYPY:\n    from typing import Any\n\n\nCELERY_CONTROL_FLOW_EXCEPTIONS = (Retry, Ignore, Reject)\n\n\nclass CeleryIntegration(Integration):\n    identifier = \"celery\"\n\n    def __init__(self, propagate_traces=True):\n        # type: (bool) -> None\n        self.propagate_traces = propagate_traces\n\n    @staticmethod\n    def setup_once():\n        # type: () -> None\n        import celery.app.trace as trace  # type: ignore\n\n        old_build_tracer = trace.build_tracer\n\n        def sentry_build_tracer(name, task, *args, **kwargs):\n            if not getattr(task, \"_sentry_is_patched\", False):\n                # Need to patch both methods because older celery sometimes\n                # short-circuits to task.run if it thinks it's safe.\n                task.__call__ = _wrap_task_call(task, task.__call__)\n                task.run = _wrap_task_call(task, task.run)\n                task.apply_async = _wrap_apply_async(task, task.apply_async)\n\n                # `build_tracer` is apparently called for every task\n                # invocation. Can't wrap every celery task for every invocation\n                # or we will get infinitely nested wrapper functions.\n                task._sentry_is_patched = True\n\n            return _wrap_tracer(task, old_build_tracer(name, task, *args, **kwargs))\n\n        trace.build_tracer = sentry_build_tracer\n\n        _patch_worker_exit()\n\n        # This logger logs every status of every task that ran on the worker.\n        # Meaning that every task's breadcrumbs are full of stuff like \"Task\n        # <foo> raised unexpected <bar>\".\n        ignore_logger(\"celery.worker.job\")\n\n\ndef _wrap_apply_async(task, f):\n    @functools.wraps(f)\n    def apply_async(*args, **kwargs):\n        hub = Hub.current\n        integration = hub.get_integration(CeleryIntegration)\n        if integration is not None and integration.propagate_traces:\n            headers = None\n            for key, value in hub.iter_trace_propagation_headers():\n                if headers is None:\n                    headers = dict(kwargs.get(\"headers\") or {})\n                headers[key] = value\n            if headers is not None:\n                kwargs[\"headers\"] = headers\n        return f(*args, **kwargs)\n\n    return apply_async\n\n\ndef _wrap_tracer(task, f):\n    # Need to wrap tracer for pushing the scope before prerun is sent, and\n    # popping it after postrun is sent.\n    #\n    # This is the reason we don't use signals for hooking in the first place.\n    # Also because in Celery 3, signal dispatch returns early if one handler\n    # crashes.\n    @functools.wraps(f)\n    def _inner(*args, **kwargs):\n        hub = Hub.current\n        if hub.get_integration(CeleryIntegration) is None:\n            return f(*args, **kwargs)\n\n        with hub.push_scope() as scope:\n            scope._name = \"celery\"\n            scope.clear_breadcrumbs()\n            scope.add_event_processor(_make_event_processor(task, *args, **kwargs))\n\n            span = Span.continue_from_headers(args[3].get(\"headers\") or {})\n            span.transaction = \"unknown celery task\"\n\n            with capture_internal_exceptions():\n                # Celery task objects are not a thing to be trusted. Even\n                # something such as attribute access can fail.\n                span.transaction = task.name\n\n            with hub.span(span):\n                return f(*args, **kwargs)\n\n    return _inner\n\n\ndef _wrap_task_call(task, f):\n    # Need to wrap task call because the exception is caught before we get to\n    # see it. Also celery's reported stacktrace is untrustworthy.\n\n    # functools.wraps is important here because celery-once looks at this\n    # method's name.\n    # https://github.com/getsentry/sentry-python/issues/421\n    @functools.wraps(f)\n    def _inner(*args, **kwargs):\n        try:\n            return f(*args, **kwargs)\n        except Exception:\n            exc_info = sys.exc_info()\n            with capture_internal_exceptions():\n                _capture_exception(task, exc_info)\n            reraise(*exc_info)\n\n    return _inner\n\n\ndef _make_event_processor(task, uuid, args, kwargs, request=None):\n    def event_processor(event, hint):\n        with capture_internal_exceptions():\n            extra = event.setdefault(\"extra\", {})\n            extra[\"celery-job\"] = {\n                \"task_name\": task.name,\n                \"args\": args,\n                \"kwargs\": kwargs,\n            }\n\n        if \"exc_info\" in hint:\n            with capture_internal_exceptions():\n                if issubclass(hint[\"exc_info\"][0], SoftTimeLimitExceeded):\n                    event[\"fingerprint\"] = [\n                        \"celery\",\n                        \"SoftTimeLimitExceeded\",\n                        getattr(task, \"name\", task),\n                    ]\n\n        return event\n\n    return event_processor\n\n\ndef _capture_exception(task, exc_info):\n    hub = Hub.current\n\n    if hub.get_integration(CeleryIntegration) is None:\n        return\n    if isinstance(exc_info[1], CELERY_CONTROL_FLOW_EXCEPTIONS):\n        return\n    if hasattr(task, \"throws\") and isinstance(exc_info[1], task.throws):\n        return\n\n    # If an integration is there, a client has to be there.\n    client = hub.client  # type: Any\n\n    event, hint = event_from_exception(\n        exc_info,\n        client_options=client.options,\n        mechanism={\"type\": \"celery\", \"handled\": False},\n    )\n\n    hub.capture_event(event, hint=hint)\n\n\ndef _patch_worker_exit():\n    # Need to flush queue before worker shutdown because a crashing worker will\n    # call os._exit\n    from billiard.pool import Worker  # type: ignore\n\n    old_workloop = Worker.workloop\n\n    def sentry_workloop(*args, **kwargs):\n        try:\n            return old_workloop(*args, **kwargs)\n        finally:\n            with capture_internal_exceptions():\n                hub = Hub.current\n                if hub.get_integration(CeleryIntegration) is not None:\n                    hub.flush()\n\n    Worker.workloop = sentry_workloop\n", "path": "sentry_sdk/integrations/celery.py"}]}