We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Exception within  LocalTaskJob._run_mini_scheduler_on_child_tasks brakes Sentry Handler
### Apache Airflow version

2.1.3 (latest released)

### Operating System

Debian GNU/Linux 10 (buster)

### Versions of Apache Airflow Providers

```
apache-airflow-providers-amazon @ file:///root/.cache/pypoetry/artifacts/7f/f7/23/fc7fd3543aa486275ef0385c29063ff0dc391b0fc95dc5aa6cab2cf4e5/apache_airflow_providers_amazon-2.2.0-py3-none-any.whl
apache-airflow-providers-celery @ file:///root/.cache/pypoetry/artifacts/14/80/39/0d9d57205da1d24189ac9c18eb3477664ed2c2618c1467c9809b9a2fbf/apache_airflow_providers_celery-2.0.0-py3-none-any.whl
apache-airflow-providers-ftp @ file:///root/.cache/pypoetry/artifacts/a5/13/da/bf14abc40193a1ee1b82bbd800e3ac230427d7684b9d40998ac3684bef/apache_airflow_providers_ftp-2.0.1-py3-none-any.whl
apache-airflow-providers-http @ file:///root/.cache/pypoetry/artifacts/fc/d7/d2/73c89ef847bbae1704fa403d7e92dba1feead757aae141613980db40ff/apache_airflow_providers_http-2.0.0-py3-none-any.whl
apache-airflow-providers-imap @ file:///root/.cache/pypoetry/artifacts/af/5d/de/21c10bfc7ac076a415dcc3fc909317547e77e38c005487552cf40ddd97/apache_airflow_providers_imap-2.0.1-py3-none-any.whl
apache-airflow-providers-postgres @ file:///root/.cache/pypoetry/artifacts/50/27/e0/9b0d8f4c0abf59967bb87a04a93d73896d9a4558994185dd8bc43bb67f/apache_airflow_providers_postgres-2.2.0-py3-none-any.whl
apache-airflow-providers-redis @ file:///root/.cache/pypoetry/artifacts/7d/95/03/5d2a65ace88ae9a9ce9134b927b1e9639c8680c13a31e58425deae55d1/apache_airflow_providers_redis-2.0.1-py3-none-any.whl
apache-airflow-providers-sqlite @ file:///root/.cache/pypoetry/artifacts/ec/e6/a3/e0d81fef662ccf79609e7d2c4e4440839a464771fd2a002d252c9a401d/apache_airflow_providers_sqlite-2.0.1-py3-none-any.whl
```


### Deployment

Other Docker-based deployment

### Deployment details

We are using the Sentry integration

### What happened

An exception within LocalTaskJobs mini scheduler was handled incorrectly by the Sentry integrations 'enrich_errors' method. This is because it assumes its applied to a method of a TypeInstance task

```
TypeError: cannot pickle 'dict_keys' object
  File "airflow/sentry.py", line 166, in wrapper
    return func(task_instance, *args, **kwargs)
  File "airflow/jobs/local_task_job.py", line 241, in _run_mini_scheduler_on_child_tasks
    partial_dag = task.dag.partial_subset(
  File "airflow/models/dag.py", line 1487, in partial_subset
    dag.task_dict = {
  File "airflow/models/dag.py", line 1488, in <dictcomp>
    t.task_id: copy.deepcopy(t, {id(t.dag): dag})  # type: ignore
  File "copy.py", line 153, in deepcopy
    y = copier(memo)
  File "airflow/models/baseoperator.py", line 970, in __deepcopy__
    setattr(result, k, copy.deepcopy(v, memo))
  File "copy.py", line 161, in deepcopy
    rv = reductor(4)

AttributeError: 'LocalTaskJob' object has no attribute 'task'
  File "airflow", line 8, in <module>
    sys.exit(main())
  File "airflow/__main__.py", line 40, in main
    args.func(args)
  File "airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "airflow/utils/cli.py", line 91, in wrapper
    return f(*args, **kwargs)
  File "airflow/cli/commands/task_command.py", line 238, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "airflow/cli/commands/task_command.py", line 64, in _run_task_by_selected_method
    _run_task_by_local_task_job(args, ti)
  File "airflow/cli/commands/task_command.py", line 121, in _run_task_by_local_task_job
    run_job.run()
  File "airflow/jobs/base_job.py", line 245, in run
    self._execute()
  File "airflow/jobs/local_task_job.py", line 128, in _execute
    self.handle_task_exit(return_code)
  File "airflow/jobs/local_task_job.py", line 166, in handle_task_exit
    self._run_mini_scheduler_on_child_tasks()
  File "airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "airflow/sentry.py", line 168, in wrapper
    self.add_tagging(task_instance)
  File "airflow/sentry.py", line 119, in add_tagging
    task = task_instance.task
```

