Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if proxy:
del kwargs['proxy']
elif hasattr(self, 'nv_config') and self.nv_config.get('proxy'):
proxy = self.nv_config.get('proxy')
if proxy:
host, port = proxy.rsplit(':', 1)
kwargs['proxy_host'] = host
kwargs['proxy_port'] = int(port)
params = kwargs.get('params')
if params:
del kwargs['params']
q = urlencode(params)
url += '?' + q
kwargs.setdefault("headers", {}).setdefault('User-Agent', DEFAULT_USER_AGENT)
r = HTTPRequest(url, **kwargs)
return ResponseManager(r)
async def _request(self, *args, **kwargs):
if hasattr(self, "nv_config") and self.nv_config.get("proxy"):
kwargs.setdefault("proxy", self.nv_config.get("proxy"))
kwargs.setdefault("headers", {}).setdefault('User-Agent', DEFAULT_USER_AGENT)
res = await super(BetterClientSession, self)._request(
*args, **kwargs)
if res.status >= 400:
raise HTTPError(res.status, res.reason, res)
return res
def setup_session(proxy: Optional[str], concurrency: int):
s = httpx.AsyncClient(
pool_limits = httpx.PoolLimits(
soft_limit = concurrency * 2,
hard_limit = concurrency,
),
timeout = httpx.Timeout(20, pool_timeout=None),
headers = {'User-Agent': DEFAULT_USER_AGENT},
http2 = True,
proxies = proxy,
)
session.set(Session(s, concurrency))