python-client/splitio/api/client.py at development · splitio/python-client

565 lines (491 loc) · 22.7 KB
"""Synchronous HTTP Client for split API."""
from collections import namedtuple
import requests
import urllib
import logging
import json
import threading
from urllib3.util import parse_url
from splitio.optional.loaders import HTTPKerberosAuth, OPTIONAL
from splitio.client.config import AuthenticateScheme
from splitio.optional.loaders import aiohttp
from splitio.util.time import get_current_epoch_time_ms
SDK_URL = 'https://sdk.split.io/api'
EVENTS_URL = 'https://events.split.io/api'
AUTH_URL = 'https://auth.split.io/api'
TELEMETRY_URL = 'https://telemetry.split.io/api'
_LOGGER = logging.getLogger(__name__)
_EXC_MSG = '{source} library is throwing exceptions'
HttpResponse = namedtuple('HttpResponse', ['status_code', 'body', 'headers'])
def _build_url(server, path, urls):
    Build URL according to server specified.
    :param server: Server for whith the request is being made.
    :type server: str
    :param path: URL path to be appended to base host.
    :type path: str
    :return: A fully qualified URL.
    :rtype: str
    url = urls[server]
    url += '/' if urls[server][:-1] != '/' else ''
    return urllib.parse.urljoin(url, path)
