feat: add BrightData provider

Author: davidteather

.sphinx/providers/brightdata.rst

@@ -0,0 +1,4 @@
+.. automodule:: proxyproviders.providers.brightdata
+   :members:
+   :undoc-members:
+   :show-inheritance:

.sphinx/providers/webshare.rst

@@ -1,3 +1,4 @@
 .. automodule:: proxyproviders.providers.webshare
    :members:
+   :undoc-members:
    :show-inheritance:

.sphinx/proxyproviders.rst

@@ -4,8 +4,12 @@ ProxyProviders Supported Providers
 Current Providers Supported:
 
 * `Webshare.io <https://www.webshare.io/?referral_code=3x5812idzzzp>`_ (referral link) - try out 10 free datacenter proxies
+* `BrightData <https://get.brightdata.com/davidteather>`_ (affiliate link)
 
 Don't see a provider you need? Open an issue or submit a PR!
 
 .. _webshare:
-.. include:: providers/webshare.rst
+.. include:: providers/webshare.rst
+
+.. _brightdata:
+.. include:: providers/brightdata.rst

README.md

@@ -98,7 +98,7 @@ def some_function(provider: ProxyProvider):
     print(proxies)
 
 webshare = Webshare(api_key="your_api_key")
-brightdata = BrightData(api_key="your_api_key")
+brightdata = BrightData(api_key="your_api_key", zone="my_zone")
 
 some_function(webshare)
 some_function(brightdata)

proxyproviders/__init__.py

@@ -1,5 +1,6 @@
 from .proxy_provider import ProxyProvider, ProxyConfig
 from .providers.webshare import Webshare
+from .providers.brightdata import BrightData
 from .models.proxy import Proxy
 
-__all__ = ["ProxyProvider", "ProxyConfig", "Webshare", "Proxy"]
+__all__ = ["ProxyProvider", "ProxyConfig", "Proxy", "Webshare", "BrightData"]

proxyproviders/models/proxy.py

@@ -30,5 +30,5 @@ class Proxy:
     created_at: Optional[datetime] = None
     """The timestamp when the proxy was created. Optional"""
 
-    protocols: List[str] = None
+    protocols: Optional[List[str]] = None
     """A list of connection protocols supported by the proxy, e.g., ['http', 'https']"""

proxyproviders/providers/brightdata.py

