(
self,
num_pools: int = 10,
headers: typing.Mapping[str, str] | None = None,
**connection_pool_kw: typing.Any,
)
| 197 | proxy_config: ProxyConfig | None = None |
| 198 | |
| 199 | def __init__( |
| 200 | self, |
| 201 | num_pools: int = 10, |
| 202 | headers: typing.Mapping[str, str] | None = None, |
| 203 | **connection_pool_kw: typing.Any, |
| 204 | ) -> None: |
| 205 | super().__init__(headers) |
| 206 | # PoolManager handles redirects itself in PoolManager.urlopen(). |
| 207 | # It always passes redirect=False to the underlying connection pool to |
| 208 | # suppress per-pool redirect handling. If the user supplied a non-Retry |
| 209 | # value (int/bool/etc) for retries and we let the pool normalize it |
| 210 | # while redirect=False, the resulting Retry object would have redirect |
| 211 | # handling disabled, which can interfere with PoolManager's own |
| 212 | # redirect logic. Normalize here so redirects remain governed solely by |
| 213 | # PoolManager logic. |
| 214 | if "retries" in connection_pool_kw: |
| 215 | retries = connection_pool_kw["retries"] |
| 216 | if not isinstance(retries, Retry): |
| 217 | retries = Retry.from_int(retries) |
| 218 | connection_pool_kw = connection_pool_kw.copy() |
| 219 | connection_pool_kw["retries"] = retries |
| 220 | self.connection_pool_kw = connection_pool_kw |
| 221 | |
| 222 | self.pools: RecentlyUsedContainer[PoolKey, HTTPConnectionPool] |
| 223 | self.pools = RecentlyUsedContainer(num_pools) |
| 224 | |
| 225 | # Locally set the pool classes and keys so other PoolManagers can |
| 226 | # override them. |
| 227 | self.pool_classes_by_scheme = pool_classes_by_scheme |
| 228 | self.key_fn_by_scheme = key_fn_by_scheme.copy() |
| 229 | |
| 230 | def __enter__(self) -> Self: |
| 231 | return self |
no test coverage detected