We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Flow S3ResultHandler Fails for Dask Worker with nthreads > 1
## Description
Specifying S3ResultHandler for a Flow running on Dask worker(s) with nthreads > 1 fails with: `KeyError: 'credential_provider'`, likely due to a race condition in using the global boto3 session (boto3.client) between threads. 

## Expected Behavior
In a multithreaded environment, boto3 recommends creating a session per thread rather than sharing the default boto3 session, i.e. boto3.client. See boto3 documentation at: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/resources.html?highlight=multithreading#multithreading-multiprocessing

This thread in Prefect's Community Slack describes using this session-per-thread approach to successfully fix a similar issue when using boto3 in Prefect tasks: https://prefect-community.slack.com/archives/CM28LL405/p1581434710167100

## Reproduction
A Flow with tasks that can run in parallel (e.g. mapped tasks or different Flow branches) and where the Flow-level result_handler is set to S3ResultHandler should reproduce this behavior.

Full stack trace:
```
February 29th 2020 at 8:09:43am | prefect.CloudTaskRunner
ERROR 
Failed to set task state with error: KeyError('credential_provider')
Traceback (most recent call last):
  File "/opt/conda/lib/python3.7/site-packages/prefect/engine/cloud/task_runner.py", line 117, in call_runner_target_handlers
    cloud_state = prepare_state_for_cloud(new_state)
  File "/opt/conda/lib/python3.7/site-packages/prefect/engine/cloud/utilities.py", line 21, in prepare_state_for_cloud
    res.store_safe_value()
  File "/opt/conda/lib/python3.7/site-packages/prefect/engine/result.py", line 93, in store_safe_value
    value = self.result_handler.write(self.value)
  File "/opt/conda/lib/python3.7/site-packages/prefect/engine/result_handlers/s3_result_handler.py", line 103, in write
    self.client.upload_fileobj(stream, Bucket=self.bucket, Key=uri)
  File "/opt/conda/lib/python3.7/site-packages/prefect/engine/result_handlers/s3_result_handler.py", line 67, in client
    self.initialize_client()
  File "/opt/conda/lib/python3.7/site-packages/prefect/engine/result_handlers/s3_result_handler.py", line 60, in initialize_client
    aws_secret_access_key=aws_secret_access_key,
  File "/opt/conda/lib/python3.7/site-packages/boto3/__init__.py", line 91, in client
    return _get_default_session().client(*args, **kwargs)
  File "/opt/conda/lib/python3.7/site-packages/boto3/session.py", line 263, in client
    aws_session_token=aws_session_token, config=config)
  File "/opt/conda/lib/python3.7/site-packages/botocore/session.py", line 823, in create_client
    credentials = self.get_credentials()
  File "/opt/conda/lib/python3.7/site-packages/botocore/session.py", line 428, in get_credentials
    'credential_provider').load_credentials()
  File "/opt/conda/lib/python3.7/site-packages/botocore/session.py", line 923, in get_component
    del self._deferred[name]
KeyError: 'credential_provider'
```

## Environment
We create a long-running Dask cluster where our Dask workers are started with --nprocs 1 --nthreads 3.

(Thanks to @JLouSRM for identifying this issue and capturing log evidence!)