@@ -0,0 +1,191 @@
+import requests
+from typing import List, Dict, Optional
+from ..proxy_provider import ProxyProvider, ProxyConfig
+from ..exceptions import ProxyFetchException,ProxyInvalidResponseException
+from ..models.proxy import Proxy
+import threading
+
+class BrightData(ProxyProvider):
+    """BrightData (formerly luminati) is a proxy provider that offers residential and datacenter proxies.
+
+    Create an account `here <https://get.brightdata.com/davidteather>`_ (affiliate link).
+    
+    You can find your API key in the account settings `here <https://brightdata.com/cp/setting/users>`_ then create "add token" with scope "limit" (`BrightData article <https://docs.brightdata.com/general/account/api-token>`_ for more info)
+
+    The BrightData API documentation is `here <https://docs.brightdata.com/api-reference/account-management-api/Get_active_Zones?playground=open>`_
+
+    :param api_key: Your BrightData API key
+    :param zone: The zone ID/name to fetch proxies for (you can get this from the BrightData dashboard)
+    :param username_suffix: Optional suffix to append to the username `more info <https://docs.brightdata.com/proxy-networks/config-options>`_ allows you to target region, city, etc. (requires use_super_proxy=True)
+    :param use_super_proxy: Optional flag to use super proxy instead of targeting specific IPs, this is enabled by default. If you want to target specific IPs or have consistent IPs for a session, set this to False.
+    :param config: Configuration for the proxy provider. View the ProxyConfig class docs for more information.
+
+    Example:
+
+    .. code-block:: python
+
+        from proxyproviders import BrightData
+
+        # Initialize the BrightData API client with an API key and a zone
+        proxy_provider = BrightData(api_key="your-api-key", zone="my_zone")
+
+        # Fetch proxies
+        proxies = proxy_provider.list_proxies() # returns one proxy for super proxy by default
+
+        # If you want to manage specific IPs
+        proxy_provider = BrightData(api_key="your-api-key", zone="my_zone", use_super_proxy=False)
+        proxies = proxy_provider.list_proxies() # returns multiple proxies for each IP in the zone (potentially thousands)
+    """
+    _BASE_URL = "https://api.brightdata.com"
+    _SUPER_PROXY_ADDRESS = "brd.superproxy.io"
+    _SUPER_PROXY_PORT = 33335
+    _PROTOCOLS: List[str] = ["http", "https"] # BrightData supports both HTTP and HTTPS
+
+    def __init__(self, api_key: str, zone: str, username_suffix: Optional[str] = None, use_super_proxy: Optional[bool] = True, config: Optional[ProxyConfig] = None):
+        super().__init__(config)
+        self.api_key = api_key
+        self.zone = zone
+        self.username_suffix = username_suffix
+        self.use_super_proxy = use_super_proxy
+
+    def _fetch_proxies(self) -> List[Proxy]:
+        username = self.get_zone_username(self.zone)
+        if self.username_suffix:
+            username += self.username_suffix
+
+        passwords = self.get_zone_passwords(self.zone)
+
+        if self.use_super_proxy:
+            # Let the super proxy handle the IP rotation
+           return [Proxy(
+                id="super",
+                username=username,
+                password=passwords["passwords"][0],
+                proxy_address=self._SUPER_PROXY_ADDRESS,
+                port=self._SUPER_PROXY_PORT,
+                protocols=self._PROTOCOLS,
+            )]
+        
+        proxies = []
+
+        # Fetch all IPs in the zone, and create a proxy for each
+        # Brightdata doesn't let us target directly so we tell the superproxy what to do
+        ips = self.list_all_ips_in_zone(self.zone)
+
+        for ip in ips:
+            ip_targeted_username = username + f"-ip-{ip['ip']}"
+
+            proxies.append(Proxy(
+                id=ip["ip"],
+                username=ip_targeted_username,
+                password=passwords["passwords"][0],
+                proxy_address=self._SUPER_PROXY_ADDRESS,
+                port=self._SUPER_PROXY_PORT,
+                country_code=ip["country"],
+                protocols=self._PROTOCOLS,
+            ))
+
+        return proxies
+
+
+    def get_active_zones(self) -> Dict:
+        """Fetches active zones from BrightData API.
+        
+        Response:
+
+        .. code-block:: json
+
+            [
+                {
+                    "name": "zone1",
+                    "type": "dc",
+                }
+            ]
+
+        """
+        return self._make_request("/zone/get_active_zones", "GET")
+    
+    def get_zone_username(self, zone: str) -> str:
+        """Fetches zone username for the given zone ID from BrightData API.
+        
+        Note: this isn't directly an API endpoint, I'm sort of reconstructing some things here and it seems to behave a little weird.
+
+        :param zone: The zone ID to fetch username for
+        """
+
+        data = self._make_request(f"/status", "GET", params={"zone": zone})
+        
+        customer = data.get("customer")
+        if not customer:
+            raise ProxyInvalidResponseException("Failed to fetch customer data")
+        
+        return f"brd-customer-{customer}-zone-{zone}"
+
+
+    def get_zone_passwords(self, zone: str) -> Dict:
+        """Fetches zone passwords from BrightData API.
+        
+        :param zone: The zone ID to fetch passwords for
+
+        Response:
+
+
+        .. code-block:: json
+
+            {
+                "passwords": [
+                    "password1",
+                    "password2",
+                ]
+            }
+
+        """
+        return self._make_request(f"/zone/passwords", "GET", params={"zone": zone})
+    
+    def list_all_ips_in_zone(self, zone: str, country: Optional[str] = None) -> Dict:
+        """Fetches all IPs in a zone from BrightData API.
+        
+        :param zone: The zone ID to fetch IPs for
+        :param country: Optional 2-letter country code to filter IPs by
+
+        Response:
+
+
+        .. code-block:: json
+
+            [
+                {
+                    "ip": "192.168.1.1",
+                    "country": "US",
+                }
+            ]
+        
+        """
+        return self._make_request(f"/zone/route_ips", "GET", params={"zone": zone, "list_countries": True, "country": country})
+                   
+    def _make_request(self, path: str, method: str, params: Optional[Dict] = None, json: Optional[Dict] = None) -> Dict:
+        """Makes a request to the BrightData API.
+        
+        :param path: The path to the endpoint
+        :param method: The HTTP method to use
+        :param params: Optional parameters to include in the request
+        :param json: Optional JSON data to include in the request
+        """
+
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+        }
+
+        url = f"{self._BASE_URL}{path}"
+        response = requests.request(method, url, headers=headers, params=params, json=json)
+
+        if response.status_code != 200:
+            raise ProxyFetchException(f"Failed to fetch from BrightData, got status code {response.status_code}, text: {response.text}")
+
+        try:    
+            data = response.json()
+        except Exception as e:
+            raise ProxyInvalidResponseException(f"Failed to parse response: {str(e)}") from e
+        
+
+        return data

