We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Arq integration ignores functions and cron jobs passed as keyword arguments
### How do you use Sentry?

Sentry Saas (sentry.io)

### Version

1.29.2

### Steps to Reproduce

`arq` allows you to pass the functions as a keyword argument, not just defining on the `WorkerSettings` class.

```python
async def my_coroutine(state):
    pass

arq.run_worker(object(), functions=[my_coroutine])
```

### Expected Result

Unhandled errors in `my_coroutine` should be sent to sentry

### Actual Result

Errors are not sent to Sentry

--- 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/arq.py`
Content:
```
1 from __future__ import absolute_import
2 
3 import sys
4 
5 from sentry_sdk._compat import reraise
6 from sentry_sdk._types import TYPE_CHECKING
7 from sentry_sdk import Hub
8 from sentry_sdk.consts import OP
9 from sentry_sdk.hub import _should_send_default_pii
10 from sentry_sdk.integrations import DidNotEnable, Integration
11 from sentry_sdk.integrations.logging import ignore_logger
12 from sentry_sdk.tracing import Transaction, TRANSACTION_SOURCE_TASK
13 from sentry_sdk.utils import (
14     capture_internal_exceptions,
15     event_from_exception,
16     SENSITIVE_DATA_SUBSTITUTE,
17     parse_version,
18 )
19 
20 try:
21     import arq.worker
22     from arq.version import VERSION as ARQ_VERSION
23     from arq.connections import ArqRedis
24     from arq.worker import JobExecutionFailed, Retry, RetryJob, Worker
25 except ImportError:
26     raise DidNotEnable("Arq is not installed")
27 
28 if TYPE_CHECKING:
29     from typing import Any, Dict, Optional, Union
30 
31     from sentry_sdk._types import EventProcessor, Event, ExcInfo, Hint
32 
33     from arq.cron import CronJob
34     from arq.jobs import Job
35     from arq.typing import WorkerCoroutine
36     from arq.worker import Function
37 
38 ARQ_CONTROL_FLOW_EXCEPTIONS = (JobExecutionFailed, Retry, RetryJob)
39 
40 
41 class ArqIntegration(Integration):
42     identifier = "arq"
43 
44     @staticmethod
45     def setup_once():
46         # type: () -> None
47 
48         try:
49             if isinstance(ARQ_VERSION, str):
50                 version = parse_version(ARQ_VERSION)
51             else:
52                 version = ARQ_VERSION.version[:2]
53 
54         except (TypeError, ValueError):
55             version = None
56 
57         if version is None:
58             raise DidNotEnable("Unparsable arq version: {}".format(ARQ_VERSION))
59 
60         if version < (0, 23):
61             raise DidNotEnable("arq 0.23 or newer required.")
62 
63         patch_enqueue_job()
64         patch_run_job()
65         patch_create_worker()
66 
67         ignore_logger("arq.worker")
68 
69 
70 def patch_enqueue_job():
71     # type: () -> None
72     old_enqueue_job = ArqRedis.enqueue_job
73 
74     async def _sentry_enqueue_job(self, function, *args, **kwargs):
75         # type: (ArqRedis, str, *Any, **Any) -> Optional[Job]
76         hub = Hub.current
77 
78         if hub.get_integration(ArqIntegration) is None:
79             return await old_enqueue_job(self, function, *args, **kwargs)
80 
81         with hub.start_span(op=OP.QUEUE_SUBMIT_ARQ, description=function):
82             return await old_enqueue_job(self, function, *args, **kwargs)
83 
84     ArqRedis.enqueue_job = _sentry_enqueue_job
85 
86 
87 def patch_run_job():
88     # type: () -> None
89     old_run_job = Worker.run_job
90 
91     async def _sentry_run_job(self, job_id, score):
92         # type: (Worker, str, int) -> None
93         hub = Hub(Hub.current)
94 
95         if hub.get_integration(ArqIntegration) is None:
96             return await old_run_job(self, job_id, score)
97 
98         with hub.push_scope() as scope:
99             scope._name = "arq"
100             scope.clear_breadcrumbs()
101 
102             transaction = Transaction(
103                 name="unknown arq task",
104                 status="ok",
105                 op=OP.QUEUE_TASK_ARQ,
106                 source=TRANSACTION_SOURCE_TASK,
107             )
108 
109             with hub.start_transaction(transaction):
110                 return await old_run_job(self, job_id, score)
111 
112     Worker.run_job = _sentry_run_job
113 
114 
115 def _capture_exception(exc_info):
116     # type: (ExcInfo) -> None
117     hub = Hub.current
118 
119     if hub.scope.transaction is not None:
120         if exc_info[0] in ARQ_CONTROL_FLOW_EXCEPTIONS:
121             hub.scope.transaction.set_status("aborted")
122             return
123 
124         hub.scope.transaction.set_status("internal_error")
125 
126     event, hint = event_from_exception(
127         exc_info,
128         client_options=hub.client.options if hub.client else None,
129         mechanism={"type": ArqIntegration.identifier, "handled": False},
130     )
131     hub.capture_event(event, hint=hint)
132 
133 
134 def _make_event_processor(ctx, *args, **kwargs):
135     # type: (Dict[Any, Any], *Any, **Any) -> EventProcessor
136     def event_processor(event, hint):
137         # type: (Event, Hint) -> Optional[Event]
138 
139         hub = Hub.current
140 
141         with capture_internal_exceptions():
142             if hub.scope.transaction is not None:
143                 hub.scope.transaction.name = ctx["job_name"]
144                 event["transaction"] = ctx["job_name"]
145 
146             tags = event.setdefault("tags", {})
147             tags["arq_task_id"] = ctx["job_id"]
148             tags["arq_task_retry"] = ctx["job_try"] > 1
149             extra = event.setdefault("extra", {})
150             extra["arq-job"] = {
151                 "task": ctx["job_name"],
152                 "args": args
153                 if _should_send_default_pii()
154                 else SENSITIVE_DATA_SUBSTITUTE,
155                 "kwargs": kwargs
156                 if _should_send_default_pii()
157                 else SENSITIVE_DATA_SUBSTITUTE,
158                 "retry": ctx["job_try"],
159             }
160 
161         return event
162 
163     return event_processor
164 
165 
166 def _wrap_coroutine(name, coroutine):
167     # type: (str, WorkerCoroutine) -> WorkerCoroutine
168     async def _sentry_coroutine(ctx, *args, **kwargs):
169         # type: (Dict[Any, Any], *Any, **Any) -> Any
170         hub = Hub.current
171         if hub.get_integration(ArqIntegration) is None:
172             return await coroutine(*args, **kwargs)
173 
174         hub.scope.add_event_processor(
175             _make_event_processor({**ctx, "job_name": name}, *args, **kwargs)
176         )
177 
178         try:
179             result = await coroutine(ctx, *args, **kwargs)
180         except Exception:
181             exc_info = sys.exc_info()
182             _capture_exception(exc_info)
183             reraise(*exc_info)
184 
185         return result
186 
187     return _sentry_coroutine
188 
189 
190 def patch_create_worker():
191     # type: () -> None
192     old_create_worker = arq.worker.create_worker
193 
194     def _sentry_create_worker(*args, **kwargs):
195         # type: (*Any, **Any) -> Worker
196         hub = Hub.current
197 
198         if hub.get_integration(ArqIntegration) is None:
199             return old_create_worker(*args, **kwargs)
200 
201         settings_cls = args[0]
202 
203         functions = settings_cls.functions
204         cron_jobs = settings_cls.cron_jobs
205 
206         settings_cls.functions = [_get_arq_function(func) for func in functions]
207         settings_cls.cron_jobs = [_get_arq_cron_job(cron_job) for cron_job in cron_jobs]
208 
209         return old_create_worker(*args, **kwargs)
210 
211     arq.worker.create_worker = _sentry_create_worker
212 
213 
214 def _get_arq_function(func):
215     # type: (Union[str, Function, WorkerCoroutine]) -> Function
216     arq_func = arq.worker.func(func)
217     arq_func.coroutine = _wrap_coroutine(arq_func.name, arq_func.coroutine)
218 
219     return arq_func
220 
221 
222 def _get_arq_cron_job(cron_job):
223     # type: (CronJob) -> CronJob
224     cron_job.coroutine = _wrap_coroutine(cron_job.name, cron_job.coroutine)
225 
226     return cron_job
227
```
--- 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/arq.py b/sentry_sdk/integrations/arq.py
--- a/sentry_sdk/integrations/arq.py
+++ b/sentry_sdk/integrations/arq.py
@@ -200,11 +200,23 @@
 
         settings_cls = args[0]
 
-        functions = settings_cls.functions
-        cron_jobs = settings_cls.cron_jobs
-
-        settings_cls.functions = [_get_arq_function(func) for func in functions]
-        settings_cls.cron_jobs = [_get_arq_cron_job(cron_job) for cron_job in cron_jobs]
+        if hasattr(settings_cls, "functions"):
+            settings_cls.functions = [
+                _get_arq_function(func) for func in settings_cls.functions
+            ]
+        if hasattr(settings_cls, "cron_jobs"):
+            settings_cls.cron_jobs = [
+                _get_arq_cron_job(cron_job) for cron_job in settings_cls.cron_jobs
+            ]
+
+        if "functions" in kwargs:
+            kwargs["functions"] = [
+                _get_arq_function(func) for func in kwargs["functions"]
+            ]
+        if "cron_jobs" in kwargs:
+            kwargs["cron_jobs"] = [
+                _get_arq_cron_job(cron_job) for cron_job in kwargs["cron_jobs"]
+            ]
 
         return old_create_worker(*args, **kwargs)

{"golden_diff": "diff --git a/sentry_sdk/integrations/arq.py b/sentry_sdk/integrations/arq.py\n--- a/sentry_sdk/integrations/arq.py\n+++ b/sentry_sdk/integrations/arq.py\n@@ -200,11 +200,23 @@\n \n         settings_cls = args[0]\n \n-        functions = settings_cls.functions\n-        cron_jobs = settings_cls.cron_jobs\n-\n-        settings_cls.functions = [_get_arq_function(func) for func in functions]\n-        settings_cls.cron_jobs = [_get_arq_cron_job(cron_job) for cron_job in cron_jobs]\n+        if hasattr(settings_cls, \"functions\"):\n+            settings_cls.functions = [\n+                _get_arq_function(func) for func in settings_cls.functions\n+            ]\n+        if hasattr(settings_cls, \"cron_jobs\"):\n+            settings_cls.cron_jobs = [\n+                _get_arq_cron_job(cron_job) for cron_job in settings_cls.cron_jobs\n+            ]\n+\n+        if \"functions\" in kwargs:\n+            kwargs[\"functions\"] = [\n+                _get_arq_function(func) for func in kwargs[\"functions\"]\n+            ]\n+        if \"cron_jobs\" in kwargs:\n+            kwargs[\"cron_jobs\"] = [\n+                _get_arq_cron_job(cron_job) for cron_job in kwargs[\"cron_jobs\"]\n+            ]\n \n         return old_create_worker(*args, **kwargs)\n", "issue": "Arq integration ignores functions and cron jobs passed as keyword arguments\n### How do you use Sentry?\n\nSentry Saas (sentry.io)\n\n### Version\n\n1.29.2\n\n### Steps to Reproduce\n\n`arq` allows you to pass the functions as a keyword argument, not just defining on the `WorkerSettings` class.\r\n\r\n```python\r\nasync def my_coroutine(state):\r\n    pass\r\n\r\narq.run_worker(object(), functions=[my_coroutine])\r\n```\n\n### Expected Result\n\nUnhandled errors in `my_coroutine` should be sent to sentry\n\n### Actual Result\n\nErrors are not sent to Sentry\n", "before_files": [{"content": "from __future__ import absolute_import\n\nimport sys\n\nfrom sentry_sdk._compat import reraise\nfrom sentry_sdk._types import TYPE_CHECKING\nfrom sentry_sdk import Hub\nfrom sentry_sdk.consts import OP\nfrom sentry_sdk.hub import _should_send_default_pii\nfrom sentry_sdk.integrations import DidNotEnable, Integration\nfrom sentry_sdk.integrations.logging import ignore_logger\nfrom sentry_sdk.tracing import Transaction, TRANSACTION_SOURCE_TASK\nfrom sentry_sdk.utils import (\n    capture_internal_exceptions,\n    event_from_exception,\n    SENSITIVE_DATA_SUBSTITUTE,\n    parse_version,\n)\n\ntry:\n    import arq.worker\n    from arq.version import VERSION as ARQ_VERSION\n    from arq.connections import ArqRedis\n    from arq.worker import JobExecutionFailed, Retry, RetryJob, Worker\nexcept ImportError:\n    raise DidNotEnable(\"Arq is not installed\")\n\nif TYPE_CHECKING:\n    from typing import Any, Dict, Optional, Union\n\n    from sentry_sdk._types import EventProcessor, Event, ExcInfo, Hint\n\n    from arq.cron import CronJob\n    from arq.jobs import Job\n    from arq.typing import WorkerCoroutine\n    from arq.worker import Function\n\nARQ_CONTROL_FLOW_EXCEPTIONS = (JobExecutionFailed, Retry, RetryJob)\n\n\nclass ArqIntegration(Integration):\n    identifier = \"arq\"\n\n    @staticmethod\n    def setup_once():\n        # type: () -> None\n\n        try:\n            if isinstance(ARQ_VERSION, str):\n                version = parse_version(ARQ_VERSION)\n            else:\n                version = ARQ_VERSION.version[:2]\n\n        except (TypeError, ValueError):\n            version = None\n\n        if version is None:\n            raise DidNotEnable(\"Unparsable arq version: {}\".format(ARQ_VERSION))\n\n        if version < (0, 23):\n            raise DidNotEnable(\"arq 0.23 or newer required.\")\n\n        patch_enqueue_job()\n        patch_run_job()\n        patch_create_worker()\n\n        ignore_logger(\"arq.worker\")\n\n\ndef patch_enqueue_job():\n    # type: () -> None\n    old_enqueue_job = ArqRedis.enqueue_job\n\n    async def _sentry_enqueue_job(self, function, *args, **kwargs):\n        # type: (ArqRedis, str, *Any, **Any) -> Optional[Job]\n        hub = Hub.current\n\n        if hub.get_integration(ArqIntegration) is None:\n            return await old_enqueue_job(self, function, *args, **kwargs)\n\n        with hub.start_span(op=OP.QUEUE_SUBMIT_ARQ, description=function):\n            return await old_enqueue_job(self, function, *args, **kwargs)\n\n    ArqRedis.enqueue_job = _sentry_enqueue_job\n\n\ndef patch_run_job():\n    # type: () -> None\n    old_run_job = Worker.run_job\n\n    async def _sentry_run_job(self, job_id, score):\n        # type: (Worker, str, int) -> None\n        hub = Hub(Hub.current)\n\n        if hub.get_integration(ArqIntegration) is None:\n            return await old_run_job(self, job_id, score)\n\n        with hub.push_scope() as scope:\n            scope._name = \"arq\"\n            scope.clear_breadcrumbs()\n\n            transaction = Transaction(\n                name=\"unknown arq task\",\n                status=\"ok\",\n                op=OP.QUEUE_TASK_ARQ,\n                source=TRANSACTION_SOURCE_TASK,\n            )\n\n            with hub.start_transaction(transaction):\n                return await old_run_job(self, job_id, score)\n\n    Worker.run_job = _sentry_run_job\n\n\ndef _capture_exception(exc_info):\n    # type: (ExcInfo) -> None\n    hub = Hub.current\n\n    if hub.scope.transaction is not None:\n        if exc_info[0] in ARQ_CONTROL_FLOW_EXCEPTIONS:\n            hub.scope.transaction.set_status(\"aborted\")\n            return\n\n        hub.scope.transaction.set_status(\"internal_error\")\n\n    event, hint = event_from_exception(\n        exc_info,\n        client_options=hub.client.options if hub.client else None,\n        mechanism={\"type\": ArqIntegration.identifier, \"handled\": False},\n    )\n    hub.capture_event(event, hint=hint)\n\n\ndef _make_event_processor(ctx, *args, **kwargs):\n    # type: (Dict[Any, Any], *Any, **Any) -> EventProcessor\n    def event_processor(event, hint):\n        # type: (Event, Hint) -> Optional[Event]\n\n        hub = Hub.current\n\n        with capture_internal_exceptions():\n            if hub.scope.transaction is not None:\n                hub.scope.transaction.name = ctx[\"job_name\"]\n                event[\"transaction\"] = ctx[\"job_name\"]\n\n            tags = event.setdefault(\"tags\", {})\n            tags[\"arq_task_id\"] = ctx[\"job_id\"]\n            tags[\"arq_task_retry\"] = ctx[\"job_try\"] > 1\n            extra = event.setdefault(\"extra\", {})\n            extra[\"arq-job\"] = {\n                \"task\": ctx[\"job_name\"],\n                \"args\": args\n                if _should_send_default_pii()\n                else SENSITIVE_DATA_SUBSTITUTE,\n                \"kwargs\": kwargs\n                if _should_send_default_pii()\n                else SENSITIVE_DATA_SUBSTITUTE,\n                \"retry\": ctx[\"job_try\"],\n            }\n\n        return event\n\n    return event_processor\n\n\ndef _wrap_coroutine(name, coroutine):\n    # type: (str, WorkerCoroutine) -> WorkerCoroutine\n    async def _sentry_coroutine(ctx, *args, **kwargs):\n        # type: (Dict[Any, Any], *Any, **Any) -> Any\n        hub = Hub.current\n        if hub.get_integration(ArqIntegration) is None:\n            return await coroutine(*args, **kwargs)\n\n        hub.scope.add_event_processor(\n            _make_event_processor({**ctx, \"job_name\": name}, *args, **kwargs)\n        )\n\n        try:\n            result = await coroutine(ctx, *args, **kwargs)\n        except Exception:\n            exc_info = sys.exc_info()\n            _capture_exception(exc_info)\n            reraise(*exc_info)\n\n        return result\n\n    return _sentry_coroutine\n\n\ndef patch_create_worker():\n    # type: () -> None\n    old_create_worker = arq.worker.create_worker\n\n    def _sentry_create_worker(*args, **kwargs):\n        # type: (*Any, **Any) -> Worker\n        hub = Hub.current\n\n        if hub.get_integration(ArqIntegration) is None:\n            return old_create_worker(*args, **kwargs)\n\n        settings_cls = args[0]\n\n        functions = settings_cls.functions\n        cron_jobs = settings_cls.cron_jobs\n\n        settings_cls.functions = [_get_arq_function(func) for func in functions]\n        settings_cls.cron_jobs = [_get_arq_cron_job(cron_job) for cron_job in cron_jobs]\n\n        return old_create_worker(*args, **kwargs)\n\n    arq.worker.create_worker = _sentry_create_worker\n\n\ndef _get_arq_function(func):\n    # type: (Union[str, Function, WorkerCoroutine]) -> Function\n    arq_func = arq.worker.func(func)\n    arq_func.coroutine = _wrap_coroutine(arq_func.name, arq_func.coroutine)\n\n    return arq_func\n\n\ndef _get_arq_cron_job(cron_job):\n    # type: (CronJob) -> CronJob\n    cron_job.coroutine = _wrap_coroutine(cron_job.name, cron_job.coroutine)\n\n    return cron_job\n", "path": "sentry_sdk/integrations/arq.py"}], "after_files": [{"content": "from __future__ import absolute_import\n\nimport sys\n\nfrom sentry_sdk._compat import reraise\nfrom sentry_sdk._types import TYPE_CHECKING\nfrom sentry_sdk import Hub\nfrom sentry_sdk.consts import OP\nfrom sentry_sdk.hub import _should_send_default_pii\nfrom sentry_sdk.integrations import DidNotEnable, Integration\nfrom sentry_sdk.integrations.logging import ignore_logger\nfrom sentry_sdk.tracing import Transaction, TRANSACTION_SOURCE_TASK\nfrom sentry_sdk.utils import (\n    capture_internal_exceptions,\n    event_from_exception,\n    SENSITIVE_DATA_SUBSTITUTE,\n    parse_version,\n)\n\ntry:\n    import arq.worker\n    from arq.version import VERSION as ARQ_VERSION\n    from arq.connections import ArqRedis\n    from arq.worker import JobExecutionFailed, Retry, RetryJob, Worker\nexcept ImportError:\n    raise DidNotEnable(\"Arq is not installed\")\n\nif TYPE_CHECKING:\n    from typing import Any, Dict, Optional, Union\n\n    from sentry_sdk._types import EventProcessor, Event, ExcInfo, Hint\n\n    from arq.cron import CronJob\n    from arq.jobs import Job\n    from arq.typing import WorkerCoroutine\n    from arq.worker import Function\n\nARQ_CONTROL_FLOW_EXCEPTIONS = (JobExecutionFailed, Retry, RetryJob)\n\n\nclass ArqIntegration(Integration):\n    identifier = \"arq\"\n\n    @staticmethod\n    def setup_once():\n        # type: () -> None\n\n        try:\n            if isinstance(ARQ_VERSION, str):\n                version = parse_version(ARQ_VERSION)\n            else:\n                version = ARQ_VERSION.version[:2]\n\n        except (TypeError, ValueError):\n            version = None\n\n        if version is None:\n            raise DidNotEnable(\"Unparsable arq version: {}\".format(ARQ_VERSION))\n\n        if version < (0, 23):\n            raise DidNotEnable(\"arq 0.23 or newer required.\")\n\n        patch_enqueue_job()\n        patch_run_job()\n        patch_create_worker()\n\n        ignore_logger(\"arq.worker\")\n\n\ndef patch_enqueue_job():\n    # type: () -> None\n    old_enqueue_job = ArqRedis.enqueue_job\n\n    async def _sentry_enqueue_job(self, function, *args, **kwargs):\n        # type: (ArqRedis, str, *Any, **Any) -> Optional[Job]\n        hub = Hub.current\n\n        if hub.get_integration(ArqIntegration) is None:\n            return await old_enqueue_job(self, function, *args, **kwargs)\n\n        with hub.start_span(op=OP.QUEUE_SUBMIT_ARQ, description=function):\n            return await old_enqueue_job(self, function, *args, **kwargs)\n\n    ArqRedis.enqueue_job = _sentry_enqueue_job\n\n\ndef patch_run_job():\n    # type: () -> None\n    old_run_job = Worker.run_job\n\n    async def _sentry_run_job(self, job_id, score):\n        # type: (Worker, str, int) -> None\n        hub = Hub(Hub.current)\n\n        if hub.get_integration(ArqIntegration) is None:\n            return await old_run_job(self, job_id, score)\n\n        with hub.push_scope() as scope:\n            scope._name = \"arq\"\n            scope.clear_breadcrumbs()\n\n            transaction = Transaction(\n                name=\"unknown arq task\",\n                status=\"ok\",\n                op=OP.QUEUE_TASK_ARQ,\n                source=TRANSACTION_SOURCE_TASK,\n            )\n\n            with hub.start_transaction(transaction):\n                return await old_run_job(self, job_id, score)\n\n    Worker.run_job = _sentry_run_job\n\n\ndef _capture_exception(exc_info):\n    # type: (ExcInfo) -> None\n    hub = Hub.current\n\n    if hub.scope.transaction is not None:\n        if exc_info[0] in ARQ_CONTROL_FLOW_EXCEPTIONS:\n            hub.scope.transaction.set_status(\"aborted\")\n            return\n\n        hub.scope.transaction.set_status(\"internal_error\")\n\n    event, hint = event_from_exception(\n        exc_info,\n        client_options=hub.client.options if hub.client else None,\n        mechanism={\"type\": ArqIntegration.identifier, \"handled\": False},\n    )\n    hub.capture_event(event, hint=hint)\n\n\ndef _make_event_processor(ctx, *args, **kwargs):\n    # type: (Dict[Any, Any], *Any, **Any) -> EventProcessor\n    def event_processor(event, hint):\n        # type: (Event, Hint) -> Optional[Event]\n\n        hub = Hub.current\n\n        with capture_internal_exceptions():\n            if hub.scope.transaction is not None:\n                hub.scope.transaction.name = ctx[\"job_name\"]\n                event[\"transaction\"] = ctx[\"job_name\"]\n\n            tags = event.setdefault(\"tags\", {})\n            tags[\"arq_task_id\"] = ctx[\"job_id\"]\n            tags[\"arq_task_retry\"] = ctx[\"job_try\"] > 1\n            extra = event.setdefault(\"extra\", {})\n            extra[\"arq-job\"] = {\n                \"task\": ctx[\"job_name\"],\n                \"args\": args\n                if _should_send_default_pii()\n                else SENSITIVE_DATA_SUBSTITUTE,\n                \"kwargs\": kwargs\n                if _should_send_default_pii()\n                else SENSITIVE_DATA_SUBSTITUTE,\n                \"retry\": ctx[\"job_try\"],\n            }\n\n        return event\n\n    return event_processor\n\n\ndef _wrap_coroutine(name, coroutine):\n    # type: (str, WorkerCoroutine) -> WorkerCoroutine\n    async def _sentry_coroutine(ctx, *args, **kwargs):\n        # type: (Dict[Any, Any], *Any, **Any) -> Any\n        hub = Hub.current\n        if hub.get_integration(ArqIntegration) is None:\n            return await coroutine(*args, **kwargs)\n\n        hub.scope.add_event_processor(\n            _make_event_processor({**ctx, \"job_name\": name}, *args, **kwargs)\n        )\n\n        try:\n            result = await coroutine(ctx, *args, **kwargs)\n        except Exception:\n            exc_info = sys.exc_info()\n            _capture_exception(exc_info)\n            reraise(*exc_info)\n\n        return result\n\n    return _sentry_coroutine\n\n\ndef patch_create_worker():\n    # type: () -> None\n    old_create_worker = arq.worker.create_worker\n\n    def _sentry_create_worker(*args, **kwargs):\n        # type: (*Any, **Any) -> Worker\n        hub = Hub.current\n\n        if hub.get_integration(ArqIntegration) is None:\n            return old_create_worker(*args, **kwargs)\n\n        settings_cls = args[0]\n\n        if hasattr(settings_cls, \"functions\"):\n            settings_cls.functions = [\n                _get_arq_function(func) for func in settings_cls.functions\n            ]\n        if hasattr(settings_cls, \"cron_jobs\"):\n            settings_cls.cron_jobs = [\n                _get_arq_cron_job(cron_job) for cron_job in settings_cls.cron_jobs\n            ]\n\n        if \"functions\" in kwargs:\n            kwargs[\"functions\"] = [\n                _get_arq_function(func) for func in kwargs[\"functions\"]\n            ]\n        if \"cron_jobs\" in kwargs:\n            kwargs[\"cron_jobs\"] = [\n                _get_arq_cron_job(cron_job) for cron_job in kwargs[\"cron_jobs\"]\n            ]\n\n        return old_create_worker(*args, **kwargs)\n\n    arq.worker.create_worker = _sentry_create_worker\n\n\ndef _get_arq_function(func):\n    # type: (Union[str, Function, WorkerCoroutine]) -> Function\n    arq_func = arq.worker.func(func)\n    arq_func.coroutine = _wrap_coroutine(arq_func.name, arq_func.coroutine)\n\n    return arq_func\n\n\ndef _get_arq_cron_job(cron_job):\n    # type: (CronJob) -> CronJob\n    cron_job.coroutine = _wrap_coroutine(cron_job.name, cron_job.coroutine)\n\n    return cron_job\n", "path": "sentry_sdk/integrations/arq.py"}]}