Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES/8463.misc.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Added a 3.11-specific overloads to ``ClientSession`` -- by :user:`max-muoto`.
176 changes: 129 additions & 47 deletions aiohttp/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
Set,
Tuple,
Type,
TypedDict,
TypeVar,
Union,
)
Expand Down Expand Up @@ -150,6 +151,37 @@
else:
SSLContext = None

if sys.version_info >= (3, 11) and TYPE_CHECKING:
from typing import Unpack


class _RequestOptions(TypedDict, total=False):
params: Union[Mapping[str, str], None]
data: Any
json: Any
cookies: Union[LooseCookies, None]
headers: Union[LooseHeaders, None]
skip_auto_headers: Union[Iterable[str], None]
auth: Union[BasicAuth, None]
allow_redirects: bool
max_redirects: int
compress: Union[str, None]
chunked: Union[bool, None]
expect100: bool
raise_for_status: Union[None, bool, Callable[[ClientResponse], Awaitable[None]]]
read_until_eof: bool
proxy: Union[StrOrURL, None]
proxy_auth: Union[BasicAuth, None]
timeout: "Union[ClientTimeout, _SENTINEL, None]"
ssl: Union[SSLContext, bool, Fingerprint]
server_hostname: Union[str, None]
proxy_headers: Union[LooseHeaders, None]
trace_request_ctx: Union[SimpleNamespace, None]
read_bufsize: Union[int, None]
auto_decompress: Union[bool, None]
max_line_size: Union[int, None]
max_field_size: Union[int, None]


@attr.s(auto_attribs=True, frozen=True, slots=True)
class ClientTimeout:
Expand Down Expand Up @@ -1024,61 +1056,111 @@ def _prepare_headers(self, headers: Optional[LooseHeaders]) -> "CIMultiDict[str]
added_names.add(key)
return result

def get(
self, url: StrOrURL, *, allow_redirects: bool = True, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP GET request."""
return _RequestContextManager(
self._request(hdrs.METH_GET, url, allow_redirects=allow_redirects, **kwargs)
)
if sys.version_info >= (3, 11) and TYPE_CHECKING:

def get(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

def options(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

def head(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

def post(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

def put(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

def patch(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

def delete(
self,
url: StrOrURL,
**kwargs: Unpack[_RequestOptions],
) -> "_RequestContextManager": ...

else:

def get(
self, url: StrOrURL, *, allow_redirects: bool = True, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP GET request."""
return _RequestContextManager(
self._request(
hdrs.METH_GET, url, allow_redirects=allow_redirects, **kwargs
)
)

def options(
self, url: StrOrURL, *, allow_redirects: bool = True, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP OPTIONS request."""
return _RequestContextManager(
self._request(
hdrs.METH_OPTIONS, url, allow_redirects=allow_redirects, **kwargs
def options(
self, url: StrOrURL, *, allow_redirects: bool = True, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP OPTIONS request."""
return _RequestContextManager(
self._request(
hdrs.METH_OPTIONS, url, allow_redirects=allow_redirects, **kwargs
)
)
)

def head(
self, url: StrOrURL, *, allow_redirects: bool = False, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP HEAD request."""
return _RequestContextManager(
self._request(
hdrs.METH_HEAD, url, allow_redirects=allow_redirects, **kwargs
def head(
self, url: StrOrURL, *, allow_redirects: bool = False, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP HEAD request."""
return _RequestContextManager(
self._request(
hdrs.METH_HEAD, url, allow_redirects=allow_redirects, **kwargs
)
)
)

def post(
self, url: StrOrURL, *, data: Any = None, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP POST request."""
return _RequestContextManager(
self._request(hdrs.METH_POST, url, data=data, **kwargs)
)
def post(
self, url: StrOrURL, *, data: Any = None, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP POST request."""
return _RequestContextManager(
self._request(hdrs.METH_POST, url, data=data, **kwargs)
)

def put(
self, url: StrOrURL, *, data: Any = None, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP PUT request."""
return _RequestContextManager(
self._request(hdrs.METH_PUT, url, data=data, **kwargs)
)
def put(
self, url: StrOrURL, *, data: Any = None, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP PUT request."""
return _RequestContextManager(
self._request(hdrs.METH_PUT, url, data=data, **kwargs)
)

def patch(
self, url: StrOrURL, *, data: Any = None, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP PATCH request."""
return _RequestContextManager(
self._request(hdrs.METH_PATCH, url, data=data, **kwargs)
)
def patch(
self, url: StrOrURL, *, data: Any = None, **kwargs: Any
) -> "_RequestContextManager":
"""Perform HTTP PATCH request."""
return _RequestContextManager(
self._request(hdrs.METH_PATCH, url, data=data, **kwargs)
)

def delete(self, url: StrOrURL, **kwargs: Any) -> "_RequestContextManager":
"""Perform HTTP DELETE request."""
return _RequestContextManager(self._request(hdrs.METH_DELETE, url, **kwargs))
def delete(self, url: StrOrURL, **kwargs: Any) -> "_RequestContextManager":
"""Perform HTTP DELETE request."""
return _RequestContextManager(
self._request(hdrs.METH_DELETE, url, **kwargs)
)

async def close(self) -> None:
"""Close underlying connector.
Expand Down