--- 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/engine/result_handlers/s3_result_handler.py`
Content:
```
1 import base64
2 import io
3 import json
4 import uuid
5 from typing import TYPE_CHECKING, Any
6 
7 import cloudpickle
8 import pendulum
9 
10 from prefect.client import Secret
11 from prefect.engine.result_handlers import ResultHandler
12 
13 if TYPE_CHECKING:
14     import boto3
15 
16 
17 class S3ResultHandler(ResultHandler):
18     """
19     Result Handler for writing to and reading from an AWS S3 Bucket.
20 
21     For authentication, there are two options: you can set a Prefect Secret containing
22     your AWS access keys which will be passed directly to the `boto3` client, or you can
23     [configure your flow's runtime environment](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#guide-configuration)
24     for `boto3`.
25 
26     Args:
27         - bucket (str): the name of the bucket to write to / read from
28         - aws_credentials_secret (str, optional): the name of the Prefect Secret
29             that stores your AWS credentials; this Secret must be a JSON string
30             with two keys: `ACCESS_KEY` and `SECRET_ACCESS_KEY` which will be
31             passed directly to `boto3`.  If not provided, `boto3`
32             will fall back on standard AWS rules for authentication.
33     """
34 
35     def __init__(self, bucket: str, aws_credentials_secret: str = None) -> None:
36         self.bucket = bucket
37         self.aws_credentials_secret = aws_credentials_secret
38         super().__init__()
39 
40     def initialize_client(self) -> None:
41         """
42         Initializes an S3 Client.
43         """
44         import boto3
45 
46         aws_access_key = None
47         aws_secret_access_key = None
48 
49         if self.aws_credentials_secret:
50             aws_credentials = Secret(self.aws_credentials_secret).get()
51             if isinstance(aws_credentials, str):
52                 aws_credentials = json.loads(aws_credentials)
53 
54             aws_access_key = aws_credentials["ACCESS_KEY"]
55             aws_secret_access_key = aws_credentials["SECRET_ACCESS_KEY"]
56 
57         s3_client = boto3.client(
58             "s3",
59             aws_access_key_id=aws_access_key,
60             aws_secret_access_key=aws_secret_access_key,
61         )
62         self.client = s3_client
63 
64     @property
65     def client(self) -> "boto3.client":
66         if not hasattr(self, "_client"):
67             self.initialize_client()
68         return self._client
69 
70     @client.setter
71     def client(self, val: Any) -> None:
72         self._client = val
73 
74     def __getstate__(self) -> dict:
75         state = self.__dict__.copy()
76         if "_client" in state:
77             del state["_client"]
78         return state
79 
80     def __setstate__(self, state: dict) -> None:
81         self.__dict__.update(state)
82 
83     def write(self, result: Any) -> str:
84         """
85         Given a result, writes the result to a location in S3
86         and returns the resulting URI.
87 
88         Args:
89             - result (Any): the written result
90 
91         Returns:
92             - str: the S3 URI
93         """
94         date = pendulum.now("utc").format("Y/M/D")  # type: ignore
95         uri = "{date}/{uuid}.prefect_result".format(date=date, uuid=uuid.uuid4())
96         self.logger.debug("Starting to upload result to {}...".format(uri))
97 
98         ## prepare data
99         binary_data = base64.b64encode(cloudpickle.dumps(result))
100         stream = io.BytesIO(binary_data)
101 
102         ## upload
103         self.client.upload_fileobj(stream, Bucket=self.bucket, Key=uri)
104         self.logger.debug("Finished uploading result to {}.".format(uri))
105         return uri
106 
107     def read(self, uri: str) -> Any:
108         """
109         Given a uri, reads a result from S3, reads it and returns it
110 
111         Args:
112             - uri (str): the S3 URI
113 
114         Returns:
115             - Any: the read result
116         """
117         try:
118             self.logger.debug("Starting to download result from {}...".format(uri))
119             stream = io.BytesIO()
120 
121             ## download
122             self.client.download_fileobj(Bucket=self.bucket, Key=uri, Fileobj=stream)
123             stream.seek(0)
124 
125             try:
126                 return_val = cloudpickle.loads(base64.b64decode(stream.read()))
127             except EOFError:
128                 return_val = None
129             self.logger.debug("Finished downloading result from {}.".format(uri))
130 
131         except Exception as exc:
132             self.logger.exception(
133                 "Unexpected error while reading from result handler: {}".format(
134                     repr(exc)
135                 )
136             )
137             return_val = None
138 
139         return return_val
140
```
--- 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/engine/result_handlers/s3_result_handler.py b/src/prefect/engine/result_handlers/s3_result_handler.py
--- a/src/prefect/engine/result_handlers/s3_result_handler.py
+++ b/src/prefect/engine/result_handlers/s3_result_handler.py
@@ -7,6 +7,7 @@
 import cloudpickle
 import pendulum
 
+import prefect
 from prefect.client import Secret
 from prefect.engine.result_handlers import ResultHandler
 
@@ -54,7 +55,10 @@
             aws_access_key = aws_credentials["ACCESS_KEY"]
             aws_secret_access_key = aws_credentials["SECRET_ACCESS_KEY"]
 
-        s3_client = boto3.client(
+        # use a new boto session when initializing in case we are in a new thread
+        # see https://boto3.amazonaws.com/v1/documentation/api/latest/guide/resources.html?#multithreading-multiprocessing
+        session = boto3.session.Session()
+        s3_client = session.client(
             "s3",
             aws_access_key_id=aws_access_key,
             aws_secret_access_key=aws_secret_access_key,
@@ -63,8 +67,13 @@
 
     @property
     def client(self) -> "boto3.client":
-        if not hasattr(self, "_client"):
+        """
+        Initializes a client if we believe we are in a new thread.
+        We consider ourselves in a new thread if we haven't stored a client yet in the current context.
+        """
+        if not prefect.context.get("boto3client"):
             self.initialize_client()
+            prefect.context["boto3client"] = self._client
         return self._client
 
     @client.setter

{"golden_diff": "diff --git a/src/prefect/engine/result_handlers/s3_result_handler.py b/src/prefect/engine/result_handlers/s3_result_handler.py\n--- a/src/prefect/engine/result_handlers/s3_result_handler.py\n+++ b/src/prefect/engine/result_handlers/s3_result_handler.py\n@@ -7,6 +7,7 @@\n import cloudpickle\n import pendulum\n \n+import prefect\n from prefect.client import Secret\n from prefect.engine.result_handlers import ResultHandler\n \n@@ -54,7 +55,10 @@\n             aws_access_key = aws_credentials[\"ACCESS_KEY\"]\n             aws_secret_access_key = aws_credentials[\"SECRET_ACCESS_KEY\"]\n \n-        s3_client = boto3.client(\n+        # use a new boto session when initializing in case we are in a new thread\n+        # see https://boto3.amazonaws.com/v1/documentation/api/latest/guide/resources.html?#multithreading-multiprocessing\n+        session = boto3.session.Session()\n+        s3_client = session.client(\n             \"s3\",\n             aws_access_key_id=aws_access_key,\n             aws_secret_access_key=aws_secret_access_key,\n@@ -63,8 +67,13 @@\n \n     @property\n     def client(self) -> \"boto3.client\":\n-        if not hasattr(self, \"_client\"):\n+        \"\"\"\n+        Initializes a client if we believe we are in a new thread.\n+        We consider ourselves in a new thread if we haven't stored a client yet in the current context.\n+        \"\"\"\n+        if not prefect.context.get(\"boto3client\"):\n             self.initialize_client()\n+            prefect.context[\"boto3client\"] = self._client\n         return self._client\n \n     @client.setter\n", "issue": "Flow S3ResultHandler Fails for Dask Worker with nthreads > 1\n## Description\r\nSpecifying S3ResultHandler for a Flow running on Dask worker(s) with nthreads > 1 fails with: `KeyError: 'credential_provider'`, likely due to a race condition in using the global boto3 session (boto3.client) between threads. \r\n\r\n## Expected Behavior\r\nIn a multithreaded environment, boto3 recommends creating a session per thread rather than sharing the default boto3 session, i.e. boto3.client. See boto3 documentation at: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/resources.html?highlight=multithreading#multithreading-multiprocessing\r\n\r\nThis thread in Prefect's Community Slack describes using this session-per-thread approach to successfully fix a similar issue when using boto3 in Prefect tasks: https://prefect-community.slack.com/archives/CM28LL405/p1581434710167100\r\n\r\n## Reproduction\r\nA Flow with tasks that can run in parallel (e.g. mapped tasks or different Flow branches) and where the Flow-level result_handler is set to S3ResultHandler should reproduce this behavior.\r\n\r\nFull stack trace:\r\n```\r\nFebruary 29th 2020 at 8:09:43am | prefect.CloudTaskRunner\r\nERROR \r\nFailed to set task state with error: KeyError('credential_provider')\r\nTraceback (most recent call last):\r\n  File \"/opt/conda/lib/python3.7/site-packages/prefect/engine/cloud/task_runner.py\", line 117, in call_runner_target_handlers\r\n    cloud_state = prepare_state_for_cloud(new_state)\r\n  File \"/opt/conda/lib/python3.7/site-packages/prefect/engine/cloud/utilities.py\", line 21, in prepare_state_for_cloud\r\n    res.store_safe_value()\r\n  File \"/opt/conda/lib/python3.7/site-packages/prefect/engine/result.py\", line 93, in store_safe_value\r\n    value = self.result_handler.write(self.value)\r\n  File \"/opt/conda/lib/python3.7/site-packages/prefect/engine/result_handlers/s3_result_handler.py\", line 103, in write\r\n    self.client.upload_fileobj(stream, Bucket=self.bucket, Key=uri)\r\n  File \"/opt/conda/lib/python3.7/site-packages/prefect/engine/result_handlers/s3_result_handler.py\", line 67, in client\r\n    self.initialize_client()\r\n  File \"/opt/conda/lib/python3.7/site-packages/prefect/engine/result_handlers/s3_result_handler.py\", line 60, in initialize_client\r\n    aws_secret_access_key=aws_secret_access_key,\r\n  File \"/opt/conda/lib/python3.7/site-packages/boto3/__init__.py\", line 91, in client\r\n    return _get_default_session().client(*args, **kwargs)\r\n  File \"/opt/conda/lib/python3.7/site-packages/boto3/session.py\", line 263, in client\r\n    aws_session_token=aws_session_token, config=config)\r\n  File \"/opt/conda/lib/python3.7/site-packages/botocore/session.py\", line 823, in create_client\r\n    credentials = self.get_credentials()\r\n  File \"/opt/conda/lib/python3.7/site-packages/botocore/session.py\", line 428, in get_credentials\r\n    'credential_provider').load_credentials()\r\n  File \"/opt/conda/lib/python3.7/site-packages/botocore/session.py\", line 923, in get_component\r\n    del self._deferred[name]\r\nKeyError: 'credential_provider'\r\n```\r\n\r\n## Environment\r\nWe create a long-running Dask cluster where our Dask workers are started with --nprocs 1 --nthreads 3.\r\n\r\n(Thanks to @JLouSRM for identifying this issue and capturing log evidence!)\r\n\n", "before_files": [{"content": "import base64\nimport io\nimport json\nimport uuid\nfrom typing import TYPE_CHECKING, Any\n\nimport cloudpickle\nimport pendulum\n\nfrom prefect.client import Secret\nfrom prefect.engine.result_handlers import ResultHandler\n\nif TYPE_CHECKING:\n    import boto3\n\n\nclass S3ResultHandler(ResultHandler):\n    \"\"\"\n    Result Handler for writing to and reading from an AWS S3 Bucket.\n\n    For authentication, there are two options: you can set a Prefect Secret containing\n    your AWS access keys which will be passed directly to the `boto3` client, or you can\n    [configure your flow's runtime environment](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#guide-configuration)\n    for `boto3`.\n\n    Args:\n        - bucket (str): the name of the bucket to write to / read from\n        - aws_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your AWS credentials; this Secret must be a JSON string\n            with two keys: `ACCESS_KEY` and `SECRET_ACCESS_KEY` which will be\n            passed directly to `boto3`.  If not provided, `boto3`\n            will fall back on standard AWS rules for authentication.\n    \"\"\"\n\n    def __init__(self, bucket: str, aws_credentials_secret: str = None) -> None:\n        self.bucket = bucket\n        self.aws_credentials_secret = aws_credentials_secret\n        super().__init__()\n\n    def initialize_client(self) -> None:\n        \"\"\"\n        Initializes an S3 Client.\n        \"\"\"\n        import boto3\n\n        aws_access_key = None\n        aws_secret_access_key = None\n\n        if self.aws_credentials_secret:\n            aws_credentials = Secret(self.aws_credentials_secret).get()\n            if isinstance(aws_credentials, str):\n                aws_credentials = json.loads(aws_credentials)\n\n            aws_access_key = aws_credentials[\"ACCESS_KEY\"]\n            aws_secret_access_key = aws_credentials[\"SECRET_ACCESS_KEY\"]\n\n        s3_client = boto3.client(\n            \"s3\",\n            aws_access_key_id=aws_access_key,\n            aws_secret_access_key=aws_secret_access_key,\n        )\n        self.client = s3_client\n\n    @property\n    def client(self) -> \"boto3.client\":\n        if not hasattr(self, \"_client\"):\n            self.initialize_client()\n        return self._client\n\n    @client.setter\n    def client(self, val: Any) -> None:\n        self._client = val\n\n    def __getstate__(self) -> dict:\n        state = self.__dict__.copy()\n        if \"_client\" in state:\n            del state[\"_client\"]\n        return state\n\n    def __setstate__(self, state: dict) -> None:\n        self.__dict__.update(state)\n\n    def write(self, result: Any) -> str:\n        \"\"\"\n        Given a result, writes the result to a location in S3\n        and returns the resulting URI.\n\n        Args:\n            - result (Any): the written result\n\n        Returns:\n            - str: the S3 URI\n        \"\"\"\n        date = pendulum.now(\"utc\").format(\"Y/M/D\")  # type: ignore\n        uri = \"{date}/{uuid}.prefect_result\".format(date=date, uuid=uuid.uuid4())\n        self.logger.debug(\"Starting to upload result to {}...\".format(uri))\n\n        ## prepare data\n        binary_data = base64.b64encode(cloudpickle.dumps(result))\n        stream = io.BytesIO(binary_data)\n\n        ## upload\n        self.client.upload_fileobj(stream, Bucket=self.bucket, Key=uri)\n        self.logger.debug(\"Finished uploading result to {}.\".format(uri))\n        return uri\n\n    def read(self, uri: str) -> Any:\n        \"\"\"\n        Given a uri, reads a result from S3, reads it and returns it\n\n        Args:\n            - uri (str): the S3 URI\n\n        Returns:\n            - Any: the read result\n        \"\"\"\n        try:\n            self.logger.debug(\"Starting to download result from {}...\".format(uri))\n            stream = io.BytesIO()\n\n            ## download\n            self.client.download_fileobj(Bucket=self.bucket, Key=uri, Fileobj=stream)\n            stream.seek(0)\n\n            try:\n                return_val = cloudpickle.loads(base64.b64decode(stream.read()))\n            except EOFError:\n                return_val = None\n            self.logger.debug(\"Finished downloading result from {}.\".format(uri))\n\n        except Exception as exc:\n            self.logger.exception(\n                \"Unexpected error while reading from result handler: {}\".format(\n                    repr(exc)\n                )\n            )\n            return_val = None\n\n        return return_val\n", "path": "src/prefect/engine/result_handlers/s3_result_handler.py"}], "after_files": [{"content": "import base64\nimport io\nimport json\nimport uuid\nfrom typing import TYPE_CHECKING, Any\n\nimport cloudpickle\nimport pendulum\n\nimport prefect\nfrom prefect.client import Secret\nfrom prefect.engine.result_handlers import ResultHandler\n\nif TYPE_CHECKING:\n    import boto3\n\n\nclass S3ResultHandler(ResultHandler):\n    \"\"\"\n    Result Handler for writing to and reading from an AWS S3 Bucket.\n\n    For authentication, there are two options: you can set a Prefect Secret containing\n    your AWS access keys which will be passed directly to the `boto3` client, or you can\n    [configure your flow's runtime environment](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#guide-configuration)\n    for `boto3`.\n\n    Args:\n        - bucket (str): the name of the bucket to write to / read from\n        - aws_credentials_secret (str, optional): the name of the Prefect Secret\n            that stores your AWS credentials; this Secret must be a JSON string\n            with two keys: `ACCESS_KEY` and `SECRET_ACCESS_KEY` which will be\n            passed directly to `boto3`.  If not provided, `boto3`\n            will fall back on standard AWS rules for authentication.\n    \"\"\"\n\n    def __init__(self, bucket: str, aws_credentials_secret: str = None) -> None:\n        self.bucket = bucket\n        self.aws_credentials_secret = aws_credentials_secret\n        super().__init__()\n\n    def initialize_client(self) -> None:\n        \"\"\"\n        Initializes an S3 Client.\n        \"\"\"\n        import boto3\n\n        aws_access_key = None\n        aws_secret_access_key = None\n\n        if self.aws_credentials_secret:\n            aws_credentials = Secret(self.aws_credentials_secret).get()\n            if isinstance(aws_credentials, str):\n                aws_credentials = json.loads(aws_credentials)\n\n            aws_access_key = aws_credentials[\"ACCESS_KEY\"]\n            aws_secret_access_key = aws_credentials[\"SECRET_ACCESS_KEY\"]\n\n        # use a new boto session when initializing in case we are in a new thread\n        # see https://boto3.amazonaws.com/v1/documentation/api/latest/guide/resources.html?#multithreading-multiprocessing\n        session = boto3.session.Session()\n        s3_client = session.client(\n            \"s3\",\n            aws_access_key_id=aws_access_key,\n            aws_secret_access_key=aws_secret_access_key,\n        )\n        self.client = s3_client\n\n    @property\n    def client(self) -> \"boto3.client\":\n        \"\"\"\n        Initializes a client if we believe we are in a new thread.\n        We consider ourselves in a new thread if we haven't stored a client yet in the current context.\n        \"\"\"\n        if not prefect.context.get(\"boto3client\"):\n            self.initialize_client()\n            prefect.context[\"boto3client\"] = self._client\n        return self._client\n\n    @client.setter\n    def client(self, val: Any) -> None:\n        self._client = val\n\n    def __getstate__(self) -> dict:\n        state = self.__dict__.copy()\n        if \"_client\" in state:\n            del state[\"_client\"]\n        return state\n\n    def __setstate__(self, state: dict) -> None:\n        self.__dict__.update(state)\n\n    def write(self, result: Any) -> str:\n        \"\"\"\n        Given a result, writes the result to a location in S3\n        and returns the resulting URI.\n\n        Args:\n            - result (Any): the written result\n\n        Returns:\n            - str: the S3 URI\n        \"\"\"\n        date = pendulum.now(\"utc\").format(\"Y/M/D\")  # type: ignore\n        uri = \"{date}/{uuid}.prefect_result\".format(date=date, uuid=uuid.uuid4())\n        self.logger.debug(\"Starting to upload result to {}...\".format(uri))\n\n        ## prepare data\n        binary_data = base64.b64encode(cloudpickle.dumps(result))\n        stream = io.BytesIO(binary_data)\n\n        ## upload\n        self.client.upload_fileobj(stream, Bucket=self.bucket, Key=uri)\n        self.logger.debug(\"Finished uploading result to {}.\".format(uri))\n        return uri\n\n    def read(self, uri: str) -> Any:\n        \"\"\"\n        Given a uri, reads a result from S3, reads it and returns it\n\n        Args:\n            - uri (str): the S3 URI\n\n        Returns:\n            - Any: the read result\n        \"\"\"\n        try:\n            self.logger.debug(\"Starting to download result from {}...\".format(uri))\n            stream = io.BytesIO()\n\n            ## download\n            self.client.download_fileobj(Bucket=self.bucket, Key=uri, Fileobj=stream)\n            stream.seek(0)\n\n            try:\n                return_val = cloudpickle.loads(base64.b64decode(stream.read()))\n            except EOFError:\n                return_val = None\n            self.logger.debug(\"Finished downloading result from {}.\".format(uri))\n\n        except Exception as exc:\n            self.logger.exception(\n                \"Unexpected error while reading from result handler: {}\".format(\n                    repr(exc)\n                )\n            )\n            return_val = None\n\n        return return_val\n", "path": "src/prefect/engine/result_handlers/s3_result_handler.py"}]}