tests/conftest.py

@@ -1,6 +1,7 @@
 import os
 import pytest
 from proxyproviders.providers.webshare import Webshare
+from proxyproviders.providers.brightdata import BrightData
 from proxyproviders.proxy_provider import ProxyConfig
 
 @pytest.fixture
@@ -11,3 +12,13 @@ def webshare_provider():
     """
     api_key = os.getenv("WEBSHARE_API_KEY", "test-api-key")
     return Webshare(api_key=api_key, config=ProxyConfig(refresh_interval=0))
+
+@pytest.fixture
+def brightdata_provider():
+    """
+    Fixture to create a BrightData provider instance for unit tests.
+    """
+    # Use dummy credentials for testing
+    api_key = os.getenv("BRIGHTDATA_API_KEY", "test-brightdata-api-key")
+    zone = "test-zone"
+    return BrightData(api_key=api_key, zone=zone)

tests/integration/test_brightdata_integration.py

@@ -0,0 +1,43 @@
+import os
+import pytest
+import responses
+from proxyproviders.providers.brightdata import BrightData
+from proxyproviders.exceptions import (
+    ProxyFetchException,
+    ProxyConversionException,
+    ProxyInvalidResponseException,
+)
+from proxyproviders.models.proxy import Proxy
+
+skip_integration = pytest.mark.skipif(
+    not (
+        os.getenv("RUN_INTEGRATION_TESTS")
+        and os.getenv("BRIGHTDATA_API_KEY")
+    ),
+    reason="Integration tests require RUN_INTEGRATION_TESTS flag and valid BRIGHTDATA_API_KEY",
+)
+
+@skip_integration
+def test_brightdata_integration():
+    """
+    Integration test: Perform a real API call to BrightData.
+    This sanity test ensures that the provider's structure and conversion functions
+    work correctly when interacting with the live API.
+    """
+    api_key = os.getenv("BRIGHTDATA_API_KEY")
+    provider = BrightData(api_key=api_key, zone="static")
+    
+    # List proxies; note that behavior will vary based on provider.use_super_proxy.
+    proxies = provider.list_proxies(force_refresh=True)
+    
+    assert isinstance(proxies, list)
+    # If using super proxy mode, one proxy is expected; otherwise, there may be many.
+    if provider.use_super_proxy:
+        assert len(proxies) == 1
+        proxy = proxies[0]
+        assert hasattr(proxy, "port")
+        assert isinstance(proxy.port, (int, str))
+    else:
+        for proxy in proxies:
+            assert hasattr(proxy, "port")
+            assert isinstance(proxy.port, (int, str))