import datetime
+import enum
import functools
import typing
import warnings
)
+class ClientState(enum.Enum):
+ UNOPENED = 1
+ OPENED = 2
+ CLOSED = 3
+
+
class BaseClient:
def __init__(
self,
}
self._trust_env = trust_env
self._netrc = NetRCInfo()
- self._is_closed = True
+ self._state = ClientState.UNOPENED
@property
def is_closed(self) -> bool:
"""
Check if the client being closed
"""
- return self._is_closed
+ return self._state == ClientState.CLOSED
@property
def trust_env(self) -> bool:
[0]: /advanced/#request-instances
"""
- self._is_closed = False
+ if self._state == ClientState.CLOSED:
+ raise RuntimeError("Cannot send a request, as the client has been closed.")
+ self._state = ClientState.OPENED
timeout = self.timeout if isinstance(timeout, UnsetType) else Timeout(timeout)
auth = self._build_request_auth(request, auth)
"""
Close transport and proxies.
"""
- if not self.is_closed:
- self._is_closed = True
+ if self._state != ClientState.CLOSED:
+ self._state = ClientState.CLOSED
self._transport.close()
for proxy in self._proxies.values():
proxy.close()
def __enter__(self: T) -> T:
+ self._state = ClientState.OPENED
+
self._transport.__enter__()
for proxy in self._proxies.values():
if proxy is not None:
proxy.__enter__()
- self._is_closed = False
return self
def __exit__(
exc_value: BaseException = None,
traceback: TracebackType = None,
) -> None:
- if not self.is_closed:
- self._is_closed = True
+ self._state = ClientState.CLOSED
- self._transport.__exit__(exc_type, exc_value, traceback)
- for proxy in self._proxies.values():
- if proxy is not None:
- proxy.__exit__(exc_type, exc_value, traceback)
+ self._transport.__exit__(exc_type, exc_value, traceback)
+ for proxy in self._proxies.values():
+ if proxy is not None:
+ proxy.__exit__(exc_type, exc_value, traceback)
def __del__(self) -> None:
self.close()
[0]: /advanced/#request-instances
"""
- self._is_closed = False
+ if self._state == ClientState.CLOSED:
+ raise RuntimeError("Cannot send a request, as the client has been closed.")
+ self._state = ClientState.OPENED
timeout = self.timeout if isinstance(timeout, UnsetType) else Timeout(timeout)
auth = self._build_request_auth(request, auth)
"""
Close transport and proxies.
"""
- if not self.is_closed:
- self._is_closed = True
+ if self._state != ClientState.CLOSED:
+ self._state = ClientState.CLOSED
await self._transport.aclose()
for proxy in self._proxies.values():
await proxy.aclose()
async def __aenter__(self: U) -> U:
+ self._state = ClientState.OPENED
+
await self._transport.__aenter__()
for proxy in self._proxies.values():
if proxy is not None:
await proxy.__aenter__()
- self._is_closed = False
return self
async def __aexit__(
exc_value: BaseException = None,
traceback: TracebackType = None,
) -> None:
- if not self.is_closed:
- self._is_closed = True
- await self._transport.__aexit__(exc_type, exc_value, traceback)
- for proxy in self._proxies.values():
- if proxy is not None:
- await proxy.__aexit__(exc_type, exc_value, traceback)
+ self._state = ClientState.CLOSED
+
+ await self._transport.__aexit__(exc_type, exc_value, traceback)
+ for proxy in self._proxies.values():
+ if proxy is not None:
+ await proxy.__aexit__(exc_type, exc_value, traceback)
def __del__(self) -> None:
- if not self.is_closed:
+ if self._state == ClientState.OPENED:
warnings.warn(
f"Unclosed {self!r}. "
"See https://www.python-httpx.org/async/#opening-and-closing-clients "
import pytest
import httpx
+from tests.utils import MockTransport
@pytest.mark.usefixtures("async_environment")
]
-@pytest.mark.usefixtures("async_environment")
-async def test_that_async_client_is_closed_by_default():
- client = httpx.AsyncClient()
-
- assert client.is_closed
+def hello_world(request):
+ return httpx.Response(200, text="Hello, world!")
@pytest.mark.usefixtures("async_environment")
-async def test_that_send_cause_async_client_to_be_not_closed():
- client = httpx.AsyncClient()
+async def test_client_closed_state_using_implicit_open():
+ client = httpx.AsyncClient(transport=MockTransport(hello_world))
+ assert not client.is_closed
await client.get("http://example.com")
assert not client.is_closed
-
await client.aclose()
-
-@pytest.mark.usefixtures("async_environment")
-async def test_that_async_client_is_not_closed_in_with_block():
- async with httpx.AsyncClient() as client:
- assert not client.is_closed
+ assert client.is_closed
+ with pytest.raises(RuntimeError):
+ await client.get("http://example.com")
@pytest.mark.usefixtures("async_environment")
-async def test_that_async_client_is_closed_after_with_block():
- async with httpx.AsyncClient() as client:
- pass
+async def test_client_closed_state_using_with_block():
+ async with httpx.AsyncClient(transport=MockTransport(hello_world)) as client:
+ assert not client.is_closed
+ await client.get("http://example.com")
assert client.is_closed
+ with pytest.raises(RuntimeError):
+ await client.get("http://example.com")
@pytest.mark.usefixtures("async_environment")
-async def test_that_async_client_caused_warning_when_being_deleted():
- async_client = httpx.AsyncClient()
-
- await async_client.get("http://example.com")
-
+async def test_deleting_unclosed_async_client_causes_warning():
+ client = httpx.AsyncClient(transport=MockTransport(hello_world))
+ await client.get("http://example.com")
with pytest.warns(UserWarning):
- del async_client
+ del client
import pytest
import httpx
+from tests.utils import MockTransport
def test_get(server):
]
-def test_that_client_is_closed_by_default():
- client = httpx.Client()
-
- assert client.is_closed
+def hello_world(request):
+ return httpx.Response(200, text="Hello, world!")
-def test_that_send_cause_client_to_be_not_closed():
- client = httpx.Client()
+def test_client_closed_state_using_implicit_open():
+ client = httpx.Client(transport=MockTransport(hello_world))
+ assert not client.is_closed
client.get("http://example.com")
assert not client.is_closed
+ client.close()
-
-def test_that_client_is_not_closed_in_with_block():
- with httpx.Client() as client:
- assert not client.is_closed
+ assert client.is_closed
+ with pytest.raises(RuntimeError):
+ client.get("http://example.com")
-def test_that_client_is_closed_after_with_block():
- with httpx.Client() as client:
- pass
+def test_client_closed_state_using_with_block():
+ with httpx.Client(transport=MockTransport(hello_world)) as client:
+ assert not client.is_closed
+ client.get("http://example.com")
assert client.is_closed
+ with pytest.raises(RuntimeError):
+ client.get("http://example.com")
projects / thirdparty / httpx.git / commitdiff
