(self)
return ProxyConnector.from_url(proxy_url)
Check env for a SOCKS proxy URL and return a connector if found
Check env for a SOCKS proxy URL and return a connector if found
Check env for a SOCKS proxy URL and return a connector if found
def _check_for_socks_proxy(self) -> Optional[ProxyConnector]:
        """Check env for a SOCKS proxy URL and return a connector if found"""
        proxy_vars = (
            "https_proxy",
            "http_proxy",
            "all_proxy",
        )
        socks_proxy_re = re.compile(r"^socks[45]h?:\/\/.+")

        proxy_url = None
        for proxy_var in proxy_vars:
            for pv in (proxy_var, proxy_var.upper()):
                proxy_url = environ.get(pv)
                if proxy_url:
                    break
            if proxy_url:
                break

        if not proxy_url or not socks_proxy_re.match(proxy_url):
            return None

        logger.debug(f"Creating a SOCKS ProxyConnector to use {proxy_url}")
        return ProxyConnector.from_url(proxy_url)
def _check_for_socks_proxy ( self ) -> Optional [ ProxyConnector ] : proxy_vars = ( "https_proxy" , "http_proxy" , "all_proxy" , ) socks_proxy_re = re . compile ( r"^socks[45]h?:\/\/.+" ) proxy_url = None for proxy_var in proxy_vars : for pv in ( proxy_var , proxy_var . upper ( ) ) : proxy_url = environ . get ( pv ) if proxy_url : break if proxy_url : break if not proxy_url or not socks_proxy_re . match ( proxy_url ) : return None logger . debug ( f"Creating a SOCKS ProxyConnector to use {proxy_url}" ) return ProxyConnector . from_url ( proxy_url )