Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
987c4bdf9c | ||
|
|
250c9cf2eb | ||
|
|
b009c0cf50 | ||
|
|
e6eddf5bc3 |
No files matched your search
@@ -11,19 +11,19 @@ and swap the HTTP client while inheriting everything else.
|
|||||||
`requirements.txt`:
|
`requirements.txt`:
|
||||||
|
|
||||||
```
|
```
|
||||||
aioweb @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioweb.git@v1.1.0
|
aioweb @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioweb.git@v1.3.0
|
||||||
```
|
```
|
||||||
|
|
||||||
Direct:
|
Direct:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install "aioweb @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioweb.git@v1.1.0"
|
pip install "aioweb @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioweb.git@v1.3.0"
|
||||||
```
|
```
|
||||||
|
|
||||||
Requires `aiohttp` and `yarl` (pulled transitively), plus the sibling `commons` library
|
Requires `aiohttp` and `yarl` (pulled transitively), plus the sibling `commons` library
|
||||||
(a private `git+ssh` package — the install needs access to the `rethink-public` org).
|
(a private `git+ssh` package — the install needs access to the `rethink-public` org).
|
||||||
|
|
||||||
Drop the `@v1.1.0` suffix from the line above to install the latest unpinned.
|
Drop the `@v1.3.0` suffix from the line above to install the latest unpinned.
|
||||||
|
|
||||||
## Usage
|
## Usage
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "aioweb"
|
name = "aioweb"
|
||||||
version = "1.1.0"
|
version = "1.3.0"
|
||||||
description = "Async HTTP session wrapper over aiohttp — proxies, header overwrites, retries, previews. Config-free, installable."
|
description = "Async HTTP session wrapper over aiohttp — proxies, header overwrites, retries, previews. Config-free, installable."
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
+49
-31
@@ -15,6 +15,7 @@ import warnings
|
|||||||
from http.cookies import SimpleCookie
|
from http.cookies import SimpleCookie
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
from multidict import CIMultiDict
|
||||||
from yarl import URL
|
from yarl import URL
|
||||||
from commons import aretry
|
from commons import aretry
|
||||||
|
|
||||||
@@ -85,12 +86,15 @@ class ExtendedSession:
|
|||||||
kwargs: passed through to aiohttp.ClientSession
|
kwargs: passed through to aiohttp.ClientSession
|
||||||
"""
|
"""
|
||||||
self.inject = inject
|
self.inject = inject
|
||||||
self.header_overwrites = overwrites or {}
|
# copied, not aliased: a shared config dict must not be mutated by (or leak a
|
||||||
self.domain_overwrites = domain_overwrites or {}
|
# clear_proxies() into) a sibling session
|
||||||
|
self.header_overwrites = dict(overwrites or {})
|
||||||
|
self.domain_overwrites = dict(domain_overwrites or {})
|
||||||
self.ephemeral_headers = {}
|
self.ephemeral_headers = {}
|
||||||
self.proxies = proxies or {}
|
self.proxies = dict(proxies or {})
|
||||||
# own header layer, not aiohttp privates, so update_headers/clear_headers work
|
# own header layer, not aiohttp privates, so update_headers/clear_headers work;
|
||||||
self._default_headers = dict(headers or {})
|
# case-insensitive like HTTP header names
|
||||||
|
self._default_headers = CIMultiDict(headers or {})
|
||||||
self._session_timeout = timeout
|
self._session_timeout = timeout
|
||||||
self._session_kwargs = kwargs
|
self._session_kwargs = kwargs
|
||||||
# aiohttp>=3.14 needs a running loop to build ClientSession; built lazily instead
|
# aiohttp>=3.14 needs a running loop to build ClientSession; built lazily instead
|
||||||
@@ -139,8 +143,12 @@ class ExtendedSession:
|
|||||||
self.header_overwrites[header] = value
|
self.header_overwrites[header] = value
|
||||||
|
|
||||||
def _apply_overwrites(self, request_headers):
|
def _apply_overwrites(self, request_headers):
|
||||||
"""apply static overwrites and ephemeral headers to a request's headers"""
|
"""apply static overwrites and ephemeral headers to a request's headers
|
||||||
request_headers = dict(request_headers or {})
|
|
||||||
|
header names match case-insensitively, so an overwrite replaces a `user-agent`
|
||||||
|
rather than sending it alongside a second `User-Agent`
|
||||||
|
"""
|
||||||
|
request_headers = CIMultiDict(request_headers or {})
|
||||||
# list()-snapshot so a concurrent mutation of the shared dicts can't raise
|
# list()-snapshot so a concurrent mutation of the shared dicts can't raise
|
||||||
# "dict changed size during iteration"
|
# "dict changed size during iteration"
|
||||||
for header, value in list(self.header_overwrites.items()):
|
for header, value in list(self.header_overwrites.items()):
|
||||||
@@ -153,7 +161,7 @@ class ExtendedSession:
|
|||||||
request_headers.update(value)
|
request_headers.update(value)
|
||||||
else:
|
else:
|
||||||
request_headers[header] = value
|
request_headers[header] = value
|
||||||
return request_headers
|
return dict(request_headers)
|
||||||
|
|
||||||
def set_ephemeral(self, header, value_callable):
|
def set_ephemeral(self, header, value_callable):
|
||||||
"""register a header generated fresh (via callable) for each request"""
|
"""register a header generated fresh (via callable) for each request"""
|
||||||
@@ -262,7 +270,7 @@ class ExtendedSession:
|
|||||||
falling back to them.
|
falling back to them.
|
||||||
"""
|
"""
|
||||||
proxies = self.proxies if proxies is None else proxies
|
proxies = self.proxies if proxies is None else proxies
|
||||||
scheme = url.split("://")[0]
|
scheme = URL(url).scheme
|
||||||
return proxies.get(scheme)
|
return proxies.get(scheme)
|
||||||
|
|
||||||
# -------------------------------------------------------------------------
|
# -------------------------------------------------------------------------
|
||||||
@@ -353,7 +361,8 @@ class ExtendedSession:
|
|||||||
"""
|
"""
|
||||||
url = self._apply_domain_overwrites(url)
|
url = self._apply_domain_overwrites(url)
|
||||||
proxy = self._get_proxy(url, kwargs.pop("proxies", None))
|
proxy = self._get_proxy(url, kwargs.pop("proxies", None))
|
||||||
merged = {**self._default_headers, **(kwargs.pop("headers", None) or {})}
|
merged = CIMultiDict(self._default_headers)
|
||||||
|
merged.update(kwargs.pop("headers", None) or {})
|
||||||
headers = self._apply_overwrites(merged)
|
headers = self._apply_overwrites(merged)
|
||||||
|
|
||||||
timeout = kwargs.get("timeout")
|
timeout = kwargs.get("timeout")
|
||||||
@@ -421,22 +430,19 @@ class ExtendedSession:
|
|||||||
kwargs["proxy"] = resolved_proxy
|
kwargs["proxy"] = resolved_proxy
|
||||||
debug = kwargs.pop("debug", False)
|
debug = kwargs.pop("debug", False)
|
||||||
|
|
||||||
merged = {**self._default_headers, **(kwargs.get("headers") or {})}
|
merged = CIMultiDict(self._default_headers)
|
||||||
|
merged.update(kwargs.get("headers") or {})
|
||||||
kwargs["headers"] = self._apply_overwrites(merged)
|
kwargs["headers"] = self._apply_overwrites(merged)
|
||||||
kwargs["headers"] = {str(k): str(v) for k, v in kwargs["headers"].items()}
|
kwargs["headers"] = {str(k): str(v) for k, v in kwargs["headers"].items()}
|
||||||
|
|
||||||
timeout = kwargs.get("timeout")
|
timeout = kwargs.get("timeout")
|
||||||
# the hang-guard deadline, captured as a bare number BEFORE the ClientTimeout
|
# the hang-guard deadline: a per-call timeout (a number, or a ClientTimeout's
|
||||||
# conversion below consumes it: a per-call numeric timeout, else the session
|
# .total), else the session default. a timeout=None (omitted, or the explicit None
|
||||||
# default. a timeout=None (omitted, or the explicit None request_with_retries
|
# request_with_retries passes) means "use the session bound" - so it guards on
|
||||||
# passes) means "use the session bound" - so it guards on self._session_timeout,
|
# self._session_timeout, NOT unguarded. unguarded only when the caller asked for no
|
||||||
# NOT unguarded. the ONLY unguarded case is a session deliberately constructed
|
# bound: a session constructed with timeout=None (the documented long-poll/streaming
|
||||||
# unbounded (Session(timeout=None) -> self._session_timeout is None): the
|
# opt-out), or a per-call ClientTimeout(total=None).
|
||||||
# documented opt-out for a genuinely unbounded call (long-poll/streaming).
|
guard_secs = self._session_timeout if timeout is None else getattr(timeout, "total", timeout)
|
||||||
if isinstance(timeout, (int, float)):
|
|
||||||
guard_secs = timeout
|
|
||||||
else:
|
|
||||||
guard_secs = self._session_timeout
|
|
||||||
if isinstance(timeout, (int, float)):
|
if isinstance(timeout, (int, float)):
|
||||||
kwargs["timeout"] = aiohttp.ClientTimeout(total=timeout)
|
kwargs["timeout"] = aiohttp.ClientTimeout(total=timeout)
|
||||||
elif timeout is None and "timeout" in kwargs:
|
elif timeout is None and "timeout" in kwargs:
|
||||||
@@ -470,10 +476,12 @@ class ExtendedSession:
|
|||||||
# get a typed failure and request_with_retries can label it a timeout. now also
|
# get a typed failure and request_with_retries can label it a timeout. now also
|
||||||
# catches the hang-guard firing (asyncio.wait_for raises asyncio.TimeoutError).
|
# catches the hang-guard firing (asyncio.wait_for raises asyncio.TimeoutError).
|
||||||
raise aiohttp.ServerTimeoutError(f"timeout for {url}: {error}") from error
|
raise aiohttp.ServerTimeoutError(f"timeout for {url}: {error}") from error
|
||||||
except aiohttp.ClientError as error:
|
except aiohttp.ClientError:
|
||||||
# re-raise the original subtype (not flattened) - request_with_retries still
|
# re-raise the original subtype (not flattened) - request_with_retries still
|
||||||
# catches the base aiohttp.ClientError below and is unaffected
|
# catches the base aiohttp.ClientError below and is unaffected. no log here:
|
||||||
log.error("client error for %s: %s", url, error)
|
# this path RAISES, so the exception carries the failure (raise XOR log); on the
|
||||||
|
# retrying path commons.aretry emits the per-attempt WARNING and the terminal
|
||||||
|
# branch logs on exhaustion - logging here would double-report the same failure.
|
||||||
raise
|
raise
|
||||||
|
|
||||||
async def request_with_retries(
|
async def request_with_retries(
|
||||||
@@ -507,7 +515,11 @@ class ExtendedSession:
|
|||||||
headers=headers, proxies=proxies, timeout=timeout, debug=debug,
|
headers=headers, proxies=proxies, timeout=timeout, debug=debug,
|
||||||
)
|
)
|
||||||
if response.status_code in retry_statuses:
|
if response.status_code in retry_statuses:
|
||||||
log.warning("retryable status %s for %s", response.status_code, url)
|
# debug, not warning: this RAISES _RetryStatus to trigger a retry, so it must
|
||||||
|
# not also warn/error the same failure (raise XOR log). commons.aretry emits
|
||||||
|
# the per-attempt WARNING with the attempt count when it catches and retries;
|
||||||
|
# the terminal branch below logs on exhaustion.
|
||||||
|
log.debug("retryable status %s for %s", response.status_code, url)
|
||||||
raise _RetryStatus(response)
|
raise _RetryStatus(response)
|
||||||
return response
|
return response
|
||||||
|
|
||||||
@@ -516,19 +528,25 @@ class ExtendedSession:
|
|||||||
attempt, attempts=attempts, backoff=1.0, factor=backoff_base,
|
attempt, attempts=attempts, backoff=1.0, factor=backoff_base,
|
||||||
jitter=False, on=(Exception,),
|
jitter=False, on=(Exception,),
|
||||||
)
|
)
|
||||||
|
# terminal path: retries are exhausted and the failure is SWALLOWED into a falsy
|
||||||
|
# FailureResponse the caller branches on (the documented contract). rule 2 mandates a
|
||||||
|
# log on a swallow; level is WARNING because the lib recovered cleanly into a
|
||||||
|
# branchable value - the failing op is the caller's to escalate once it sees the falsy
|
||||||
|
# result, so the lib does not claim ERROR on the caller's behalf. all four branches are
|
||||||
|
# the same event class (exhausted retries) and stay at one uniform level.
|
||||||
except _RetryStatus as exhausted:
|
except _RetryStatus as exhausted:
|
||||||
log.error("all %d attempts failed for %s (last status %s)",
|
log.warning("all %d attempts failed for %s (last status %s)",
|
||||||
attempts, url, exhausted.response.status_code)
|
attempts, url, exhausted.response.status_code)
|
||||||
return exhausted.response
|
return exhausted.response
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
# catch before ClientError so a timeout is labeled as such, not generic
|
# catch before ClientError so a timeout is labeled as such, not generic
|
||||||
log.error("all %d attempts timed out for %s", attempts, url)
|
log.warning("all %d attempts timed out for %s", attempts, url)
|
||||||
return FailureResponse(reason="timeout", url=url)
|
return FailureResponse(reason="timeout", url=url)
|
||||||
except aiohttp.ClientError as error:
|
except aiohttp.ClientError as error:
|
||||||
log.error("all %d attempts failed for %s (client error: %s)", attempts, url, error)
|
log.warning("all %d attempts failed for %s (client error: %s)", attempts, url, error)
|
||||||
return FailureResponse(reason=f"client error: {error}", url=url)
|
return FailureResponse(reason=f"client error: {error}", url=url)
|
||||||
except Exception as error:
|
except Exception as error:
|
||||||
log.error("all %d attempts failed for %s (unexpected: %s)", attempts, url, error)
|
log.warning("all %d attempts failed for %s (unexpected: %s)", attempts, url, error)
|
||||||
return FailureResponse(reason=f"unexpected error: {error}", url=url)
|
return FailureResponse(reason=f"unexpected error: {error}", url=url)
|
||||||
|
|
||||||
# -------------------------------------------------------------------------
|
# -------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in new issue
Block a user