We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add proxies to slack_notifier
## Current behavior
`prefect.utilities.notifications.slack_notifier` sends requests to the Slack webhook using `requests.post` without allowing  `proxies={"http": "0.123.456.789"}` to be passed, which is problematic when Internet access must be done through a proxy.

 It is possible to go around this by setting HTTP_PROXY / HTTPS_PROXY environment variables, but this is not ideal / possible when running on private infrastructure which by default does not access the Internet and should not have these variables set.



## Proposed behavior
Add an optional `proxies` argument to `prefect.utilities.notifications.slack_notifier` which gets passed to `requests.post`.



## Example
To use Slack notifications from Prefect running on private infrastructure.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/utilities/notifications/notifications.py`
Content:
```
1 """
2 Tools and utilities for notifications and callbacks.
3 
4 For an in-depth guide to setting up your system for using Slack notifications, [please see our
5 tutorial](/core/advanced_tutorials/slack-notifications.html).
6 """
7 import smtplib
8 from email.header import Header
9 from email.mime.multipart import MIMEMultipart
10 from email.mime.text import MIMEText
11 from typing import TYPE_CHECKING, Any, Callable, Union, cast
12 
13 from toolz import curry
14 
15 import prefect
16 
17 if TYPE_CHECKING:
18     import prefect.engine.state
19     import prefect.client
20     from prefect import Flow, Task  # noqa
21 
22 TrackedObjectType = Union["Flow", "Task"]
23 
24 __all__ = ["callback_factory", "gmail_notifier", "slack_notifier"]
25 
26 
27 def callback_factory(
28     fn: Callable[[Any, "prefect.engine.state.State"], Any],
29     check: Callable[["prefect.engine.state.State"], bool],
30 ) -> Callable:
31     """
32     Utility for generating state handlers that serve as callbacks, under arbitrary
33     state-based checks.
34 
35     Args:
36         - fn (Callable): a function with signature `fn(obj, state: State) -> None`
37             that will be called anytime the associated state-check passes; in general, it is
38             expected that this function will have side effects (e.g., sends an email).  The
39             first argument to this function is the `Task` or `Flow` it is attached to.
40         - check (Callable): a function with signature `check(state: State) -> bool`
41             that is used for determining when the callback function should be called
42 
43     Returns:
44         - state_handler (Callable): a state handler function that can be attached to both Tasks
45             and Flows
46 
47     Example:
48         ```python
49         from prefect import Task, Flow
50         from prefect.utilities.notifications import callback_factory
51 
52         fn = lambda obj, state: print(state)
53         check = lambda state: state.is_successful()
54         callback = callback_factory(fn, check)
55 
56         t = Task(state_handlers=[callback])
57         f = Flow("My Example Flow", tasks=[t], state_handlers=[callback])
58         f.run()
59         # prints:
60         # Success("Task run succeeded.")
61         # Success("All reference tasks succeeded.")
62         ```
63     """
64 
65     def state_handler(
66         obj: Any,
67         old_state: "prefect.engine.state.State",
68         new_state: "prefect.engine.state.State",
69     ) -> "prefect.engine.state.State":
70         if check(new_state) is True:
71             fn(obj, new_state)
72         return new_state
73 
74     return state_handler
75 
76 
77 def email_message_formatter(
78     tracked_obj: TrackedObjectType, state: "prefect.engine.state.State", email_to: str
79 ) -> str:
80     if isinstance(state.result, Exception):
81         msg = "<pre>{}</pre>".format(repr(state.result))
82     else:
83         msg = '"{}"'.format(state.message)
84 
85     html = """
86     <html>
87       <head></head>
88       <body>
89         <table align="left" border="0" cellpadding="2px" cellspacing="2px">
90           <tr>
91             <td style="border-left: 2px solid {color};">
92               <img src="https://emoji.slack-edge.com/TAN3D79AL/prefect/2497370f58500a5a.png">
93             </td>
94             <td style="border-left: 2px solid {color}; padding-left: 6px;">
95               {text}
96             </td>
97           </tr>
98         </table>
99       </body>
100     </html>
101     """
102     color = state.color
103     text = """
104     <pre>{name}</pre> is now in a <font color="{color}"><b>{state}</b></font> state
105     <br><br>
106     Message: {msg}
107     """.format(
108         name=tracked_obj.name, color=state.color, state=type(state).__name__, msg=msg
109     )
110 
111     contents = MIMEMultipart("alternative")
112     contents.attach(MIMEText(text, "plain"))
113     contents.attach(MIMEText(html.format(color=color, text=text), "html"))
114 
115     contents["Subject"] = Header(
116         "Prefect state change notification for {}".format(tracked_obj.name), "UTF-8"
117     )
118     contents["From"] = "notifications@prefect.io"
119     contents["To"] = email_to
120 
121     return contents.as_string()
122 
123 
124 def slack_message_formatter(
125     tracked_obj: TrackedObjectType,
126     state: "prefect.engine.state.State",
127     backend_info: bool = True,
128 ) -> dict:
129     # see https://api.slack.com/docs/message-attachments
130     fields = []
131     if isinstance(state.result, Exception):
132         value = "```{}```".format(repr(state.result))
133     else:
134         value = cast(str, state.message)
135     if value is not None:
136         fields.append({"title": "Message", "value": value, "short": False})
137 
138     notification_payload = {
139         "fallback": "State change notification",
140         "color": state.color,
141         "author_name": "Prefect",
142         "author_link": "https://www.prefect.io/",
143         "author_icon": "https://emoji.slack-edge.com/TAN3D79AL/prefect/2497370f58500a5a.png",
144         "title": type(state).__name__,
145         "fields": fields,
146         #                "title_link": "https://www.prefect.io/",
147         "text": "{0} is now in a {1} state".format(
148             tracked_obj.name, type(state).__name__
149         ),
150         "footer": "Prefect notification",
151     }
152 
153     if backend_info and prefect.context.get("flow_run_id"):
154         url = None
155 
156         if isinstance(tracked_obj, prefect.Flow):
157             url = prefect.client.Client().get_cloud_url(
158                 "flow-run", prefect.context["flow_run_id"], as_user=False
159             )
160         elif isinstance(tracked_obj, prefect.Task):
161             url = prefect.client.Client().get_cloud_url(
162                 "task-run", prefect.context.get("task_run_id", ""), as_user=False
163             )
164 
165         if url:
166             notification_payload.update(title_link=url)
167 
168     data = {"attachments": [notification_payload]}
169     return data
170 
171 
172 @curry
173 def gmail_notifier(
174     tracked_obj: TrackedObjectType,
175     old_state: "prefect.engine.state.State",
176     new_state: "prefect.engine.state.State",
177     ignore_states: list = None,
178     only_states: list = None,
179 ) -> "prefect.engine.state.State":
180     """
181     Email state change handler - configured to work solely with Gmail; works as a standalone
182     state handler, or can be called from within a custom state handler.  This function is
183     curried meaning that it can be called multiple times to partially bind any keyword
184     arguments (see example below).
185 
186     The username and password Gmail credentials will be taken from your `"EMAIL_USERNAME"` and
187     `"EMAIL_PASSWORD"` secrets, respectively; note the username will also serve as the
188     destination email address for the notification.
189 
190     Args:
191         - tracked_obj (Task or Flow): Task or Flow object the handler is registered with
192         - old_state (State): previous state of tracked object
193         - new_state (State): new state of tracked object
194         - ignore_states ([State], optional): list of `State` classes to ignore, e.g.,
195             `[Running, Scheduled]`. If `new_state` is an instance of one of the passed states,
196             no notification will occur.
197         - only_states ([State], optional): similar to `ignore_states`, but instead _only_
198             notifies you if the Task / Flow is in a state from the provided list of `State`
199             classes
200 
201     Returns:
202         - State: the `new_state` object that was provided
203 
204     Raises:
205         - ValueError: if the email notification fails for any reason
206 
207     Example:
208         ```python
209         from prefect import task
210         from prefect.utilities.notifications import gmail_notifier
211 
212         @task(state_handlers=[gmail_notifier(ignore_states=[Running])]) # uses currying
213         def add(x, y):
214             return x + y
215         ```
216     """
217     username = cast(str, prefect.client.Secret("EMAIL_USERNAME").get())
218     password = cast(str, prefect.client.Secret("EMAIL_PASSWORD").get())
219     ignore_states = ignore_states or []
220     only_states = only_states or []
221 
222     if any(isinstance(new_state, ignored) for ignored in ignore_states):
223         return new_state
224 
225     if only_states and not any(
226         [isinstance(new_state, included) for included in only_states]
227     ):
228         return new_state
229 
230     body = email_message_formatter(tracked_obj, new_state, username)
231 
232     server = smtplib.SMTP_SSL("smtp.gmail.com", 465)
233     server.login(username, password)
234     try:
235         server.sendmail("notifications@prefect.io", username, body)
236     except Exception as exc:
237         raise ValueError(
238             "Email notification for {} failed".format(tracked_obj)
239         ) from exc
240     finally:
241         server.quit()
242 
243     return new_state
244 
245 
246 @curry
247 def slack_notifier(
248     tracked_obj: TrackedObjectType,
249     old_state: "prefect.engine.state.State",
250     new_state: "prefect.engine.state.State",
251     ignore_states: list = None,
252     only_states: list = None,
253     webhook_secret: str = None,
254     backend_info: bool = True,
255 ) -> "prefect.engine.state.State":
256     """
257     Slack state change handler; requires having the Prefect slack app installed.  Works as a
258     standalone state handler, or can be called from within a custom state handler.  This
259     function is curried meaning that it can be called multiple times to partially bind any
260     keyword arguments (see example below).
261 
262     Args:
263         - tracked_obj (Task or Flow): Task or Flow object the handler is
264             registered with
265         - old_state (State): previous state of tracked object
266         - new_state (State): new state of tracked object
267         - ignore_states ([State], optional): list of `State` classes to ignore, e.g.,
268             `[Running, Scheduled]`. If `new_state` is an instance of one of the passed states,
269             no notification will occur.
270         - only_states ([State], optional): similar to `ignore_states`, but instead _only_
271             notifies you if the Task / Flow is in a state from the provided list of `State`
272             classes
273         - webhook_secret (str, optional): the name of the Prefect Secret that stores your slack
274             webhook URL; defaults to `"SLACK_WEBHOOK_URL"`
275         - backend_info (bool, optional): Whether to supply slack notification with urls
276             pointing to backend pages; defaults to True
277 
278     Returns:
279         - State: the `new_state` object that was provided
280 
281     Raises:
282         - ValueError: if the slack notification fails for any reason
283 
284     Example:
285         ```python
286         from prefect import task
287         from prefect.utilities.notifications import slack_notifier
288 
289         @task(state_handlers=[slack_notifier(ignore_states=[Running])]) # uses currying
290         def add(x, y):
291             return x + y
292         ```
293     """
294     webhook_url = cast(
295         str, prefect.client.Secret(webhook_secret or "SLACK_WEBHOOK_URL").get()
296     )
297     ignore_states = ignore_states or []
298     only_states = only_states or []
299 
300     if any(isinstance(new_state, ignored) for ignored in ignore_states):
301         return new_state
302 
303     if only_states and not any(
304         [isinstance(new_state, included) for included in only_states]
305     ):
306         return new_state
307 
308     # 'import requests' is expensive time-wise, we should do this just-in-time to keep
309     # the 'import prefect' time low
310     import requests
311 
312     form_data = slack_message_formatter(tracked_obj, new_state, backend_info)
313     r = requests.post(webhook_url, json=form_data)
314     if not r.ok:
315         raise ValueError("Slack notification for {} failed".format(tracked_obj))
316     return new_state
317
```
--- 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/src/prefect/utilities/notifications/notifications.py b/src/prefect/utilities/notifications/notifications.py
--- a/src/prefect/utilities/notifications/notifications.py
+++ b/src/prefect/utilities/notifications/notifications.py
@@ -252,6 +252,7 @@
     only_states: list = None,
     webhook_secret: str = None,
     backend_info: bool = True,
+    proxies: dict = None,
 ) -> "prefect.engine.state.State":
     """
     Slack state change handler; requires having the Prefect slack app installed.  Works as a
@@ -274,6 +275,9 @@
             webhook URL; defaults to `"SLACK_WEBHOOK_URL"`
         - backend_info (bool, optional): Whether to supply slack notification with urls
             pointing to backend pages; defaults to True
+        - proxies (dict), optional): `dict` with "http" and/or "https" keys, passed to
+         `requests.post` - for situations where a proxy is required to send requests to the
+          Slack webhook
 
     Returns:
         - State: the `new_state` object that was provided
@@ -310,7 +314,7 @@
     import requests
 
     form_data = slack_message_formatter(tracked_obj, new_state, backend_info)
-    r = requests.post(webhook_url, json=form_data)
+    r = requests.post(webhook_url, json=form_data, proxies=proxies)
     if not r.ok:
         raise ValueError("Slack notification for {} failed".format(tracked_obj))
     return new_state

{"golden_diff": "diff --git a/src/prefect/utilities/notifications/notifications.py b/src/prefect/utilities/notifications/notifications.py\n--- a/src/prefect/utilities/notifications/notifications.py\n+++ b/src/prefect/utilities/notifications/notifications.py\n@@ -252,6 +252,7 @@\n     only_states: list = None,\n     webhook_secret: str = None,\n     backend_info: bool = True,\n+    proxies: dict = None,\n ) -> \"prefect.engine.state.State\":\n     \"\"\"\n     Slack state change handler; requires having the Prefect slack app installed.  Works as a\n@@ -274,6 +275,9 @@\n             webhook URL; defaults to `\"SLACK_WEBHOOK_URL\"`\n         - backend_info (bool, optional): Whether to supply slack notification with urls\n             pointing to backend pages; defaults to True\n+        - proxies (dict), optional): `dict` with \"http\" and/or \"https\" keys, passed to\n+         `requests.post` - for situations where a proxy is required to send requests to the\n+          Slack webhook\n \n     Returns:\n         - State: the `new_state` object that was provided\n@@ -310,7 +314,7 @@\n     import requests\n \n     form_data = slack_message_formatter(tracked_obj, new_state, backend_info)\n-    r = requests.post(webhook_url, json=form_data)\n+    r = requests.post(webhook_url, json=form_data, proxies=proxies)\n     if not r.ok:\n         raise ValueError(\"Slack notification for {} failed\".format(tracked_obj))\n     return new_state\n", "issue": "Add proxies to slack_notifier\n## Current behavior\r\n`prefect.utilities.notifications.slack_notifier` sends requests to the Slack webhook using `requests.post` without allowing  `proxies={\"http\": \"0.123.456.789\"}` to be passed, which is problematic when Internet access must be done through a proxy.\r\n\r\n It is possible to go around this by setting HTTP_PROXY / HTTPS_PROXY environment variables, but this is not ideal / possible when running on private infrastructure which by default does not access the Internet and should not have these variables set.\r\n\r\n\r\n\r\n## Proposed behavior\r\nAdd an optional `proxies` argument to `prefect.utilities.notifications.slack_notifier` which gets passed to `requests.post`.\r\n\r\n\r\n\r\n## Example\r\nTo use Slack notifications from Prefect running on private infrastructure.\n", "before_files": [{"content": "\"\"\"\nTools and utilities for notifications and callbacks.\n\nFor an in-depth guide to setting up your system for using Slack notifications, [please see our\ntutorial](/core/advanced_tutorials/slack-notifications.html).\n\"\"\"\nimport smtplib\nfrom email.header import Header\nfrom email.mime.multipart import MIMEMultipart\nfrom email.mime.text import MIMEText\nfrom typing import TYPE_CHECKING, Any, Callable, Union, cast\n\nfrom toolz import curry\n\nimport prefect\n\nif TYPE_CHECKING:\n    import prefect.engine.state\n    import prefect.client\n    from prefect import Flow, Task  # noqa\n\nTrackedObjectType = Union[\"Flow\", \"Task\"]\n\n__all__ = [\"callback_factory\", \"gmail_notifier\", \"slack_notifier\"]\n\n\ndef callback_factory(\n    fn: Callable[[Any, \"prefect.engine.state.State\"], Any],\n    check: Callable[[\"prefect.engine.state.State\"], bool],\n) -> Callable:\n    \"\"\"\n    Utility for generating state handlers that serve as callbacks, under arbitrary\n    state-based checks.\n\n    Args:\n        - fn (Callable): a function with signature `fn(obj, state: State) -> None`\n            that will be called anytime the associated state-check passes; in general, it is\n            expected that this function will have side effects (e.g., sends an email).  The\n            first argument to this function is the `Task` or `Flow` it is attached to.\n        - check (Callable): a function with signature `check(state: State) -> bool`\n            that is used for determining when the callback function should be called\n\n    Returns:\n        - state_handler (Callable): a state handler function that can be attached to both Tasks\n            and Flows\n\n    Example:\n        ```python\n        from prefect import Task, Flow\n        from prefect.utilities.notifications import callback_factory\n\n        fn = lambda obj, state: print(state)\n        check = lambda state: state.is_successful()\n        callback = callback_factory(fn, check)\n\n        t = Task(state_handlers=[callback])\n        f = Flow(\"My Example Flow\", tasks=[t], state_handlers=[callback])\n        f.run()\n        # prints:\n        # Success(\"Task run succeeded.\")\n        # Success(\"All reference tasks succeeded.\")\n        ```\n    \"\"\"\n\n    def state_handler(\n        obj: Any,\n        old_state: \"prefect.engine.state.State\",\n        new_state: \"prefect.engine.state.State\",\n    ) -> \"prefect.engine.state.State\":\n        if check(new_state) is True:\n            fn(obj, new_state)\n        return new_state\n\n    return state_handler\n\n\ndef email_message_formatter(\n    tracked_obj: TrackedObjectType, state: \"prefect.engine.state.State\", email_to: str\n) -> str:\n    if isinstance(state.result, Exception):\n        msg = \"<pre>{}</pre>\".format(repr(state.result))\n    else:\n        msg = '\"{}\"'.format(state.message)\n\n    html = \"\"\"\n    <html>\n      <head></head>\n      <body>\n        <table align=\"left\" border=\"0\" cellpadding=\"2px\" cellspacing=\"2px\">\n          <tr>\n            <td style=\"border-left: 2px solid {color};\">\n              <img src=\"https://emoji.slack-edge.com/TAN3D79AL/prefect/2497370f58500a5a.png\">\n            </td>\n            <td style=\"border-left: 2px solid {color}; padding-left: 6px;\">\n              {text}\n            </td>\n          </tr>\n        </table>\n      </body>\n    </html>\n    \"\"\"\n    color = state.color\n    text = \"\"\"\n    <pre>{name}</pre> is now in a <font color=\"{color}\"><b>{state}</b></font> state\n    <br><br>\n    Message: {msg}\n    \"\"\".format(\n        name=tracked_obj.name, color=state.color, state=type(state).__name__, msg=msg\n    )\n\n    contents = MIMEMultipart(\"alternative\")\n    contents.attach(MIMEText(text, \"plain\"))\n    contents.attach(MIMEText(html.format(color=color, text=text), \"html\"))\n\n    contents[\"Subject\"] = Header(\n        \"Prefect state change notification for {}\".format(tracked_obj.name), \"UTF-8\"\n    )\n    contents[\"From\"] = \"notifications@prefect.io\"\n    contents[\"To\"] = email_to\n\n    return contents.as_string()\n\n\ndef slack_message_formatter(\n    tracked_obj: TrackedObjectType,\n    state: \"prefect.engine.state.State\",\n    backend_info: bool = True,\n) -> dict:\n    # see https://api.slack.com/docs/message-attachments\n    fields = []\n    if isinstance(state.result, Exception):\n        value = \"```{}```\".format(repr(state.result))\n    else:\n        value = cast(str, state.message)\n    if value is not None:\n        fields.append({\"title\": \"Message\", \"value\": value, \"short\": False})\n\n    notification_payload = {\n        \"fallback\": \"State change notification\",\n        \"color\": state.color,\n        \"author_name\": \"Prefect\",\n        \"author_link\": \"https://www.prefect.io/\",\n        \"author_icon\": \"https://emoji.slack-edge.com/TAN3D79AL/prefect/2497370f58500a5a.png\",\n        \"title\": type(state).__name__,\n        \"fields\": fields,\n        #                \"title_link\": \"https://www.prefect.io/\",\n        \"text\": \"{0} is now in a {1} state\".format(\n            tracked_obj.name, type(state).__name__\n        ),\n        \"footer\": \"Prefect notification\",\n    }\n\n    if backend_info and prefect.context.get(\"flow_run_id\"):\n        url = None\n\n        if isinstance(tracked_obj, prefect.Flow):\n            url = prefect.client.Client().get_cloud_url(\n                \"flow-run\", prefect.context[\"flow_run_id\"], as_user=False\n            )\n        elif isinstance(tracked_obj, prefect.Task):\n            url = prefect.client.Client().get_cloud_url(\n                \"task-run\", prefect.context.get(\"task_run_id\", \"\"), as_user=False\n            )\n\n        if url:\n            notification_payload.update(title_link=url)\n\n    data = {\"attachments\": [notification_payload]}\n    return data\n\n\n@curry\ndef gmail_notifier(\n    tracked_obj: TrackedObjectType,\n    old_state: \"prefect.engine.state.State\",\n    new_state: \"prefect.engine.state.State\",\n    ignore_states: list = None,\n    only_states: list = None,\n) -> \"prefect.engine.state.State\":\n    \"\"\"\n    Email state change handler - configured to work solely with Gmail; works as a standalone\n    state handler, or can be called from within a custom state handler.  This function is\n    curried meaning that it can be called multiple times to partially bind any keyword\n    arguments (see example below).\n\n    The username and password Gmail credentials will be taken from your `\"EMAIL_USERNAME\"` and\n    `\"EMAIL_PASSWORD\"` secrets, respectively; note the username will also serve as the\n    destination email address for the notification.\n\n    Args:\n        - tracked_obj (Task or Flow): Task or Flow object the handler is registered with\n        - old_state (State): previous state of tracked object\n        - new_state (State): new state of tracked object\n        - ignore_states ([State], optional): list of `State` classes to ignore, e.g.,\n            `[Running, Scheduled]`. If `new_state` is an instance of one of the passed states,\n            no notification will occur.\n        - only_states ([State], optional): similar to `ignore_states`, but instead _only_\n            notifies you if the Task / Flow is in a state from the provided list of `State`\n            classes\n\n    Returns:\n        - State: the `new_state` object that was provided\n\n    Raises:\n        - ValueError: if the email notification fails for any reason\n\n    Example:\n        ```python\n        from prefect import task\n        from prefect.utilities.notifications import gmail_notifier\n\n        @task(state_handlers=[gmail_notifier(ignore_states=[Running])]) # uses currying\n        def add(x, y):\n            return x + y\n        ```\n    \"\"\"\n    username = cast(str, prefect.client.Secret(\"EMAIL_USERNAME\").get())\n    password = cast(str, prefect.client.Secret(\"EMAIL_PASSWORD\").get())\n    ignore_states = ignore_states or []\n    only_states = only_states or []\n\n    if any(isinstance(new_state, ignored) for ignored in ignore_states):\n        return new_state\n\n    if only_states and not any(\n        [isinstance(new_state, included) for included in only_states]\n    ):\n        return new_state\n\n    body = email_message_formatter(tracked_obj, new_state, username)\n\n    server = smtplib.SMTP_SSL(\"smtp.gmail.com\", 465)\n    server.login(username, password)\n    try:\n        server.sendmail(\"notifications@prefect.io\", username, body)\n    except Exception as exc:\n        raise ValueError(\n            \"Email notification for {} failed\".format(tracked_obj)\n        ) from exc\n    finally:\n        server.quit()\n\n    return new_state\n\n\n@curry\ndef slack_notifier(\n    tracked_obj: TrackedObjectType,\n    old_state: \"prefect.engine.state.State\",\n    new_state: \"prefect.engine.state.State\",\n    ignore_states: list = None,\n    only_states: list = None,\n    webhook_secret: str = None,\n    backend_info: bool = True,\n) -> \"prefect.engine.state.State\":\n    \"\"\"\n    Slack state change handler; requires having the Prefect slack app installed.  Works as a\n    standalone state handler, or can be called from within a custom state handler.  This\n    function is curried meaning that it can be called multiple times to partially bind any\n    keyword arguments (see example below).\n\n    Args:\n        - tracked_obj (Task or Flow): Task or Flow object the handler is\n            registered with\n        - old_state (State): previous state of tracked object\n        - new_state (State): new state of tracked object\n        - ignore_states ([State], optional): list of `State` classes to ignore, e.g.,\n            `[Running, Scheduled]`. If `new_state` is an instance of one of the passed states,\n            no notification will occur.\n        - only_states ([State], optional): similar to `ignore_states`, but instead _only_\n            notifies you if the Task / Flow is in a state from the provided list of `State`\n            classes\n        - webhook_secret (str, optional): the name of the Prefect Secret that stores your slack\n            webhook URL; defaults to `\"SLACK_WEBHOOK_URL\"`\n        - backend_info (bool, optional): Whether to supply slack notification with urls\n            pointing to backend pages; defaults to True\n\n    Returns:\n        - State: the `new_state` object that was provided\n\n    Raises:\n        - ValueError: if the slack notification fails for any reason\n\n    Example:\n        ```python\n        from prefect import task\n        from prefect.utilities.notifications import slack_notifier\n\n        @task(state_handlers=[slack_notifier(ignore_states=[Running])]) # uses currying\n        def add(x, y):\n            return x + y\n        ```\n    \"\"\"\n    webhook_url = cast(\n        str, prefect.client.Secret(webhook_secret or \"SLACK_WEBHOOK_URL\").get()\n    )\n    ignore_states = ignore_states or []\n    only_states = only_states or []\n\n    if any(isinstance(new_state, ignored) for ignored in ignore_states):\n        return new_state\n\n    if only_states and not any(\n        [isinstance(new_state, included) for included in only_states]\n    ):\n        return new_state\n\n    # 'import requests' is expensive time-wise, we should do this just-in-time to keep\n    # the 'import prefect' time low\n    import requests\n\n    form_data = slack_message_formatter(tracked_obj, new_state, backend_info)\n    r = requests.post(webhook_url, json=form_data)\n    if not r.ok:\n        raise ValueError(\"Slack notification for {} failed\".format(tracked_obj))\n    return new_state\n", "path": "src/prefect/utilities/notifications/notifications.py"}], "after_files": [{"content": "\"\"\"\nTools and utilities for notifications and callbacks.\n\nFor an in-depth guide to setting up your system for using Slack notifications, [please see our\ntutorial](/core/advanced_tutorials/slack-notifications.html).\n\"\"\"\nimport smtplib\nfrom email.header import Header\nfrom email.mime.multipart import MIMEMultipart\nfrom email.mime.text import MIMEText\nfrom typing import TYPE_CHECKING, Any, Callable, Union, cast\n\nfrom toolz import curry\n\nimport prefect\n\nif TYPE_CHECKING:\n    import prefect.engine.state\n    import prefect.client\n    from prefect import Flow, Task  # noqa\n\nTrackedObjectType = Union[\"Flow\", \"Task\"]\n\n__all__ = [\"callback_factory\", \"gmail_notifier\", \"slack_notifier\"]\n\n\ndef callback_factory(\n    fn: Callable[[Any, \"prefect.engine.state.State\"], Any],\n    check: Callable[[\"prefect.engine.state.State\"], bool],\n) -> Callable:\n    \"\"\"\n    Utility for generating state handlers that serve as callbacks, under arbitrary\n    state-based checks.\n\n    Args:\n        - fn (Callable): a function with signature `fn(obj, state: State) -> None`\n            that will be called anytime the associated state-check passes; in general, it is\n            expected that this function will have side effects (e.g., sends an email).  The\n            first argument to this function is the `Task` or `Flow` it is attached to.\n        - check (Callable): a function with signature `check(state: State) -> bool`\n            that is used for determining when the callback function should be called\n\n    Returns:\n        - state_handler (Callable): a state handler function that can be attached to both Tasks\n            and Flows\n\n    Example:\n        ```python\n        from prefect import Task, Flow\n        from prefect.utilities.notifications import callback_factory\n\n        fn = lambda obj, state: print(state)\n        check = lambda state: state.is_successful()\n        callback = callback_factory(fn, check)\n\n        t = Task(state_handlers=[callback])\n        f = Flow(\"My Example Flow\", tasks=[t], state_handlers=[callback])\n        f.run()\n        # prints:\n        # Success(\"Task run succeeded.\")\n        # Success(\"All reference tasks succeeded.\")\n        ```\n    \"\"\"\n\n    def state_handler(\n        obj: Any,\n        old_state: \"prefect.engine.state.State\",\n        new_state: \"prefect.engine.state.State\",\n    ) -> \"prefect.engine.state.State\":\n        if check(new_state) is True:\n            fn(obj, new_state)\n        return new_state\n\n    return state_handler\n\n\ndef email_message_formatter(\n    tracked_obj: TrackedObjectType, state: \"prefect.engine.state.State\", email_to: str\n) -> str:\n    if isinstance(state.result, Exception):\n        msg = \"<pre>{}</pre>\".format(repr(state.result))\n    else:\n        msg = '\"{}\"'.format(state.message)\n\n    html = \"\"\"\n    <html>\n      <head></head>\n      <body>\n        <table align=\"left\" border=\"0\" cellpadding=\"2px\" cellspacing=\"2px\">\n          <tr>\n            <td style=\"border-left: 2px solid {color};\">\n              <img src=\"https://emoji.slack-edge.com/TAN3D79AL/prefect/2497370f58500a5a.png\">\n            </td>\n            <td style=\"border-left: 2px solid {color}; padding-left: 6px;\">\n              {text}\n            </td>\n          </tr>\n        </table>\n      </body>\n    </html>\n    \"\"\"\n    color = state.color\n    text = \"\"\"\n    <pre>{name}</pre> is now in a <font color=\"{color}\"><b>{state}</b></font> state\n    <br><br>\n    Message: {msg}\n    \"\"\".format(\n        name=tracked_obj.name, color=state.color, state=type(state).__name__, msg=msg\n    )\n\n    contents = MIMEMultipart(\"alternative\")\n    contents.attach(MIMEText(text, \"plain\"))\n    contents.attach(MIMEText(html.format(color=color, text=text), \"html\"))\n\n    contents[\"Subject\"] = Header(\n        \"Prefect state change notification for {}\".format(tracked_obj.name), \"UTF-8\"\n    )\n    contents[\"From\"] = \"notifications@prefect.io\"\n    contents[\"To\"] = email_to\n\n    return contents.as_string()\n\n\ndef slack_message_formatter(\n    tracked_obj: TrackedObjectType,\n    state: \"prefect.engine.state.State\",\n    backend_info: bool = True,\n) -> dict:\n    # see https://api.slack.com/docs/message-attachments\n    fields = []\n    if isinstance(state.result, Exception):\n        value = \"```{}```\".format(repr(state.result))\n    else:\n        value = cast(str, state.message)\n    if value is not None:\n        fields.append({\"title\": \"Message\", \"value\": value, \"short\": False})\n\n    notification_payload = {\n        \"fallback\": \"State change notification\",\n        \"color\": state.color,\n        \"author_name\": \"Prefect\",\n        \"author_link\": \"https://www.prefect.io/\",\n        \"author_icon\": \"https://emoji.slack-edge.com/TAN3D79AL/prefect/2497370f58500a5a.png\",\n        \"title\": type(state).__name__,\n        \"fields\": fields,\n        #                \"title_link\": \"https://www.prefect.io/\",\n        \"text\": \"{0} is now in a {1} state\".format(\n            tracked_obj.name, type(state).__name__\n        ),\n        \"footer\": \"Prefect notification\",\n    }\n\n    if backend_info and prefect.context.get(\"flow_run_id\"):\n        url = None\n\n        if isinstance(tracked_obj, prefect.Flow):\n            url = prefect.client.Client().get_cloud_url(\n                \"flow-run\", prefect.context[\"flow_run_id\"], as_user=False\n            )\n        elif isinstance(tracked_obj, prefect.Task):\n            url = prefect.client.Client().get_cloud_url(\n                \"task-run\", prefect.context.get(\"task_run_id\", \"\"), as_user=False\n            )\n\n        if url:\n            notification_payload.update(title_link=url)\n\n    data = {\"attachments\": [notification_payload]}\n    return data\n\n\n@curry\ndef gmail_notifier(\n    tracked_obj: TrackedObjectType,\n    old_state: \"prefect.engine.state.State\",\n    new_state: \"prefect.engine.state.State\",\n    ignore_states: list = None,\n    only_states: list = None,\n) -> \"prefect.engine.state.State\":\n    \"\"\"\n    Email state change handler - configured to work solely with Gmail; works as a standalone\n    state handler, or can be called from within a custom state handler.  This function is\n    curried meaning that it can be called multiple times to partially bind any keyword\n    arguments (see example below).\n\n    The username and password Gmail credentials will be taken from your `\"EMAIL_USERNAME\"` and\n    `\"EMAIL_PASSWORD\"` secrets, respectively; note the username will also serve as the\n    destination email address for the notification.\n\n    Args:\n        - tracked_obj (Task or Flow): Task or Flow object the handler is registered with\n        - old_state (State): previous state of tracked object\n        - new_state (State): new state of tracked object\n        - ignore_states ([State], optional): list of `State` classes to ignore, e.g.,\n            `[Running, Scheduled]`. If `new_state` is an instance of one of the passed states,\n            no notification will occur.\n        - only_states ([State], optional): similar to `ignore_states`, but instead _only_\n            notifies you if the Task / Flow is in a state from the provided list of `State`\n            classes\n\n    Returns:\n        - State: the `new_state` object that was provided\n\n    Raises:\n        - ValueError: if the email notification fails for any reason\n\n    Example:\n        ```python\n        from prefect import task\n        from prefect.utilities.notifications import gmail_notifier\n\n        @task(state_handlers=[gmail_notifier(ignore_states=[Running])]) # uses currying\n        def add(x, y):\n            return x + y\n        ```\n    \"\"\"\n    username = cast(str, prefect.client.Secret(\"EMAIL_USERNAME\").get())\n    password = cast(str, prefect.client.Secret(\"EMAIL_PASSWORD\").get())\n    ignore_states = ignore_states or []\n    only_states = only_states or []\n\n    if any(isinstance(new_state, ignored) for ignored in ignore_states):\n        return new_state\n\n    if only_states and not any(\n        [isinstance(new_state, included) for included in only_states]\n    ):\n        return new_state\n\n    body = email_message_formatter(tracked_obj, new_state, username)\n\n    server = smtplib.SMTP_SSL(\"smtp.gmail.com\", 465)\n    server.login(username, password)\n    try:\n        server.sendmail(\"notifications@prefect.io\", username, body)\n    except Exception as exc:\n        raise ValueError(\n            \"Email notification for {} failed\".format(tracked_obj)\n        ) from exc\n    finally:\n        server.quit()\n\n    return new_state\n\n\n@curry\ndef slack_notifier(\n    tracked_obj: TrackedObjectType,\n    old_state: \"prefect.engine.state.State\",\n    new_state: \"prefect.engine.state.State\",\n    ignore_states: list = None,\n    only_states: list = None,\n    webhook_secret: str = None,\n    backend_info: bool = True,\n    proxies: dict = None,\n) -> \"prefect.engine.state.State\":\n    \"\"\"\n    Slack state change handler; requires having the Prefect slack app installed.  Works as a\n    standalone state handler, or can be called from within a custom state handler.  This\n    function is curried meaning that it can be called multiple times to partially bind any\n    keyword arguments (see example below).\n\n    Args:\n        - tracked_obj (Task or Flow): Task or Flow object the handler is\n            registered with\n        - old_state (State): previous state of tracked object\n        - new_state (State): new state of tracked object\n        - ignore_states ([State], optional): list of `State` classes to ignore, e.g.,\n            `[Running, Scheduled]`. If `new_state` is an instance of one of the passed states,\n            no notification will occur.\n        - only_states ([State], optional): similar to `ignore_states`, but instead _only_\n            notifies you if the Task / Flow is in a state from the provided list of `State`\n            classes\n        - webhook_secret (str, optional): the name of the Prefect Secret that stores your slack\n            webhook URL; defaults to `\"SLACK_WEBHOOK_URL\"`\n        - backend_info (bool, optional): Whether to supply slack notification with urls\n            pointing to backend pages; defaults to True\n        - proxies (dict), optional): `dict` with \"http\" and/or \"https\" keys, passed to\n         `requests.post` - for situations where a proxy is required to send requests to the\n          Slack webhook\n\n    Returns:\n        - State: the `new_state` object that was provided\n\n    Raises:\n        - ValueError: if the slack notification fails for any reason\n\n    Example:\n        ```python\n        from prefect import task\n        from prefect.utilities.notifications import slack_notifier\n\n        @task(state_handlers=[slack_notifier(ignore_states=[Running])]) # uses currying\n        def add(x, y):\n            return x + y\n        ```\n    \"\"\"\n    webhook_url = cast(\n        str, prefect.client.Secret(webhook_secret or \"SLACK_WEBHOOK_URL\").get()\n    )\n    ignore_states = ignore_states or []\n    only_states = only_states or []\n\n    if any(isinstance(new_state, ignored) for ignored in ignore_states):\n        return new_state\n\n    if only_states and not any(\n        [isinstance(new_state, included) for included in only_states]\n    ):\n        return new_state\n\n    # 'import requests' is expensive time-wise, we should do this just-in-time to keep\n    # the 'import prefect' time low\n    import requests\n\n    form_data = slack_message_formatter(tracked_obj, new_state, backend_info)\n    r = requests.post(webhook_url, json=form_data, proxies=proxies)\n    if not r.ok:\n        raise ValueError(\"Slack notification for {} failed\".format(tracked_obj))\n    return new_state\n", "path": "src/prefect/utilities/notifications/notifications.py"}]}