### What you expected to happen

The error to be handled correctly and passed on to Sentry without raising another exception within the error handling system

### How to reproduce

In this case we were trying to backfill task for a DAG that at that point had a compilation error. This is quite an edge case yes :-)

### Anything else

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/sentry.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 
19 """Sentry Integration"""
20 import logging
21 from functools import wraps
22 
23 from airflow.configuration import conf
24 from airflow.utils.session import find_session_idx, provide_session
25 from airflow.utils.state import State
26 
27 log = logging.getLogger(__name__)
28 
29 
30 class DummySentry:
31     """Blank class for Sentry."""
32 
33     @classmethod
34     def add_tagging(cls, task_instance):
35         """Blank function for tagging."""
36 
37     @classmethod
38     def add_breadcrumbs(cls, task_instance, session=None):
39         """Blank function for breadcrumbs."""
40 
41     @classmethod
42     def enrich_errors(cls, run):
43         """Blank function for formatting a TaskInstance._run_raw_task."""
44         return run
45 
46     def flush(self):
47         """Blank function for flushing errors."""
48 
49 
50 Sentry: DummySentry = DummySentry()
51 if conf.getboolean("sentry", 'sentry_on', fallback=False):
52     import sentry_sdk
53 
54     # Verify blinker installation
55     from blinker import signal  # noqa: F401
56     from sentry_sdk.integrations.flask import FlaskIntegration
57     from sentry_sdk.integrations.logging import ignore_logger
58 
59     class ConfiguredSentry(DummySentry):
60         """Configure Sentry SDK."""
61 
62         SCOPE_TAGS = frozenset(("task_id", "dag_id", "execution_date", "operator", "try_number"))
63         SCOPE_CRUMBS = frozenset(("task_id", "state", "operator", "duration"))
64 
65         UNSUPPORTED_SENTRY_OPTIONS = frozenset(
66             (
67                 "integrations",
68                 "in_app_include",
69                 "in_app_exclude",
70                 "ignore_errors",
71                 "before_breadcrumb",
72                 "before_send",
73                 "transport",
74             )
75         )
76 
77         def __init__(self):
78             """Initialize the Sentry SDK."""
79             ignore_logger("airflow.task")
80             ignore_logger("airflow.jobs.backfill_job.BackfillJob")
81             executor_name = conf.get("core", "EXECUTOR")
82 
83             sentry_flask = FlaskIntegration()
84 
85             # LoggingIntegration is set by default.
86             integrations = [sentry_flask]
87 
88             if executor_name == "CeleryExecutor":
89                 from sentry_sdk.integrations.celery import CeleryIntegration
90 
91                 sentry_celery = CeleryIntegration()
92                 integrations.append(sentry_celery)
93 
94             dsn = None
95             sentry_config_opts = conf.getsection("sentry") or {}
96             if sentry_config_opts:
97                 sentry_config_opts.pop("sentry_on")
98                 old_way_dsn = sentry_config_opts.pop("sentry_dsn", None)
99                 new_way_dsn = sentry_config_opts.pop("dsn", None)
100                 # supported backward compatibility with old way dsn option
101                 dsn = old_way_dsn or new_way_dsn
102 
103                 unsupported_options = self.UNSUPPORTED_SENTRY_OPTIONS.intersection(sentry_config_opts.keys())
104                 if unsupported_options:
105                     log.warning(
106                         "There are unsupported options in [sentry] section: %s",
107                         ", ".join(unsupported_options),
108                     )
109 
110             if dsn:
111                 sentry_sdk.init(dsn=dsn, integrations=integrations, **sentry_config_opts)
112             else:
113                 # Setting up Sentry using environment variables.
114                 log.debug("Defaulting to SENTRY_DSN in environment.")
115                 sentry_sdk.init(integrations=integrations, **sentry_config_opts)
116 
117         def add_tagging(self, task_instance):
118             """Function to add tagging for a task_instance."""
119             task = task_instance.task
120 
121             with sentry_sdk.configure_scope() as scope:
122                 for tag_name in self.SCOPE_TAGS:
123                     attribute = getattr(task_instance, tag_name)
124                     if tag_name == "operator":
125                         attribute = task.__class__.__name__
126                     scope.set_tag(tag_name, attribute)
127 
128         @provide_session
129         def add_breadcrumbs(self, task_instance, session=None):
130             """Function to add breadcrumbs inside of a task_instance."""
131             if session is None:
132                 return
133             dr = task_instance.get_dagrun(session)
134             task_instances = dr.get_task_instances(
135                 state={State.SUCCESS, State.FAILED},
136                 session=session,
137             )
138 
139             for ti in task_instances:
140                 data = {}
141                 for crumb_tag in self.SCOPE_CRUMBS:
142                     data[crumb_tag] = getattr(ti, crumb_tag)
143 
144                 sentry_sdk.add_breadcrumb(category="completed_tasks", data=data, level="info")
145 
146         def enrich_errors(self, func):
147             """Wrap TaskInstance._run_raw_task to support task specific tags and breadcrumbs."""
148             session_args_idx = find_session_idx(func)
149 
150             @wraps(func)
151             def wrapper(task_instance, *args, **kwargs):
152                 # Wrapping the _run_raw_task function with push_scope to contain
153                 # tags and breadcrumbs to a specific Task Instance
154 
155                 try:
156                     session = kwargs.get('session', args[session_args_idx])
157                 except IndexError:
158                     session = None
159 
160                 with sentry_sdk.push_scope():
161                     try:
162                         return func(task_instance, *args, **kwargs)
163                     except Exception as e:
164                         self.add_tagging(task_instance)
165                         self.add_breadcrumbs(task_instance, session=session)
166                         sentry_sdk.capture_exception(e)
167                         raise
168 
169             return wrapper
170 
171         def flush(self):
172             sentry_sdk.flush()
173 
174     Sentry = ConfiguredSentry()
175
```
--- 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/airflow/sentry.py b/airflow/sentry.py
--- a/airflow/sentry.py
+++ b/airflow/sentry.py
@@ -144,11 +144,14 @@
                 sentry_sdk.add_breadcrumb(category="completed_tasks", data=data, level="info")
 
         def enrich_errors(self, func):
-            """Wrap TaskInstance._run_raw_task to support task specific tags and breadcrumbs."""
+            """
+            Wrap TaskInstance._run_raw_task and LocalTaskJob._run_mini_scheduler_on_child_tasks
+             to support task specific tags and breadcrumbs.
+            """
             session_args_idx = find_session_idx(func)
 
             @wraps(func)
-            def wrapper(task_instance, *args, **kwargs):
+            def wrapper(_self, *args, **kwargs):
                 # Wrapping the _run_raw_task function with push_scope to contain
                 # tags and breadcrumbs to a specific Task Instance
 
@@ -159,8 +162,14 @@
 
                 with sentry_sdk.push_scope():
                     try:
-                        return func(task_instance, *args, **kwargs)
+                        return func(_self, *args, **kwargs)
                     except Exception as e:
+                        # Is a LocalTaskJob get the task instance
+                        if hasattr(_self, 'task_instance'):
+                            task_instance = _self.task_instance
+                        else:
+                            task_instance = _self
+
                         self.add_tagging(task_instance)
                         self.add_breadcrumbs(task_instance, session=session)
                         sentry_sdk.capture_exception(e)

{"golden_diff": "diff --git a/airflow/sentry.py b/airflow/sentry.py\n--- a/airflow/sentry.py\n+++ b/airflow/sentry.py\n@@ -144,11 +144,14 @@\n                 sentry_sdk.add_breadcrumb(category=\"completed_tasks\", data=data, level=\"info\")\n \n         def enrich_errors(self, func):\n-            \"\"\"Wrap TaskInstance._run_raw_task to support task specific tags and breadcrumbs.\"\"\"\n+            \"\"\"\n+            Wrap TaskInstance._run_raw_task and LocalTaskJob._run_mini_scheduler_on_child_tasks\n+             to support task specific tags and breadcrumbs.\n+            \"\"\"\n             session_args_idx = find_session_idx(func)\n \n             @wraps(func)\n-            def wrapper(task_instance, *args, **kwargs):\n+            def wrapper(_self, *args, **kwargs):\n                 # Wrapping the _run_raw_task function with push_scope to contain\n                 # tags and breadcrumbs to a specific Task Instance\n \n@@ -159,8 +162,14 @@\n \n                 with sentry_sdk.push_scope():\n                     try:\n-                        return func(task_instance, *args, **kwargs)\n+                        return func(_self, *args, **kwargs)\n                     except Exception as e:\n+                        # Is a LocalTaskJob get the task instance\n+                        if hasattr(_self, 'task_instance'):\n+                            task_instance = _self.task_instance\n+                        else:\n+                            task_instance = _self\n+\n                         self.add_tagging(task_instance)\n                         self.add_breadcrumbs(task_instance, session=session)\n                         sentry_sdk.capture_exception(e)\n", "issue": "Exception within  LocalTaskJob._run_mini_scheduler_on_child_tasks brakes Sentry Handler\n### Apache Airflow version\r\n\r\n2.1.3 (latest released)\r\n\r\n### Operating System\r\n\r\nDebian GNU/Linux 10 (buster)\r\n\r\n### Versions of Apache Airflow Providers\r\n\r\n```\r\napache-airflow-providers-amazon @ file:///root/.cache/pypoetry/artifacts/7f/f7/23/fc7fd3543aa486275ef0385c29063ff0dc391b0fc95dc5aa6cab2cf4e5/apache_airflow_providers_amazon-2.2.0-py3-none-any.whl\r\napache-airflow-providers-celery @ file:///root/.cache/pypoetry/artifacts/14/80/39/0d9d57205da1d24189ac9c18eb3477664ed2c2618c1467c9809b9a2fbf/apache_airflow_providers_celery-2.0.0-py3-none-any.whl\r\napache-airflow-providers-ftp @ file:///root/.cache/pypoetry/artifacts/a5/13/da/bf14abc40193a1ee1b82bbd800e3ac230427d7684b9d40998ac3684bef/apache_airflow_providers_ftp-2.0.1-py3-none-any.whl\r\napache-airflow-providers-http @ file:///root/.cache/pypoetry/artifacts/fc/d7/d2/73c89ef847bbae1704fa403d7e92dba1feead757aae141613980db40ff/apache_airflow_providers_http-2.0.0-py3-none-any.whl\r\napache-airflow-providers-imap @ file:///root/.cache/pypoetry/artifacts/af/5d/de/21c10bfc7ac076a415dcc3fc909317547e77e38c005487552cf40ddd97/apache_airflow_providers_imap-2.0.1-py3-none-any.whl\r\napache-airflow-providers-postgres @ file:///root/.cache/pypoetry/artifacts/50/27/e0/9b0d8f4c0abf59967bb87a04a93d73896d9a4558994185dd8bc43bb67f/apache_airflow_providers_postgres-2.2.0-py3-none-any.whl\r\napache-airflow-providers-redis @ file:///root/.cache/pypoetry/artifacts/7d/95/03/5d2a65ace88ae9a9ce9134b927b1e9639c8680c13a31e58425deae55d1/apache_airflow_providers_redis-2.0.1-py3-none-any.whl\r\napache-airflow-providers-sqlite @ file:///root/.cache/pypoetry/artifacts/ec/e6/a3/e0d81fef662ccf79609e7d2c4e4440839a464771fd2a002d252c9a401d/apache_airflow_providers_sqlite-2.0.1-py3-none-any.whl\r\n```\r\n\r\n\r\n### Deployment\r\n\r\nOther Docker-based deployment\r\n\r\n### Deployment details\r\n\r\nWe are using the Sentry integration\r\n\r\n### What happened\r\n\r\nAn exception within LocalTaskJobs mini scheduler was handled incorrectly by the Sentry integrations 'enrich_errors' method. This is because it assumes its applied to a method of a TypeInstance task\r\n\r\n```\r\nTypeError: cannot pickle 'dict_keys' object\r\n  File \"airflow/sentry.py\", line 166, in wrapper\r\n    return func(task_instance, *args, **kwargs)\r\n  File \"airflow/jobs/local_task_job.py\", line 241, in _run_mini_scheduler_on_child_tasks\r\n    partial_dag = task.dag.partial_subset(\r\n  File \"airflow/models/dag.py\", line 1487, in partial_subset\r\n    dag.task_dict = {\r\n  File \"airflow/models/dag.py\", line 1488, in <dictcomp>\r\n    t.task_id: copy.deepcopy(t, {id(t.dag): dag})  # type: ignore\r\n  File \"copy.py\", line 153, in deepcopy\r\n    y = copier(memo)\r\n  File \"airflow/models/baseoperator.py\", line 970, in __deepcopy__\r\n    setattr(result, k, copy.deepcopy(v, memo))\r\n  File \"copy.py\", line 161, in deepcopy\r\n    rv = reductor(4)\r\n\r\nAttributeError: 'LocalTaskJob' object has no attribute 'task'\r\n  File \"airflow\", line 8, in <module>\r\n    sys.exit(main())\r\n  File \"airflow/__main__.py\", line 40, in main\r\n    args.func(args)\r\n  File \"airflow/cli/cli_parser.py\", line 48, in command\r\n    return func(*args, **kwargs)\r\n  File \"airflow/utils/cli.py\", line 91, in wrapper\r\n    return f(*args, **kwargs)\r\n  File \"airflow/cli/commands/task_command.py\", line 238, in task_run\r\n    _run_task_by_selected_method(args, dag, ti)\r\n  File \"airflow/cli/commands/task_command.py\", line 64, in _run_task_by_selected_method\r\n    _run_task_by_local_task_job(args, ti)\r\n  File \"airflow/cli/commands/task_command.py\", line 121, in _run_task_by_local_task_job\r\n    run_job.run()\r\n  File \"airflow/jobs/base_job.py\", line 245, in run\r\n    self._execute()\r\n  File \"airflow/jobs/local_task_job.py\", line 128, in _execute\r\n    self.handle_task_exit(return_code)\r\n  File \"airflow/jobs/local_task_job.py\", line 166, in handle_task_exit\r\n    self._run_mini_scheduler_on_child_tasks()\r\n  File \"airflow/utils/session.py\", line 70, in wrapper\r\n    return func(*args, session=session, **kwargs)\r\n  File \"airflow/sentry.py\", line 168, in wrapper\r\n    self.add_tagging(task_instance)\r\n  File \"airflow/sentry.py\", line 119, in add_tagging\r\n    task = task_instance.task\r\n```\r\n\r\n### What you expected to happen\r\n\r\nThe error to be handled correctly and passed on to Sentry without raising another exception within the error handling system\r\n\r\n### How to reproduce\r\n\r\nIn this case we were trying to backfill task for a DAG that at that point had a compilation error. This is quite an edge case yes :-)\r\n\r\n### Anything else\r\n\r\n_No response_\r\n\r\n### Are you willing to submit PR?\r\n\r\n- [X] Yes I am willing to submit a PR!\r\n\r\n### Code of Conduct\r\n\r\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\r\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  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,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\n\"\"\"Sentry Integration\"\"\"\nimport logging\nfrom functools import wraps\n\nfrom airflow.configuration import conf\nfrom airflow.utils.session import find_session_idx, provide_session\nfrom airflow.utils.state import State\n\nlog = logging.getLogger(__name__)\n\n\nclass DummySentry:\n    \"\"\"Blank class for Sentry.\"\"\"\n\n    @classmethod\n    def add_tagging(cls, task_instance):\n        \"\"\"Blank function for tagging.\"\"\"\n\n    @classmethod\n    def add_breadcrumbs(cls, task_instance, session=None):\n        \"\"\"Blank function for breadcrumbs.\"\"\"\n\n    @classmethod\n    def enrich_errors(cls, run):\n        \"\"\"Blank function for formatting a TaskInstance._run_raw_task.\"\"\"\n        return run\n\n    def flush(self):\n        \"\"\"Blank function for flushing errors.\"\"\"\n\n\nSentry: DummySentry = DummySentry()\nif conf.getboolean(\"sentry\", 'sentry_on', fallback=False):\n    import sentry_sdk\n\n    # Verify blinker installation\n    from blinker import signal  # noqa: F401\n    from sentry_sdk.integrations.flask import FlaskIntegration\n    from sentry_sdk.integrations.logging import ignore_logger\n\n    class ConfiguredSentry(DummySentry):\n        \"\"\"Configure Sentry SDK.\"\"\"\n\n        SCOPE_TAGS = frozenset((\"task_id\", \"dag_id\", \"execution_date\", \"operator\", \"try_number\"))\n        SCOPE_CRUMBS = frozenset((\"task_id\", \"state\", \"operator\", \"duration\"))\n\n        UNSUPPORTED_SENTRY_OPTIONS = frozenset(\n            (\n                \"integrations\",\n                \"in_app_include\",\n                \"in_app_exclude\",\n                \"ignore_errors\",\n                \"before_breadcrumb\",\n                \"before_send\",\n                \"transport\",\n            )\n        )\n\n        def __init__(self):\n            \"\"\"Initialize the Sentry SDK.\"\"\"\n            ignore_logger(\"airflow.task\")\n            ignore_logger(\"airflow.jobs.backfill_job.BackfillJob\")\n            executor_name = conf.get(\"core\", \"EXECUTOR\")\n\n            sentry_flask = FlaskIntegration()\n\n            # LoggingIntegration is set by default.\n            integrations = [sentry_flask]\n\n            if executor_name == \"CeleryExecutor\":\n                from sentry_sdk.integrations.celery import CeleryIntegration\n\n                sentry_celery = CeleryIntegration()\n                integrations.append(sentry_celery)\n\n            dsn = None\n            sentry_config_opts = conf.getsection(\"sentry\") or {}\n            if sentry_config_opts:\n                sentry_config_opts.pop(\"sentry_on\")\n                old_way_dsn = sentry_config_opts.pop(\"sentry_dsn\", None)\n                new_way_dsn = sentry_config_opts.pop(\"dsn\", None)\n                # supported backward compatibility with old way dsn option\n                dsn = old_way_dsn or new_way_dsn\n\n                unsupported_options = self.UNSUPPORTED_SENTRY_OPTIONS.intersection(sentry_config_opts.keys())\n                if unsupported_options:\n                    log.warning(\n                        \"There are unsupported options in [sentry] section: %s\",\n                        \", \".join(unsupported_options),\n                    )\n\n            if dsn:\n                sentry_sdk.init(dsn=dsn, integrations=integrations, **sentry_config_opts)\n            else:\n                # Setting up Sentry using environment variables.\n                log.debug(\"Defaulting to SENTRY_DSN in environment.\")\n                sentry_sdk.init(integrations=integrations, **sentry_config_opts)\n\n        def add_tagging(self, task_instance):\n            \"\"\"Function to add tagging for a task_instance.\"\"\"\n            task = task_instance.task\n\n            with sentry_sdk.configure_scope() as scope:\n                for tag_name in self.SCOPE_TAGS:\n                    attribute = getattr(task_instance, tag_name)\n                    if tag_name == \"operator\":\n                        attribute = task.__class__.__name__\n                    scope.set_tag(tag_name, attribute)\n\n        @provide_session\n        def add_breadcrumbs(self, task_instance, session=None):\n            \"\"\"Function to add breadcrumbs inside of a task_instance.\"\"\"\n            if session is None:\n                return\n            dr = task_instance.get_dagrun(session)\n            task_instances = dr.get_task_instances(\n                state={State.SUCCESS, State.FAILED},\n                session=session,\n            )\n\n            for ti in task_instances:\n                data = {}\n                for crumb_tag in self.SCOPE_CRUMBS:\n                    data[crumb_tag] = getattr(ti, crumb_tag)\n\n                sentry_sdk.add_breadcrumb(category=\"completed_tasks\", data=data, level=\"info\")\n\n        def enrich_errors(self, func):\n            \"\"\"Wrap TaskInstance._run_raw_task to support task specific tags and breadcrumbs.\"\"\"\n            session_args_idx = find_session_idx(func)\n\n            @wraps(func)\n            def wrapper(task_instance, *args, **kwargs):\n                # Wrapping the _run_raw_task function with push_scope to contain\n                # tags and breadcrumbs to a specific Task Instance\n\n                try:\n                    session = kwargs.get('session', args[session_args_idx])\n                except IndexError:\n                    session = None\n\n                with sentry_sdk.push_scope():\n                    try:\n                        return func(task_instance, *args, **kwargs)\n                    except Exception as e:\n                        self.add_tagging(task_instance)\n                        self.add_breadcrumbs(task_instance, session=session)\n                        sentry_sdk.capture_exception(e)\n                        raise\n\n            return wrapper\n\n        def flush(self):\n            sentry_sdk.flush()\n\n    Sentry = ConfiguredSentry()\n", "path": "airflow/sentry.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  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,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\n\"\"\"Sentry Integration\"\"\"\nimport logging\nfrom functools import wraps\n\nfrom airflow.configuration import conf\nfrom airflow.utils.session import find_session_idx, provide_session\nfrom airflow.utils.state import State\n\nlog = logging.getLogger(__name__)\n\n\nclass DummySentry:\n    \"\"\"Blank class for Sentry.\"\"\"\n\n    @classmethod\n    def add_tagging(cls, task_instance):\n        \"\"\"Blank function for tagging.\"\"\"\n\n    @classmethod\n    def add_breadcrumbs(cls, task_instance, session=None):\n        \"\"\"Blank function for breadcrumbs.\"\"\"\n\n    @classmethod\n    def enrich_errors(cls, run):\n        \"\"\"Blank function for formatting a TaskInstance._run_raw_task.\"\"\"\n        return run\n\n    def flush(self):\n        \"\"\"Blank function for flushing errors.\"\"\"\n\n\nSentry: DummySentry = DummySentry()\nif conf.getboolean(\"sentry\", 'sentry_on', fallback=False):\n    import sentry_sdk\n\n    # Verify blinker installation\n    from blinker import signal  # noqa: F401\n    from sentry_sdk.integrations.flask import FlaskIntegration\n    from sentry_sdk.integrations.logging import ignore_logger\n\n    class ConfiguredSentry(DummySentry):\n        \"\"\"Configure Sentry SDK.\"\"\"\n\n        SCOPE_TAGS = frozenset((\"task_id\", \"dag_id\", \"execution_date\", \"operator\", \"try_number\"))\n        SCOPE_CRUMBS = frozenset((\"task_id\", \"state\", \"operator\", \"duration\"))\n\n        UNSUPPORTED_SENTRY_OPTIONS = frozenset(\n            (\n                \"integrations\",\n                \"in_app_include\",\n                \"in_app_exclude\",\n                \"ignore_errors\",\n                \"before_breadcrumb\",\n                \"before_send\",\n                \"transport\",\n            )\n        )\n\n        def __init__(self):\n            \"\"\"Initialize the Sentry SDK.\"\"\"\n            ignore_logger(\"airflow.task\")\n            ignore_logger(\"airflow.jobs.backfill_job.BackfillJob\")\n            executor_name = conf.get(\"core\", \"EXECUTOR\")\n\n            sentry_flask = FlaskIntegration()\n\n            # LoggingIntegration is set by default.\n            integrations = [sentry_flask]\n\n            if executor_name == \"CeleryExecutor\":\n                from sentry_sdk.integrations.celery import CeleryIntegration\n\n                sentry_celery = CeleryIntegration()\n                integrations.append(sentry_celery)\n\n            dsn = None\n            sentry_config_opts = conf.getsection(\"sentry\") or {}\n            if sentry_config_opts:\n                sentry_config_opts.pop(\"sentry_on\")\n                old_way_dsn = sentry_config_opts.pop(\"sentry_dsn\", None)\n                new_way_dsn = sentry_config_opts.pop(\"dsn\", None)\n                # supported backward compatibility with old way dsn option\n                dsn = old_way_dsn or new_way_dsn\n\n                unsupported_options = self.UNSUPPORTED_SENTRY_OPTIONS.intersection(sentry_config_opts.keys())\n                if unsupported_options:\n                    log.warning(\n                        \"There are unsupported options in [sentry] section: %s\",\n                        \", \".join(unsupported_options),\n                    )\n\n            if dsn:\n                sentry_sdk.init(dsn=dsn, integrations=integrations, **sentry_config_opts)\n            else:\n                # Setting up Sentry using environment variables.\n                log.debug(\"Defaulting to SENTRY_DSN in environment.\")\n                sentry_sdk.init(integrations=integrations, **sentry_config_opts)\n\n        def add_tagging(self, task_instance):\n            \"\"\"Function to add tagging for a task_instance.\"\"\"\n            task = task_instance.task\n\n            with sentry_sdk.configure_scope() as scope:\n                for tag_name in self.SCOPE_TAGS:\n                    attribute = getattr(task_instance, tag_name)\n                    if tag_name == \"operator\":\n                        attribute = task.__class__.__name__\n                    scope.set_tag(tag_name, attribute)\n\n        @provide_session\n        def add_breadcrumbs(self, task_instance, session=None):\n            \"\"\"Function to add breadcrumbs inside of a task_instance.\"\"\"\n            if session is None:\n                return\n            dr = task_instance.get_dagrun(session)\n            task_instances = dr.get_task_instances(\n                state={State.SUCCESS, State.FAILED},\n                session=session,\n            )\n\n            for ti in task_instances:\n                data = {}\n                for crumb_tag in self.SCOPE_CRUMBS:\n                    data[crumb_tag] = getattr(ti, crumb_tag)\n\n                sentry_sdk.add_breadcrumb(category=\"completed_tasks\", data=data, level=\"info\")\n\n        def enrich_errors(self, func):\n            \"\"\"\n            Wrap TaskInstance._run_raw_task and LocalTaskJob._run_mini_scheduler_on_child_tasks\n             to support task specific tags and breadcrumbs.\n            \"\"\"\n            session_args_idx = find_session_idx(func)\n\n            @wraps(func)\n            def wrapper(_self, *args, **kwargs):\n                # Wrapping the _run_raw_task function with push_scope to contain\n                # tags and breadcrumbs to a specific Task Instance\n\n                try:\n                    session = kwargs.get('session', args[session_args_idx])\n                except IndexError:\n                    session = None\n\n                with sentry_sdk.push_scope():\n                    try:\n                        return func(_self, *args, **kwargs)\n                    except Exception as e:\n                        # Is a LocalTaskJob get the task instance\n                        if hasattr(_self, 'task_instance'):\n                            task_instance = _self.task_instance\n                        else:\n                            task_instance = _self\n\n                        self.add_tagging(task_instance)\n                        self.add_breadcrumbs(task_instance, session=session)\n                        sentry_sdk.capture_exception(e)\n                        raise\n\n            return wrapper\n\n        def flush(self):\n            sentry_sdk.flush()\n\n    Sentry = ConfiguredSentry()\n", "path": "airflow/sentry.py"}]}