Method new_proxy_info

src/crawlee/proxy_configuration.py:114–157

Return a new ProxyInfo object based on the configured proxy rotation strategy. Args: session_id: Session identifier. If provided, same proxy URL will be returned for subsequent calls with this ID. Will be auto-generated for tiered proxies if not p

(
        self, session_id: str | None, request: Request | None, proxy_tier: int | None
    )

        return URL(url)

    async def new_proxy_info(
        self, session_id: str | None, request: Request | None, proxy_tier: int | None
    ) -> ProxyInfo | None:
        """Return a new ProxyInfo object based on the configured proxy rotation strategy.

        Args:
            session_id: Session identifier. If provided, same proxy URL will be returned for
                subsequent calls with this ID. Will be auto-generated for tiered proxies if
                not provided.
            request: Request object used for proxy rotation and tier selection. Required for
                tiered proxies to track retries and adjust tier accordingly.
            proxy_tier: Specific proxy tier to use. If not provided, will be automatically
                selected based on configuration.
        """
        if self._proxy_tier_tracker is not None and session_id is None:
            session_id = crypto_random_object_id(6)

        url, proxy_tier = await self._pick_url(session_id, request, proxy_tier)

        if url is None:
            return None

        if url.port is None:
            raise ValueError(f'Port is None for URL: {url}')

        if url.host is None:
            raise ValueError(f'Host is None for URL: {url}')

        info = ProxyInfo(
            url=str(url),
            scheme=url.scheme,
            hostname=url.host,
            port=url.port,
            username=url.user or '',
            password=url.password or '',
        )

        if session_id is not None:
            info.session_id = session_id

        if proxy_tier is not None:
            info.proxy_tier = proxy_tier

        return info

    async def new_url(
        self, session_id: str | None = None, request: Request | None = None, proxy_tier: int | None = None