We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`web3.providers.ipc.get_default_ipc_path()` returned `None` instead of `str`
* Version: 6.0.0
* Python: 3.10
* OS: macky whacky
* `pip freeze` output

```
will provide if needed
```


### What was wrong?

The method `get_deafult_ipc_path()` located in `web3.providers.ipc` returned `None` even though its return type is documented as `str`.

I am not sure why it returned `None` in my case, probably forgot some flag in `geth`.

Edit: it returned None in my case because i was checking it before starting geth, so it didnt exist yet.

### How can it be fixed?

Either change type to `Optional[str]` or raise an error instead of returning `None` ?

---
**Note:** We prefer to use issues to track our work. If you think you've encountered a bug in web3py or 
have a feature request, you're in the right place. If you have implementation or usage questions, 
please refer to our [documentation](https://web3py.readthedocs.io/en/latest/) and/or join the conversation 
on [discord](https://discord.gg/GHryRvPB84).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `web3/providers/ipc.py`
Content:
```
1 from json import (
2     JSONDecodeError,
3 )
4 import logging
5 import os
6 from pathlib import (
7     Path,
8 )
9 import socket
10 import sys
11 import threading
12 from types import (
13     TracebackType,
14 )
15 from typing import (
16     Any,
17     Type,
18     Union,
19 )
20 
21 from web3._utils.threads import (
22     Timeout,
23 )
24 from web3.types import (
25     RPCEndpoint,
26     RPCResponse,
27 )
28 
29 from .base import (
30     JSONBaseProvider,
31 )
32 
33 
34 def get_ipc_socket(ipc_path: str, timeout: float = 2.0) -> socket.socket:
35     if sys.platform == "win32":
36         # On Windows named pipe is used. Simulate socket with it.
37         from web3._utils.windows import (
38             NamedPipe,
39         )
40 
41         return NamedPipe(ipc_path)
42     else:
43         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
44         sock.connect(ipc_path)
45         sock.settimeout(timeout)
46         return sock
47 
48 
49 class PersistantSocket:
50     sock = None
51 
52     def __init__(self, ipc_path: str) -> None:
53         self.ipc_path = ipc_path
54 
55     def __enter__(self) -> socket.socket:
56         if not self.ipc_path:
57             raise FileNotFoundError(
58                 f"cannot connect to IPC socket at path: {self.ipc_path!r}"
59             )
60 
61         if not self.sock:
62             self.sock = self._open()
63         return self.sock
64 
65     def __exit__(
66         self,
67         exc_type: Type[BaseException],
68         exc_value: BaseException,
69         traceback: TracebackType,
70     ) -> None:
71         # only close the socket if there was an error
72         if exc_value is not None:
73             try:
74                 self.sock.close()
75             except Exception:
76                 pass
77             self.sock = None
78 
79     def _open(self) -> socket.socket:
80         return get_ipc_socket(self.ipc_path)
81 
82     def reset(self) -> socket.socket:
83         self.sock.close()
84         self.sock = self._open()
85         return self.sock
86 
87 
88 # type ignored b/c missing return statement is by design here
89 def get_default_ipc_path() -> str:  # type: ignore
90     if sys.platform == "darwin":
91         ipc_path = os.path.expanduser(
92             os.path.join("~", "Library", "Ethereum", "geth.ipc")
93         )
94         if os.path.exists(ipc_path):
95             return ipc_path
96 
97     elif sys.platform.startswith("linux") or sys.platform.startswith("freebsd"):
98         ipc_path = os.path.expanduser(os.path.join("~", ".ethereum", "geth.ipc"))
99         if os.path.exists(ipc_path):
100             return ipc_path
101 
102     elif sys.platform == "win32":
103         ipc_path = os.path.join("\\\\", ".", "pipe", "geth.ipc")
104         if os.path.exists(ipc_path):
105             return ipc_path
106 
107     else:
108         raise ValueError(
109             f"Unsupported platform '{sys.platform}'.  Only darwin/linux/win32/"
110             "freebsd are supported.  You must specify the ipc_path"
111         )
112 
113 
114 # type ignored b/c missing return statement is by design here
115 def get_dev_ipc_path() -> str:  # type: ignore
116     if os.environ.get("WEB3_PROVIDER_URI", ""):
117         ipc_path = os.environ.get("WEB3_PROVIDER_URI")
118         if os.path.exists(ipc_path):
119             return ipc_path
120     elif sys.platform == "darwin":
121         tmpdir = os.environ.get("TMPDIR", "")
122         ipc_path = os.path.expanduser(os.path.join(tmpdir, "geth.ipc"))
123         if os.path.exists(ipc_path):
124             return ipc_path
125 
126     elif sys.platform.startswith("linux") or sys.platform.startswith("freebsd"):
127         ipc_path = os.path.expanduser(os.path.join("/tmp", "geth.ipc"))
128         if os.path.exists(ipc_path):
129             return ipc_path
130 
131     elif sys.platform == "win32":
132         ipc_path = os.path.join("\\\\", ".", "pipe", "geth.ipc")
133         if os.path.exists(ipc_path):
134             return ipc_path
135 
136         ipc_path = os.path.join("\\\\", ".", "pipe", "jsonrpc.ipc")
137         if os.path.exists(ipc_path):
138             return ipc_path
139 
140     else:
141         raise ValueError(
142             f"Unsupported platform '{sys.platform}'.  Only darwin/linux/win32/"
143             "freebsd are supported.  You must specify the ipc_path"
144         )
145 
146 
147 class IPCProvider(JSONBaseProvider):
148     logger = logging.getLogger("web3.providers.IPCProvider")
149     _socket = None
150 
151     def __init__(
152         self,
153         ipc_path: Union[str, Path] = None,
154         timeout: int = 10,
155         *args: Any,
156         **kwargs: Any,
157     ) -> None:
158         if ipc_path is None:
159             self.ipc_path = get_default_ipc_path()
160         elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):
161             self.ipc_path = str(Path(ipc_path).expanduser().resolve())
162         else:
163             raise TypeError("ipc_path must be of type string or pathlib.Path")
164 
165         self.timeout = timeout
166         self._lock = threading.Lock()
167         self._socket = PersistantSocket(self.ipc_path)
168         super().__init__()
169 
170     def __str__(self) -> str:
171         return f"<{self.__class__.__name__} {self.ipc_path}>"
172 
173     def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
174         self.logger.debug(
175             f"Making request IPC. Path: {self.ipc_path}, Method: {method}"
176         )
177         request = self.encode_rpc_request(method, params)
178 
179         with self._lock, self._socket as sock:
180             try:
181                 sock.sendall(request)
182             except BrokenPipeError:
183                 # one extra attempt, then give up
184                 sock = self._socket.reset()
185                 sock.sendall(request)
186 
187             raw_response = b""
188             with Timeout(self.timeout) as timeout:
189                 while True:
190                     try:
191                         raw_response += sock.recv(4096)
192                     except socket.timeout:
193                         timeout.sleep(0)
194                         continue
195                     if raw_response == b"":
196                         timeout.sleep(0)
197                     elif has_valid_json_rpc_ending(raw_response):
198                         try:
199                             response = self.decode_rpc_response(raw_response)
200                         except JSONDecodeError:
201                             timeout.sleep(0)
202                             continue
203                         else:
204                             return response
205                     else:
206                         timeout.sleep(0)
207                         continue
208 
209 
210 # A valid JSON RPC response can only end in } or ] http://www.jsonrpc.org/specification
211 def has_valid_json_rpc_ending(raw_response: bytes) -> bool:
212     stripped_raw_response = raw_response.rstrip()
213     for valid_ending in [b"}", b"]"]:
214         if stripped_raw_response.endswith(valid_ending):
215             return True
216     else:
217         return False
218
```
--- 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/web3/providers/ipc.py b/web3/providers/ipc.py
--- a/web3/providers/ipc.py
+++ b/web3/providers/ipc.py
@@ -14,6 +14,7 @@
 )
 from typing import (
     Any,
+    Optional,
     Type,
     Union,
 )
@@ -85,24 +86,26 @@
         return self.sock
 
 
-# type ignored b/c missing return statement is by design here
-def get_default_ipc_path() -> str:  # type: ignore
+def get_default_ipc_path() -> Optional[str]:
     if sys.platform == "darwin":
         ipc_path = os.path.expanduser(
             os.path.join("~", "Library", "Ethereum", "geth.ipc")
         )
         if os.path.exists(ipc_path):
             return ipc_path
+        return None
 
     elif sys.platform.startswith("linux") or sys.platform.startswith("freebsd"):
         ipc_path = os.path.expanduser(os.path.join("~", ".ethereum", "geth.ipc"))
         if os.path.exists(ipc_path):
             return ipc_path
+        return None
 
     elif sys.platform == "win32":
         ipc_path = os.path.join("\\\\", ".", "pipe", "geth.ipc")
         if os.path.exists(ipc_path):
             return ipc_path
+        return None
 
     else:
         raise ValueError(
@@ -111,22 +114,25 @@
         )
 
 
-# type ignored b/c missing return statement is by design here
-def get_dev_ipc_path() -> str:  # type: ignore
+def get_dev_ipc_path() -> Optional[str]:
     if os.environ.get("WEB3_PROVIDER_URI", ""):
         ipc_path = os.environ.get("WEB3_PROVIDER_URI")
         if os.path.exists(ipc_path):
             return ipc_path
+        return None
+
     elif sys.platform == "darwin":
         tmpdir = os.environ.get("TMPDIR", "")
         ipc_path = os.path.expanduser(os.path.join(tmpdir, "geth.ipc"))
         if os.path.exists(ipc_path):
             return ipc_path
+        return None
 
     elif sys.platform.startswith("linux") or sys.platform.startswith("freebsd"):
         ipc_path = os.path.expanduser(os.path.join("/tmp", "geth.ipc"))
         if os.path.exists(ipc_path):
             return ipc_path
+        return None
 
     elif sys.platform == "win32":
         ipc_path = os.path.join("\\\\", ".", "pipe", "geth.ipc")

{"golden_diff": "diff --git a/web3/providers/ipc.py b/web3/providers/ipc.py\n--- a/web3/providers/ipc.py\n+++ b/web3/providers/ipc.py\n@@ -14,6 +14,7 @@\n )\n from typing import (\n     Any,\n+    Optional,\n     Type,\n     Union,\n )\n@@ -85,24 +86,26 @@\n         return self.sock\n \n \n-# type ignored b/c missing return statement is by design here\n-def get_default_ipc_path() -> str:  # type: ignore\n+def get_default_ipc_path() -> Optional[str]:\n     if sys.platform == \"darwin\":\n         ipc_path = os.path.expanduser(\n             os.path.join(\"~\", \"Library\", \"Ethereum\", \"geth.ipc\")\n         )\n         if os.path.exists(ipc_path):\n             return ipc_path\n+        return None\n \n     elif sys.platform.startswith(\"linux\") or sys.platform.startswith(\"freebsd\"):\n         ipc_path = os.path.expanduser(os.path.join(\"~\", \".ethereum\", \"geth.ipc\"))\n         if os.path.exists(ipc_path):\n             return ipc_path\n+        return None\n \n     elif sys.platform == \"win32\":\n         ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"geth.ipc\")\n         if os.path.exists(ipc_path):\n             return ipc_path\n+        return None\n \n     else:\n         raise ValueError(\n@@ -111,22 +114,25 @@\n         )\n \n \n-# type ignored b/c missing return statement is by design here\n-def get_dev_ipc_path() -> str:  # type: ignore\n+def get_dev_ipc_path() -> Optional[str]:\n     if os.environ.get(\"WEB3_PROVIDER_URI\", \"\"):\n         ipc_path = os.environ.get(\"WEB3_PROVIDER_URI\")\n         if os.path.exists(ipc_path):\n             return ipc_path\n+        return None\n+\n     elif sys.platform == \"darwin\":\n         tmpdir = os.environ.get(\"TMPDIR\", \"\")\n         ipc_path = os.path.expanduser(os.path.join(tmpdir, \"geth.ipc\"))\n         if os.path.exists(ipc_path):\n             return ipc_path\n+        return None\n \n     elif sys.platform.startswith(\"linux\") or sys.platform.startswith(\"freebsd\"):\n         ipc_path = os.path.expanduser(os.path.join(\"/tmp\", \"geth.ipc\"))\n         if os.path.exists(ipc_path):\n             return ipc_path\n+        return None\n \n     elif sys.platform == \"win32\":\n         ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"geth.ipc\")\n", "issue": "`web3.providers.ipc.get_default_ipc_path()` returned `None` instead of `str`\n* Version: 6.0.0\r\n* Python: 3.10\r\n* OS: macky whacky\r\n* `pip freeze` output\r\n\r\n```\r\nwill provide if needed\r\n```\r\n\r\n\r\n### What was wrong?\r\n\r\nThe method `get_deafult_ipc_path()` located in `web3.providers.ipc` returned `None` even though its return type is documented as `str`.\r\n\r\nI am not sure why it returned `None` in my case, probably forgot some flag in `geth`.\r\n\r\nEdit: it returned None in my case because i was checking it before starting geth, so it didnt exist yet.\r\n\r\n### How can it be fixed?\r\n\r\nEither change type to `Optional[str]` or raise an error instead of returning `None` ?\r\n\r\n---\r\n**Note:** We prefer to use issues to track our work. If you think you've encountered a bug in web3py or \r\nhave a feature request, you're in the right place. If you have implementation or usage questions, \r\nplease refer to our [documentation](https://web3py.readthedocs.io/en/latest/) and/or join the conversation \r\non [discord](https://discord.gg/GHryRvPB84).\r\n\n", "before_files": [{"content": "from json import (\n    JSONDecodeError,\n)\nimport logging\nimport os\nfrom pathlib import (\n    Path,\n)\nimport socket\nimport sys\nimport threading\nfrom types import (\n    TracebackType,\n)\nfrom typing import (\n    Any,\n    Type,\n    Union,\n)\n\nfrom web3._utils.threads import (\n    Timeout,\n)\nfrom web3.types import (\n    RPCEndpoint,\n    RPCResponse,\n)\n\nfrom .base import (\n    JSONBaseProvider,\n)\n\n\ndef get_ipc_socket(ipc_path: str, timeout: float = 2.0) -> socket.socket:\n    if sys.platform == \"win32\":\n        # On Windows named pipe is used. Simulate socket with it.\n        from web3._utils.windows import (\n            NamedPipe,\n        )\n\n        return NamedPipe(ipc_path)\n    else:\n        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)\n        sock.connect(ipc_path)\n        sock.settimeout(timeout)\n        return sock\n\n\nclass PersistantSocket:\n    sock = None\n\n    def __init__(self, ipc_path: str) -> None:\n        self.ipc_path = ipc_path\n\n    def __enter__(self) -> socket.socket:\n        if not self.ipc_path:\n            raise FileNotFoundError(\n                f\"cannot connect to IPC socket at path: {self.ipc_path!r}\"\n            )\n\n        if not self.sock:\n            self.sock = self._open()\n        return self.sock\n\n    def __exit__(\n        self,\n        exc_type: Type[BaseException],\n        exc_value: BaseException,\n        traceback: TracebackType,\n    ) -> None:\n        # only close the socket if there was an error\n        if exc_value is not None:\n            try:\n                self.sock.close()\n            except Exception:\n                pass\n            self.sock = None\n\n    def _open(self) -> socket.socket:\n        return get_ipc_socket(self.ipc_path)\n\n    def reset(self) -> socket.socket:\n        self.sock.close()\n        self.sock = self._open()\n        return self.sock\n\n\n# type ignored b/c missing return statement is by design here\ndef get_default_ipc_path() -> str:  # type: ignore\n    if sys.platform == \"darwin\":\n        ipc_path = os.path.expanduser(\n            os.path.join(\"~\", \"Library\", \"Ethereum\", \"geth.ipc\")\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform.startswith(\"linux\") or sys.platform.startswith(\"freebsd\"):\n        ipc_path = os.path.expanduser(os.path.join(\"~\", \".ethereum\", \"geth.ipc\"))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform == \"win32\":\n        ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"geth.ipc\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            f\"Unsupported platform '{sys.platform}'.  Only darwin/linux/win32/\"\n            \"freebsd are supported.  You must specify the ipc_path\"\n        )\n\n\n# type ignored b/c missing return statement is by design here\ndef get_dev_ipc_path() -> str:  # type: ignore\n    if os.environ.get(\"WEB3_PROVIDER_URI\", \"\"):\n        ipc_path = os.environ.get(\"WEB3_PROVIDER_URI\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n    elif sys.platform == \"darwin\":\n        tmpdir = os.environ.get(\"TMPDIR\", \"\")\n        ipc_path = os.path.expanduser(os.path.join(tmpdir, \"geth.ipc\"))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform.startswith(\"linux\") or sys.platform.startswith(\"freebsd\"):\n        ipc_path = os.path.expanduser(os.path.join(\"/tmp\", \"geth.ipc\"))\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    elif sys.platform == \"win32\":\n        ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"geth.ipc\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"jsonrpc.ipc\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            f\"Unsupported platform '{sys.platform}'.  Only darwin/linux/win32/\"\n            \"freebsd are supported.  You must specify the ipc_path\"\n        )\n\n\nclass IPCProvider(JSONBaseProvider):\n    logger = logging.getLogger(\"web3.providers.IPCProvider\")\n    _socket = None\n\n    def __init__(\n        self,\n        ipc_path: Union[str, Path] = None,\n        timeout: int = 10,\n        *args: Any,\n        **kwargs: Any,\n    ) -> None:\n        if ipc_path is None:\n            self.ipc_path = get_default_ipc_path()\n        elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):\n            self.ipc_path = str(Path(ipc_path).expanduser().resolve())\n        else:\n            raise TypeError(\"ipc_path must be of type string or pathlib.Path\")\n\n        self.timeout = timeout\n        self._lock = threading.Lock()\n        self._socket = PersistantSocket(self.ipc_path)\n        super().__init__()\n\n    def __str__(self) -> str:\n        return f\"<{self.__class__.__name__} {self.ipc_path}>\"\n\n    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:\n        self.logger.debug(\n            f\"Making request IPC. Path: {self.ipc_path}, Method: {method}\"\n        )\n        request = self.encode_rpc_request(method, params)\n\n        with self._lock, self._socket as sock:\n            try:\n                sock.sendall(request)\n            except BrokenPipeError:\n                # one extra attempt, then give up\n                sock = self._socket.reset()\n                sock.sendall(request)\n\n            raw_response = b\"\"\n            with Timeout(self.timeout) as timeout:\n                while True:\n                    try:\n                        raw_response += sock.recv(4096)\n                    except socket.timeout:\n                        timeout.sleep(0)\n                        continue\n                    if raw_response == b\"\":\n                        timeout.sleep(0)\n                    elif has_valid_json_rpc_ending(raw_response):\n                        try:\n                            response = self.decode_rpc_response(raw_response)\n                        except JSONDecodeError:\n                            timeout.sleep(0)\n                            continue\n                        else:\n                            return response\n                    else:\n                        timeout.sleep(0)\n                        continue\n\n\n# A valid JSON RPC response can only end in } or ] http://www.jsonrpc.org/specification\ndef has_valid_json_rpc_ending(raw_response: bytes) -> bool:\n    stripped_raw_response = raw_response.rstrip()\n    for valid_ending in [b\"}\", b\"]\"]:\n        if stripped_raw_response.endswith(valid_ending):\n            return True\n    else:\n        return False\n", "path": "web3/providers/ipc.py"}], "after_files": [{"content": "from json import (\n    JSONDecodeError,\n)\nimport logging\nimport os\nfrom pathlib import (\n    Path,\n)\nimport socket\nimport sys\nimport threading\nfrom types import (\n    TracebackType,\n)\nfrom typing import (\n    Any,\n    Optional,\n    Type,\n    Union,\n)\n\nfrom web3._utils.threads import (\n    Timeout,\n)\nfrom web3.types import (\n    RPCEndpoint,\n    RPCResponse,\n)\n\nfrom .base import (\n    JSONBaseProvider,\n)\n\n\ndef get_ipc_socket(ipc_path: str, timeout: float = 2.0) -> socket.socket:\n    if sys.platform == \"win32\":\n        # On Windows named pipe is used. Simulate socket with it.\n        from web3._utils.windows import (\n            NamedPipe,\n        )\n\n        return NamedPipe(ipc_path)\n    else:\n        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)\n        sock.connect(ipc_path)\n        sock.settimeout(timeout)\n        return sock\n\n\nclass PersistantSocket:\n    sock = None\n\n    def __init__(self, ipc_path: str) -> None:\n        self.ipc_path = ipc_path\n\n    def __enter__(self) -> socket.socket:\n        if not self.ipc_path:\n            raise FileNotFoundError(\n                f\"cannot connect to IPC socket at path: {self.ipc_path!r}\"\n            )\n\n        if not self.sock:\n            self.sock = self._open()\n        return self.sock\n\n    def __exit__(\n        self,\n        exc_type: Type[BaseException],\n        exc_value: BaseException,\n        traceback: TracebackType,\n    ) -> None:\n        # only close the socket if there was an error\n        if exc_value is not None:\n            try:\n                self.sock.close()\n            except Exception:\n                pass\n            self.sock = None\n\n    def _open(self) -> socket.socket:\n        return get_ipc_socket(self.ipc_path)\n\n    def reset(self) -> socket.socket:\n        self.sock.close()\n        self.sock = self._open()\n        return self.sock\n\n\ndef get_default_ipc_path() -> Optional[str]:\n    if sys.platform == \"darwin\":\n        ipc_path = os.path.expanduser(\n            os.path.join(\"~\", \"Library\", \"Ethereum\", \"geth.ipc\")\n        )\n        if os.path.exists(ipc_path):\n            return ipc_path\n        return None\n\n    elif sys.platform.startswith(\"linux\") or sys.platform.startswith(\"freebsd\"):\n        ipc_path = os.path.expanduser(os.path.join(\"~\", \".ethereum\", \"geth.ipc\"))\n        if os.path.exists(ipc_path):\n            return ipc_path\n        return None\n\n    elif sys.platform == \"win32\":\n        ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"geth.ipc\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n        return None\n\n    else:\n        raise ValueError(\n            f\"Unsupported platform '{sys.platform}'.  Only darwin/linux/win32/\"\n            \"freebsd are supported.  You must specify the ipc_path\"\n        )\n\n\ndef get_dev_ipc_path() -> Optional[str]:\n    if os.environ.get(\"WEB3_PROVIDER_URI\", \"\"):\n        ipc_path = os.environ.get(\"WEB3_PROVIDER_URI\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n        return None\n\n    elif sys.platform == \"darwin\":\n        tmpdir = os.environ.get(\"TMPDIR\", \"\")\n        ipc_path = os.path.expanduser(os.path.join(tmpdir, \"geth.ipc\"))\n        if os.path.exists(ipc_path):\n            return ipc_path\n        return None\n\n    elif sys.platform.startswith(\"linux\") or sys.platform.startswith(\"freebsd\"):\n        ipc_path = os.path.expanduser(os.path.join(\"/tmp\", \"geth.ipc\"))\n        if os.path.exists(ipc_path):\n            return ipc_path\n        return None\n\n    elif sys.platform == \"win32\":\n        ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"geth.ipc\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n        ipc_path = os.path.join(\"\\\\\\\\\", \".\", \"pipe\", \"jsonrpc.ipc\")\n        if os.path.exists(ipc_path):\n            return ipc_path\n\n    else:\n        raise ValueError(\n            f\"Unsupported platform '{sys.platform}'.  Only darwin/linux/win32/\"\n            \"freebsd are supported.  You must specify the ipc_path\"\n        )\n\n\nclass IPCProvider(JSONBaseProvider):\n    logger = logging.getLogger(\"web3.providers.IPCProvider\")\n    _socket = None\n\n    def __init__(\n        self,\n        ipc_path: Union[str, Path] = None,\n        timeout: int = 10,\n        *args: Any,\n        **kwargs: Any,\n    ) -> None:\n        if ipc_path is None:\n            self.ipc_path = get_default_ipc_path()\n        elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):\n            self.ipc_path = str(Path(ipc_path).expanduser().resolve())\n        else:\n            raise TypeError(\"ipc_path must be of type string or pathlib.Path\")\n\n        self.timeout = timeout\n        self._lock = threading.Lock()\n        self._socket = PersistantSocket(self.ipc_path)\n        super().__init__()\n\n    def __str__(self) -> str:\n        return f\"<{self.__class__.__name__} {self.ipc_path}>\"\n\n    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:\n        self.logger.debug(\n            f\"Making request IPC. Path: {self.ipc_path}, Method: {method}\"\n        )\n        request = self.encode_rpc_request(method, params)\n\n        with self._lock, self._socket as sock:\n            try:\n                sock.sendall(request)\n            except BrokenPipeError:\n                # one extra attempt, then give up\n                sock = self._socket.reset()\n                sock.sendall(request)\n\n            raw_response = b\"\"\n            with Timeout(self.timeout) as timeout:\n                while True:\n                    try:\n                        raw_response += sock.recv(4096)\n                    except socket.timeout:\n                        timeout.sleep(0)\n                        continue\n                    if raw_response == b\"\":\n                        timeout.sleep(0)\n                    elif has_valid_json_rpc_ending(raw_response):\n                        try:\n                            response = self.decode_rpc_response(raw_response)\n                        except JSONDecodeError:\n                            timeout.sleep(0)\n                            continue\n                        else:\n                            return response\n                    else:\n                        timeout.sleep(0)\n                        continue\n\n\n# A valid JSON RPC response can only end in } or ] http://www.jsonrpc.org/specification\ndef has_valid_json_rpc_ending(raw_response: bytes) -> bool:\n    stripped_raw_response = raw_response.rstrip()\n    for valid_ending in [b\"}\", b\"]\"]:\n        if stripped_raw_response.endswith(valid_ending):\n            return True\n    else:\n        return False\n", "path": "web3/providers/ipc.py"}]}