Using Algolia Python API client with custom Requester (e.g. to pass proxies)

app.py

import os
from algoliasearch.search_client import SearchClient
from algoliasearch.configs import SearchConfig
from algoliasearch.http.transporter import Transporter
from custom_requester import CustomRequester

app_id = ""
api_key = ""
index_name = ""

requester = CustomRequester()
config = SearchConfig(app_id, api_key)
transporter = Transporter(requester, config)

client = SearchClient(transporter, config)
index = client.init_index(index_name)

obj = index.get_object("one")
print(obj)

custom_requester.py

import requests

from requests import Timeout, RequestException
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util import Retry

from algoliasearch.http.transporter import Response, Request


class CustomRequester(object):

    def __init__(self):
        # type: () -> None

        self.session = requests.Session()  # type: ignore
        self.proxies = {
            "http" : "localhost:8080",
            "https" : "localhost:8080"
        }

        # Ask urllib not to make retries on its own.
        self.session.mount(
            'https://', HTTPAdapter(max_retries=Retry(connect=0))
        )

    def send(self, request):
        # type: (Request) -> Response

        req = requests.Request(method=request.verb, url=request.url,
                               headers=request.headers,
                               data=request.data_as_string)

        r = req.prepare()  # type: ignore

        requests_timeout = (request.connect_timeout, request.timeout)

        try:
            print("> Sending request...")
            response = self.session.send(  # type: ignore
                r, timeout=requests_timeout, proxies=self.proxies
            )
        except Timeout as e:
            return Response(error_message=str(e), is_timed_out_error=True)
        except RequestException as e:
            return Response(error_message=str(e), is_network_error=True)

        return Response(
            response.status_code,
            response.json(),
            response.reason
        )