def _construct_urls(sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
    return {
        'sdk': sdk_url if sdk_url is not None else SDK_URL,
        'events': events_url if events_url is not None else EVENTS_URL,
        'auth': auth_url if auth_url is not None else AUTH_URL,
        'telemetry': telemetry_url if telemetry_url is not None else TELEMETRY_URL,
def _build_basic_headers(sdk_key):
    Build basic headers with auth.
    :param sdk_key: API token used to identify backend calls.
    :type sdk_key: str
    return {
        'Content-Type': 'application/json',
        'Authorization': "Bearer %s" % sdk_key
class HttpClientException(Exception):
    """HTTP Client exception."""
    def __init__(self, message):
        """
        Class constructor.
        :param message: Information on why this exception happened.
        :type message: str
        """
        Exception.__init__(self, message)
class HTTPAdapterWithProxyKerberosAuth(requests.adapters.HTTPAdapter):
    """HTTPAdapter override for Kerberos Proxy auth"""
    def __init__(self, principal=None, password=None):
        requests.adapters.HTTPAdapter.__init__(self)
        self._principal = principal
        self._password = password
    def proxy_headers(self, proxy):
        headers = {}
        if self._principal is not None:
            auth = HTTPKerberosAuth(principal=self._principal, password=self._password)
        else:
            auth = HTTPKerberosAuth()
        negotiate_details = auth.generate_request_header(None, parse_url(proxy).host, is_preemptive=True)
        headers['Proxy-Authorization'] = negotiate_details
        return headers
class HttpClientBase(object, metaclass=abc.ABCMeta):
    """HttpClient wrapper template."""
    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
        """
        Class constructor.
        :param timeout: How many milliseconds to wait until the server responds.
        :type timeout: int
        :param sdk_url: Optional alternative sdk URL.
        :type sdk_url: str
        :param events_url: Optional alternative events URL.
        :type events_url: str
        :param auth_url: Optional alternative auth URL.
        :type auth_url: str
        :param telemetry_url: Optional alternative telemetry URL.
        :type telemetry_url: str
        """
        _LOGGER.debug("Initializing httpclient")
        self._timeout = timeout/1000 if timeout else None # Convert ms to seconds.
        self._urls = _construct_urls(sdk_url, events_url, auth_url, telemetry_url)
    @abc.abstractmethod
    def get(self, server, path, apikey):
        """http get request"""
    @abc.abstractmethod
    def post(self, server, path, apikey):
        """http post request"""
    def set_telemetry_data(self, metric_name, telemetry_runtime_producer):
        """
        Set the data needed for telemetry call
        :param metric_name: metric name for telemetry
        :type metric_name: str
        :param telemetry_runtime_producer: telemetry recording instance
        :type telemetry_runtime_producer: splitio.engine.telemetry.TelemetryRuntimeProducer
        """
        self._telemetry_runtime_producer = telemetry_runtime_producer
        self._metric_name = metric_name
    def is_sdk_endpoint_overridden(self):
        return self._urls['sdk'] != SDK_URL
    def _get_headers(self, extra_headers, sdk_key):
        headers = _build_basic_headers(sdk_key)
        if extra_headers is not None:
            headers.update(extra_headers)
        return headers
    def _record_telemetry(self, status_code, elapsed):
        """
        Record Telemetry info
        :param status_code: http request status code
        :type status_code: int
        :param elapsed: response time elapsed.
        :type status_code: int
        """
        self._telemetry_runtime_producer.record_sync_latency(self._metric_name, elapsed)
        if 200 <= status_code < 300:
            self._telemetry_runtime_producer.record_successful_sync(self._metric_name, get_current_epoch_time_ms())
            return
        self._telemetry_runtime_producer.record_sync_error(self._metric_name, status_code)
class HttpClient(HttpClientBase):
    """HttpClient wrapper."""
    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
        """
        Class constructor.
        :param timeout: How many milliseconds to wait until the server responds.
        :type timeout: int
        :param sdk_url: Optional alternative sdk URL.
        :type sdk_url: str
        :param events_url: Optional alternative events URL.
        :type events_url: str
        :param auth_url: Optional alternative auth URL.
        :type auth_url: str
        :param telemetry_url: Optional alternative telemetry URL.
        :type telemetry_url: str
        """
        HttpClientBase.__init__(self, timeout, sdk_url, events_url, auth_url, telemetry_url)
    def get(self, server, path, sdk_key, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
        """
        Issue a get request.
        :param server: Whether the request is for SDK server, Events server or Auth server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param sdk_key: sdk key.
        :type sdk_key: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        start = get_current_epoch_time_ms()
        try:
            response = requests.get(
                _build_url(server, path, self._urls),
                params=query,
                headers=self._get_headers(extra_headers, sdk_key),
                timeout=self._timeout
            self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
            return HttpResponse(response.status_code, response.text, response.headers)
        except requests.exceptions.ChunkedEncodingError as exc:
            _LOGGER.error("IncompleteRead exception detected: %s", exc)
            return HttpResponse(400, "", {})
        except Exception as exc:  # pylint: disable=broad-except
            raise HttpClientException(_EXC_MSG.format(source='request')) from exc
    def post(self, server, path, sdk_key, body, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
        """
        Issue a POST request.
        :param server: Whether the request is for SDK server or Events server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param sdk_key: sdk key.
        :type sdk_key: str
        :param body: body sent in the request.
        :type body: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        start = get_current_epoch_time_ms()
        try:
            response = requests.post(
                _build_url(server, path, self._urls),
                json=body,
                params=query,
                headers=self._get_headers(extra_headers, sdk_key),
                timeout=self._timeout,
            self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
            return HttpResponse(response.status_code, response.text, response.headers)
        except Exception as exc:  # pylint: disable=broad-except
            raise HttpClientException(_EXC_MSG.format(source='request')) from exc
class HttpClientAsync(HttpClientBase):
    """HttpClientAsync wrapper."""
    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None):
        """
        Class constructor.
        :param timeout: How many milliseconds to wait until the server responds.
        :type timeout: int
        :param sdk_url: Optional alternative sdk URL.
        :type sdk_url: str
        :param events_url: Optional alternative events URL.
        :type events_url: str
        :param auth_url: Optional alternative auth URL.
        :type auth_url: str
        :param telemetry_url: Optional alternative telemetry URL.
        :type telemetry_url: str
        """
        HttpClientBase.__init__(self, timeout, sdk_url, events_url, auth_url, telemetry_url)
        self._session = aiohttp.ClientSession()
    async def get(self, server, path, apikey, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
        """
        Issue a get request.
        :param server: Whether the request is for SDK server, Events server or Auth server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param apikey: api token.
        :type apikey: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        start = get_current_epoch_time_ms()
        headers = self._get_headers(extra_headers, apikey)
        try:
            url = _build_url(server, path, self._urls)
            _LOGGER.debug("GET request: %s", url)
            _LOGGER.debug("query params: %s", query)
            _LOGGER.debug("headers: %s", headers)
            async with self._session.get(
                params=query,
                headers=headers,
                timeout=self._timeout
            ) as response:
                body = await response.text()
                _LOGGER.debug("Response:")
                _LOGGER.debug(response)
                _LOGGER.debug(body)
                await self._record_telemetry(response.status, get_current_epoch_time_ms() - start)
                return HttpResponse(response.status, body, response.headers)
        except aiohttp.ClientPayloadError as exc:
                _LOGGER.error("ContentLengthError exception detected: %s", exc)
                return HttpResponse(400, "", {})
        except aiohttp.ClientError as exc:  # pylint: disable=broad-except
            raise HttpClientException(_EXC_MSG.format(source='aiohttp')) from exc
    async def post(self, server, path, apikey, body, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
        """
        Issue a POST request.
        :param server: Whether the request is for SDK server or Events server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param apikey: api token.
        :type apikey: str
        :param body: body sent in the request.
        :type body: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        headers = self._get_headers(extra_headers, apikey)
        start = get_current_epoch_time_ms()
        try:
            headers['Accept-Encoding'] = 'gzip'
            _LOGGER.debug("POST request: %s", _build_url(server, path, self._urls))
            _LOGGER.debug("query params: %s", query)
            _LOGGER.debug("headers: %s", headers)
            _LOGGER.debug("payload: ")
            _LOGGER.debug(str(json.dumps(body)).encode('utf-8'))
            async with self._session.post(
                _build_url(server, path, self._urls),
                params=query,
                headers=headers,
                json=body,
                timeout=self._timeout
            ) as response:
                body = await response.text()
                _LOGGER.debug("Response:")
                _LOGGER.debug(response)
                _LOGGER.debug(body)
                await self._record_telemetry(response.status, get_current_epoch_time_ms() - start)
                return HttpResponse(response.status, body, response.headers)
        except aiohttp.ClientError as exc:  # pylint: disable=broad-except
            raise HttpClientException(_EXC_MSG.format(source='aiohttp')) from exc
    async def _record_telemetry(self, status_code, elapsed):
        """
        Record Telemetry info
        :param status_code: http request status code
        :type status_code: int
        :param elapsed: response time elapsed.
        :type status_code: int
        """
        await self._telemetry_runtime_producer.record_sync_latency(self._metric_name, elapsed)
        if 200 <= status_code < 300:
            await self._telemetry_runtime_producer.record_successful_sync(self._metric_name, get_current_epoch_time_ms())
            return
        await self._telemetry_runtime_producer.record_sync_error(self._metric_name, status_code)
    async def close_session(self):
        if not self._session.closed:
            await self._session.close()
class HttpClientKerberos(HttpClientBase):
    """HttpClient wrapper."""
    def __init__(self, timeout=None, sdk_url=None, events_url=None, auth_url=None, telemetry_url=None, authentication_scheme=None, authentication_params=None):
        """
        Class constructor.
        :param timeout: How many milliseconds to wait until the server responds.
        :type timeout: int
        :param sdk_url: Optional alternative sdk URL.
        :type sdk_url: str
        :param events_url: Optional alternative events URL.
        :type events_url: str
        :param auth_url: Optional alternative auth URL.
        :type auth_url: str
        :param telemetry_url: Optional alternative telemetry URL.
        :type telemetry_url: str
        :param authentication_scheme: Optional authentication scheme to use.
        :type authentication_scheme: splitio.client.config.AuthenticateScheme
        :param authentication_params: Optional authentication username and password to use.
        :type authentication_params: [str, str]
        """
        _LOGGER.debug("Initializing httpclient for Kerberos auth")
        self._timeout = timeout/1000 if timeout else None # Convert ms to seconds.
        self._urls = _construct_urls(sdk_url, events_url, auth_url, telemetry_url)
        self._authentication_scheme = authentication_scheme
        self._authentication_params = authentication_params
        self._lock = threading.RLock()
        self._sessions = {'sdk': requests.Session(),
                          'events': requests.Session(),
                          'auth': requests.Session(),
                          'telemetry': requests.Session()}
        self._set_authentication()
    def get(self, server, path, sdk_key, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
        """
        Issue a get request.
        :param server: Whether the request is for SDK server, Events server or Auth server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param sdk_key: sdk key.
        :type sdk_key: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        with self._lock:
            start = get_current_epoch_time_ms()
            try:
                return self._do_get(server, path, sdk_key, query, extra_headers, start)
            except requests.exceptions.ProxyError as exc:
                _LOGGER.debug("Proxy Exception caught, resetting the http session")
                self._sessions[server].close()
                self._sessions[server] = requests.Session()
                self._set_authentication(server_name=server)
                    return self._do_get(server, path, sdk_key, query, extra_headers, start)
                except Exception as exc:
                    raise HttpClientException(_EXC_MSG.format(source='request')) from exc
            except Exception as exc:  # pylint: disable=broad-except
                raise HttpClientException(_EXC_MSG.format(source='request')) from exc
    def _do_get(self, server, path, sdk_key, query, extra_headers, start):
        """
        Issue a get request.
        :param server: Whether the request is for SDK server, Events server or Auth server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param sdk_key: sdk key.
        :type sdk_key: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        with self._sessions[server].get(
            _build_url(server, path, self._urls),
            headers=self._get_headers(extra_headers, sdk_key),
            params=query,
            timeout=self._timeout
        ) as response:
            self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
            return HttpResponse(response.status_code, response.text, response.headers)
    def post(self, server, path, sdk_key, body, query=None, extra_headers=None):  # pylint: disable=too-many-arguments
        """
        Issue a POST request.
        :param server: Whether the request is for SDK server or Events server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param sdk_key: sdk key.
        :type sdk_key: str
        :param body: body sent in the request.
        :type body: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        with self._lock:
            start = get_current_epoch_time_ms()
            try:
                return self._do_post(server, path, sdk_key, query, extra_headers, body, start)
            except requests.exceptions.ProxyError as exc:
                _LOGGER.debug("Proxy Exception caught, resetting the http session")
                self._sessions[server].close()
                self._sessions[server] = requests.Session()
                self._set_authentication(server_name=server)
                    return self._do_post(server, path, sdk_key, query, extra_headers, body, start)
                except Exception as exc:
                    raise HttpClientException(_EXC_MSG.format(source='request')) from exc
            except Exception as exc:  # pylint: disable=broad-except
                raise HttpClientException(_EXC_MSG.format(source='request')) from exc
    def _do_post(self, server, path, sdk_key, query, extra_headers, body, start):
        """
        Issue a POST request.
        :param server: Whether the request is for SDK server or Events server.
        :typee server: str
        :param path: path to append to the host url.
        :type path: str
        :param sdk_key: sdk key.
        :type sdk_key: str
        :param body: body sent in the request.
        :type body: str
        :param query: Query string passed as dictionary.
        :type query: dict
        :param extra_headers: key/value pairs of possible extra headers.
        :type extra_headers: dict
        :return: Tuple of status_code & response text
        :rtype: HttpResponse
        """
        with self._sessions[server].post(
            _build_url(server, path, self._urls),
            params=query,
            headers=self._get_headers(extra_headers, sdk_key),
            json=body,
            timeout=self._timeout,
        ) as response:
            self._record_telemetry(response.status_code, get_current_epoch_time_ms() - start)
            return HttpResponse(response.status_code, response.text, response.headers)
    def _set_authentication(self, server_name=None):
        """
        Set the authentication for all self._sessions variables based on authentication scheme.
        :param server: If set, will only add the auth for its session variable, otherwise will set all sessions.
        :typee server: str
        """
        for server in ['sdk', 'events', 'auth', 'telemetry']:
            if server_name is not None and server_name != server:
                continue
            if self._authentication_scheme == AuthenticateScheme.KERBEROS_SPNEGO:
                _LOGGER.debug("Using Kerberos Spnego Authentication")
                if self._authentication_params != [None, None]:
                    self._sessions[server].auth = HTTPKerberosAuth(principal=self._authentication_params[0], password=self._authentication_params[1], mutual_authentication=OPTIONAL)
                else:
                    self._sessions[server].auth = HTTPKerberosAuth(mutual_authentication=OPTIONAL)
            elif self._authentication_scheme == AuthenticateScheme.KERBEROS_PROXY:
                _LOGGER.debug("Using Kerberos Proxy Authentication")
                if self._authentication_params != [None, None]:
                    self._sessions[server].mount('https://', HTTPAdapterWithProxyKerberosAuth(principal=self._authentication_params[0], password=self._authentication_params[1]))
                else:
                    self._sessions[server].mount('https://', HTTPAdapterWithProxyKerberosAuth())
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

FilesExpand file tree

client.py

Latest commit

History

client.py

File metadata and controls
