prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 369: invalid start byte
  **Describe the bug**: 
  Sending PUT request to FastAPI with binary file encoded, its returns a error.
  
  **To Reproduce**
  
  1. Configure FastAPI with elasticapm.contrib.starlette
  2. Send a PUT request with binary file
  
  **Environment (please complete the following information)**
  - OS: Linux
  - Python version: 3.8
  - Framework and version: fastapi 0.61.2
  - APM Server version: 8
  - Agent version: 8
  
  The same problema was resolved in the main code : [#344](https://github.com/elastic/apm-agent-python/issues/334)
  
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `elasticapm/contrib/starlette/utils.py`
  Content:
  ```
  1 #  BSD 3-Clause License
  2 #
  3 #  Copyright (c) 2019, Elasticsearch BV
  4 #  All rights reserved.
  5 #
  6 #  Redistribution and use in source and binary forms, with or without
  7 #  modification, are permitted provided that the following conditions are met:
  8 #
  9 #  * Redistributions of source code must retain the above copyright notice, this
  10 #    list of conditions and the following disclaimer.
  11 #
  12 #  * Redistributions in binary form must reproduce the above copyright notice,
  13 #    this list of conditions and the following disclaimer in the documentation
  14 #    and/or other materials provided with the distribution.
  15 #
  16 #  * Neither the name of the copyright holder nor the names of its
  17 #    contributors may be used to endorse or promote products derived from
  18 #    this software without specific prior written permission.
  19 #
  20 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  22 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  23 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  24 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  25 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  26 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  27 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  28 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  29 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  30 import asyncio
  31 
  32 from starlette.datastructures import Headers
  33 from starlette.requests import Request
  34 from starlette.types import Message
  35 
  36 from elasticapm.conf import Config, constants
  37 from elasticapm.utils import get_url_dict
  38 
  39 
  40 async def get_data_from_request(request: Request, config: Config, event_type: str) -> dict:
  41     """Loads data from incoming request for APM capturing.
  42 
  43     Args:
  44         request (Request)
  45         config (Config)
  46         event_type (str)
  47 
  48     Returns:
  49         dict
  50     """
  51     result = {
  52         "method": request.method,
  53         "socket": {"remote_address": _get_client_ip(request)},
  54         "cookies": request.cookies,
  55     }
  56     if config.capture_headers:
  57         result["headers"] = dict(request.headers)
  58 
  59     if request.method in constants.HTTP_WITH_BODY:
  60         if config.capture_body not in ("all", event_type):
  61             result["body"] = "[REDACTED]"
  62         else:
  63             body = None
  64             try:
  65                 body = await get_body(request)
  66             except Exception:
  67                 pass
  68             if body is not None:
  69                 result["body"] = body
  70 
  71     result["url"] = get_url_dict(str(request.url))
  72 
  73     return result
  74 
  75 
  76 async def get_data_from_response(message: dict, config: Config, event_type: str) -> dict:
  77     """Loads data from response for APM capturing.
  78 
  79     Args:
  80         message (dict)
  81         config (Config)
  82         event_type (str)
  83 
  84     Returns:
  85         dict
  86     """
  87     result = {}
  88 
  89     if "status" in message:
  90         result["status_code"] = message["status"]
  91 
  92     if config.capture_headers and "headers" in message:
  93         headers = Headers(raw=message["headers"])
  94         result["headers"] = {key: ";".join(headers.getlist(key)) for key in headers.keys()}
  95 
  96     return result
  97 
  98 
  99 async def set_body(request: Request, body: bytes):
  100     """Overwrites body in Starlette.
  101 
  102     Args:
  103         request (Request)
  104         body (bytes)
  105     """
  106 
  107     async def receive() -> Message:
  108         await asyncio.sleep(0)
  109         return {"type": "http.request", "body": body}
  110 
  111     request._receive = receive
  112 
  113 
  114 async def get_body(request: Request) -> str:
  115     """Gets body from the request.
  116 
  117     When we consume the body, we replace the streaming mechanism with
  118     a mocked version -- this workaround came from
  119     https://github.com/encode/starlette/issues/495#issuecomment-513138055
  120 
  121     Args:
  122         request (Request)
  123 
  124     Returns:
  125         str
  126     """
  127     body = await request.body()
  128     await set_body(request, body)
  129 
  130     request._stream_consumed = False
  131 
  132     return body.decode("utf-8")
  133 
  134 
  135 async def query_params_to_dict(query_params: str) -> dict:
  136     """Transforms query params from URL to dictionary
  137 
  138     Args:
  139         query_params (str)
  140 
  141     Returns:
  142         dict
  143 
  144     Examples:
  145         >>> print(query_params_to_dict(b"key=val&key2=val2"))
  146         {"key": "val", "key2": "val2"}
  147     """
  148     query_params = query_params.split("&")
  149     res = {}
  150     for param in query_params:
  151         key, val = param.split("=")
  152         res[key] = val
  153 
  154     return res
  155 
  156 
  157 def _get_client_ip(request: Request):
  158     x_forwarded_for = request.headers.get("HTTP_X_FORWARDED_FOR")
  159     if x_forwarded_for:
  160         ip = x_forwarded_for.split(",")[0]
  161     else:
  162         ip = request.headers.get("REMOTE_ADDR")
  163     return ip
  164
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/elasticapm/contrib/starlette/utils.py b/elasticapm/contrib/starlette/utils.py
  --- a/elasticapm/contrib/starlette/utils.py
  +++ b/elasticapm/contrib/starlette/utils.py
  @@ -129,7 +129,7 @@
   
       request._stream_consumed = False
   
  -    return body.decode("utf-8")
  +    return body.decode("utf-8", errors="replace")
   
   
   async def query_params_to_dict(query_params: str) -> dict:
  
verification_info: {"golden_diff": "diff --git a/elasticapm/contrib/starlette/utils.py b/elasticapm/contrib/starlette/utils.py\n--- a/elasticapm/contrib/starlette/utils.py\n+++ b/elasticapm/contrib/starlette/utils.py\n@@ -129,7 +129,7 @@\n \n     request._stream_consumed = False\n \n-    return body.decode(\"utf-8\")\n+    return body.decode(\"utf-8\", errors=\"replace\")\n \n \n async def query_params_to_dict(query_params: str) -> dict:\n", "issue": "UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 369: invalid start byte\n**Describe the bug**: \r\nSending PUT request to FastAPI with binary file encoded, its returns a error.\r\n\r\n**To Reproduce**\r\n\r\n1. Configure FastAPI with elasticapm.contrib.starlette\r\n2. Send a PUT request with binary file\r\n\r\n**Environment (please complete the following information)**\r\n- OS: Linux\r\n- Python version: 3.8\r\n- Framework and version: fastapi 0.61.2\r\n- APM Server version: 8\r\n- Agent version: 8\r\n\r\nThe same problema was resolved in the main code : [#344](https://github.com/elastic/apm-agent-python/issues/334)\r\n\n", "before_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\nimport asyncio\n\nfrom starlette.datastructures import Headers\nfrom starlette.requests import Request\nfrom starlette.types import Message\n\nfrom elasticapm.conf import Config, constants\nfrom elasticapm.utils import get_url_dict\n\n\nasync def get_data_from_request(request: Request, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from incoming request for APM capturing.\n\n    Args:\n        request (Request)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {\n        \"method\": request.method,\n        \"socket\": {\"remote_address\": _get_client_ip(request)},\n        \"cookies\": request.cookies,\n    }\n    if config.capture_headers:\n        result[\"headers\"] = dict(request.headers)\n\n    if request.method in constants.HTTP_WITH_BODY:\n        if config.capture_body not in (\"all\", event_type):\n            result[\"body\"] = \"[REDACTED]\"\n        else:\n            body = None\n            try:\n                body = await get_body(request)\n            except Exception:\n                pass\n            if body is not None:\n                result[\"body\"] = body\n\n    result[\"url\"] = get_url_dict(str(request.url))\n\n    return result\n\n\nasync def get_data_from_response(message: dict, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from response for APM capturing.\n\n    Args:\n        message (dict)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {}\n\n    if \"status\" in message:\n        result[\"status_code\"] = message[\"status\"]\n\n    if config.capture_headers and \"headers\" in message:\n        headers = Headers(raw=message[\"headers\"])\n        result[\"headers\"] = {key: \";\".join(headers.getlist(key)) for key in headers.keys()}\n\n    return result\n\n\nasync def set_body(request: Request, body: bytes):\n    \"\"\"Overwrites body in Starlette.\n\n    Args:\n        request (Request)\n        body (bytes)\n    \"\"\"\n\n    async def receive() -> Message:\n        await asyncio.sleep(0)\n        return {\"type\": \"http.request\", \"body\": body}\n\n    request._receive = receive\n\n\nasync def get_body(request: Request) -> str:\n    \"\"\"Gets body from the request.\n\n    When we consume the body, we replace the streaming mechanism with\n    a mocked version -- this workaround came from\n    https://github.com/encode/starlette/issues/495#issuecomment-513138055\n\n    Args:\n        request (Request)\n\n    Returns:\n        str\n    \"\"\"\n    body = await request.body()\n    await set_body(request, body)\n\n    request._stream_consumed = False\n\n    return body.decode(\"utf-8\")\n\n\nasync def query_params_to_dict(query_params: str) -> dict:\n    \"\"\"Transforms query params from URL to dictionary\n\n    Args:\n        query_params (str)\n\n    Returns:\n        dict\n\n    Examples:\n        >>> print(query_params_to_dict(b\"key=val&key2=val2\"))\n        {\"key\": \"val\", \"key2\": \"val2\"}\n    \"\"\"\n    query_params = query_params.split(\"&\")\n    res = {}\n    for param in query_params:\n        key, val = param.split(\"=\")\n        res[key] = val\n\n    return res\n\n\ndef _get_client_ip(request: Request):\n    x_forwarded_for = request.headers.get(\"HTTP_X_FORWARDED_FOR\")\n    if x_forwarded_for:\n        ip = x_forwarded_for.split(\",\")[0]\n    else:\n        ip = request.headers.get(\"REMOTE_ADDR\")\n    return ip\n", "path": "elasticapm/contrib/starlette/utils.py"}], "after_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\nimport asyncio\n\nfrom starlette.datastructures import Headers\nfrom starlette.requests import Request\nfrom starlette.types import Message\n\nfrom elasticapm.conf import Config, constants\nfrom elasticapm.utils import get_url_dict\n\n\nasync def get_data_from_request(request: Request, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from incoming request for APM capturing.\n\n    Args:\n        request (Request)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {\n        \"method\": request.method,\n        \"socket\": {\"remote_address\": _get_client_ip(request)},\n        \"cookies\": request.cookies,\n    }\n    if config.capture_headers:\n        result[\"headers\"] = dict(request.headers)\n\n    if request.method in constants.HTTP_WITH_BODY:\n        if config.capture_body not in (\"all\", event_type):\n            result[\"body\"] = \"[REDACTED]\"\n        else:\n            body = None\n            try:\n                body = await get_body(request)\n            except Exception:\n                pass\n            if body is not None:\n                result[\"body\"] = body\n\n    result[\"url\"] = get_url_dict(str(request.url))\n\n    return result\n\n\nasync def get_data_from_response(message: dict, config: Config, event_type: str) -> dict:\n    \"\"\"Loads data from response for APM capturing.\n\n    Args:\n        message (dict)\n        config (Config)\n        event_type (str)\n\n    Returns:\n        dict\n    \"\"\"\n    result = {}\n\n    if \"status\" in message:\n        result[\"status_code\"] = message[\"status\"]\n\n    if config.capture_headers and \"headers\" in message:\n        headers = Headers(raw=message[\"headers\"])\n        result[\"headers\"] = {key: \";\".join(headers.getlist(key)) for key in headers.keys()}\n\n    return result\n\n\nasync def set_body(request: Request, body: bytes):\n    \"\"\"Overwrites body in Starlette.\n\n    Args:\n        request (Request)\n        body (bytes)\n    \"\"\"\n\n    async def receive() -> Message:\n        await asyncio.sleep(0)\n        return {\"type\": \"http.request\", \"body\": body}\n\n    request._receive = receive\n\n\nasync def get_body(request: Request) -> str:\n    \"\"\"Gets body from the request.\n\n    When we consume the body, we replace the streaming mechanism with\n    a mocked version -- this workaround came from\n    https://github.com/encode/starlette/issues/495#issuecomment-513138055\n\n    Args:\n        request (Request)\n\n    Returns:\n        str\n    \"\"\"\n    body = await request.body()\n    await set_body(request, body)\n\n    request._stream_consumed = False\n\n    return body.decode(\"utf-8\", errors=\"replace\")\n\n\nasync def query_params_to_dict(query_params: str) -> dict:\n    \"\"\"Transforms query params from URL to dictionary\n\n    Args:\n        query_params (str)\n\n    Returns:\n        dict\n\n    Examples:\n        >>> print(query_params_to_dict(b\"key=val&key2=val2\"))\n        {\"key\": \"val\", \"key2\": \"val2\"}\n    \"\"\"\n    query_params = query_params.split(\"&\")\n    res = {}\n    for param in query_params:\n        key, val = param.split(\"=\")\n        res[key] = val\n\n    return res\n\n\ndef _get_client_ip(request: Request):\n    x_forwarded_for = request.headers.get(\"HTTP_X_FORWARDED_FOR\")\n    if x_forwarded_for:\n        ip = x_forwarded_for.split(\",\")[0]\n    else:\n        ip = request.headers.get(\"REMOTE_ADDR\")\n    return ip\n", "path": "elasticapm/contrib/starlette/utils.py"}]}