GeezerVison

Also, please refer to the API documentation to get a full understanding of how the API works and the supported arguments. All required arguments are annotated with red asterisks " * " and argument examples are set.

Behind a corporate proxy or TLS inspection, point the clients at your CA bundle and proxy:

from massive import RESTClient, WebSocketClient

client = RESTClient(

proxy="http://user:[email protected]:8080", # REST doesn't read HTTPS_PROXY

ca_certs="/path/to/corporate-ca.pem", # or set SSL_CERT_FILE / REQUESTS_CA_BUNDLE

)

ws_client = WebSocketClient(ca_certs="/path/to/corporate-ca.pem")

ws_client.run(handle_msg, proxy="http://user:[email protected]:8080") # defaults to the system proxy

Both clients verify TLS with `ca_certs` if given, then the `SSL_CERT_FILE` or `REQUESTS_CA_BUNDLE` environment variable, then the bundled certifi certificates.

`RESTClient` keeps up to `maxsize=10` connections open for reuse across threads. Other urllib3 pool options can be passed with `connection_pool_kw`.

Sometimes you may find it useful to see the actual request and response details while working with the API. The `RESTClient` allows for this through its `trace=True` option.

import os

import certifi

def ca_bundle_path() -> str:

"""

return (

os.environ.get("SSL_CERT_FILE")

or os.environ.get("REQUESTS_CA_BUNDLE")

or certifi.where()

)

ContractsClient,

)

from .vX import VXClient

from typing import Optional, Any

from typing import Optional, Any, Dict

import os

BASE = "https://api.massive.com"

verbose: bool = False,

trace: bool = False,

custom_json: Optional[Any] = None,

maxsize: int = 10,

ca_certs: Optional[str] = None,

proxy: Optional[str] = None,

connection_pool_kw: Optional[Dict[str, Any]] = None,

):

"""

super().__init__(

api_key=api_key,

connect_timeout=connect_timeout,

verbose=verbose,

trace=trace,

custom_json=custom_json,

maxsize=maxsize,

ca_certs=ca_certs,

proxy=proxy,

connection_pool_kw=connection_pool_kw,

)

self.vx = VXClient(

api_key=api_key,

verbose=verbose,

trace=trace,

custom_json=custom_json,

maxsize=maxsize,

ca_certs=ca_certs,

proxy=proxy,

connection_pool_kw=connection_pool_kw,

)

import certifi

import json

import urllib3

import inspect

from datetime import datetime

from importlib.metadata import version, PackageNotFoundError

from .models.request import RequestOptionBuilder

from ..certs import ca_bundle_path

from ..logging import get_logger

import logging

from urllib.parse import urlencode, urlparse

verbose: bool,

trace: bool,

custom_json: Optional[Any] = None,

maxsize: int = 10,

ca_certs: Optional[str] = None,

proxy: Optional[str] = None,

connection_pool_kw: Optional[Dict[str, Any]] = None,

):

if api_key is None:

raise AuthError(

# https://urllib3.readthedocs.io/en/stable/reference/urllib3.poolmanager.html

# https://urllib3.readthedocs.io/en/stable/reference/urllib3.connectionpool.html#urllib3.HTTPConnectionPool

self.client = urllib3.PoolManager(

num_pools=num_pools,

headers=self.headers, # default headers sent with each request.

ca_certs=certifi.where(),

cert_reqs="CERT_REQUIRED",

retries=retry_strategy, # use the customized Retry instance

timeout=self.timeout, # set timeout for each request

)

pool_kw: Dict[str, Any] = {

"num_pools": num_pools,

# connections kept per host; urllib3 defaults to 1, which makes threaded

# clients reconnect (and redo the TLS handshake) for most requests

"maxsize": maxsize,

"headers": self.headers, # default headers sent with each request.

"ca_certs": ca_certs or ca_bundle_path(),

"cert_reqs": "CERT_REQUIRED",

"retries": retry_strategy, # use the customized Retry instance

"timeout": self.timeout, # set timeout for each request

}

# any other urllib3 pool options, which take precedence over the defaults

pool_kw.update(connection_pool_kw or {})

self.client: urllib3.PoolManager

if proxy:

# https://urllib3.readthedocs.io/en/stable/advanced-usage.html#proxies

proxy_auth = urllib3.util.parse_url(proxy).auth

if proxy_auth:

pool_kw.setdefault(

"proxy_headers", urllib3.make_headers(proxy_basic_auth=proxy_auth)

)

self.client = urllib3.ProxyManager(proxy, **pool_kw)

else:

self.client = urllib3.PoolManager(**pool_kw)

if verbose:

logger.setLevel(logging.DEBUG)

import asyncio

import ssl

import time

import certifi

from .models import *

from websockets.asyncio.client import connect, ClientConnection

from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError

from ..certs import ca_bundle_path

from ..logging import get_logger

import logging

from ..exceptions import AuthError

max_reconnects: Optional[int] = 5,

secure: bool = True,

custom_json: Optional[Any] = None,

ca_certs: Optional[str] = None,

**kwargs,

):

"""

if api_key is None:

self.schedule_resub = True

self._server_errors: Counter = Counter()

self._server_errors_since = 0.0

self.ca_certs = ca_certs

if custom_json:

self.json = custom_json

else:

reconnects = 0

logger.debug("connect: %s", self.url)

# darwin needs some extra <3

ssl_context = None

if self.url.startswith("wss://"):

ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

ssl_context.load_verify_locations(certifi.where())

ssl_context = self._ssl_context()

last_exc = None

async for s in connect(

):

raise last_exc

def _ssl_context(self) -> Optional[ssl.SSLContext]:

# darwin needs some extra <3

if not self.url.startswith("wss://"):

return None

context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

context.load_verify_locations(self.ca_certs or ca_bundle_path())

return context

def run(

self,

handle_msg: Union[

import os

import unittest

from unittest import mock

import certifi

import urllib3

from massive import RESTClient

class ConnectionOptionsTest(unittest.TestCase):

def client(self, env=None, **kwargs):

with mock.patch.dict(os.environ, env or {}, clear=True):

return RESTClient(api_key="key", **kwargs)

def test_ca_bundle_defaults_to_certifi(self):

c = self.client()

self.assertEqual(c.client.connection_pool_kw["ca_certs"], certifi.where())

def test_ca_bundle_from_env(self):

c = self.client({"REQUESTS_CA_BUNDLE": "/requests-ca.pem"})

self.assertEqual(c.client.connection_pool_kw["ca_certs"], "/requests-ca.pem")

c = self.client(

{"SSL_CERT_FILE": "/ssl-ca.pem", "REQUESTS_CA_BUNDLE": "/requests-ca.pem"}

)

self.assertEqual(c.client.connection_pool_kw["ca_certs"], "/ssl-ca.pem")

def test_ca_certs_argument_wins_over_env(self):

c = self.client({"SSL_CERT_FILE": "/ssl-ca.pem"}, ca_certs="/explicit-ca.pem")

self.assertEqual(c.client.connection_pool_kw["ca_certs"], "/explicit-ca.pem")

def test_ca_certs_applies_to_vx(self):

c = self.client(ca_certs="/explicit-ca.pem")

self.assertEqual(c.vx.client.connection_pool_kw["ca_certs"], "/explicit-ca.pem")

def test_maxsize(self):

self.assertEqual(self.client().client.connection_pool_kw["maxsize"], 10)

c = self.client(maxsize=32)

self.assertEqual(c.client.connection_pool_kw["maxsize"], 32)

def test_connection_pool_kw_overrides_defaults(self):

timeout = urllib3.Timeout(connect=1, read=2)

c = self.client(connection_pool_kw={"block": True, "timeout": timeout})

self.assertTrue(c.client.connection_pool_kw["block"])

self.assertIs(c.client.connection_pool_kw["timeout"], timeout)

# defaults that weren't overridden are kept

self.assertEqual(c.client.connection_pool_kw["cert_reqs"], "CERT_REQUIRED")

def test_no_proxy_by_default(self):

c = self.client({"HTTPS_PROXY": "http://proxy.example:8080"})

self.assertNotIsInstance(c.client, urllib3.ProxyManager)

def test_proxy(self):

c = self.client(proxy="http://user:[email protected]:8080")

self.assertIsInstance(c.client, urllib3.ProxyManager)

self.assertEqual(c.client.proxy.host, "proxy.example")

self.assertEqual(

c.client.proxy_headers["proxy-authorization"],

urllib3.make_headers(proxy_basic_auth="user:secret")["proxy-authorization"],

)

self.assertEqual(c.client.connection_pool_kw["maxsize"], 10)

if __name__ == "__main__":

unittest.main()

import os

import unittest

from unittest import mock

import certifi

from massive import WebSocketClient

class WebSocketSSLTest(unittest.TestCase):

def ca_path(self, env=None, **kwargs):

with mock.patch.dict(os.environ, env or {}, clear=True):

c = WebSocketClient(api_key="key", **kwargs)

with mock.patch("massive.websocket.ssl.SSLContext") as context:

c._ssl_context()

return context.return_value.load_verify_locations.call_args.args[0]

def test_defaults_to_certifi(self):

self.assertEqual(self.ca_path(), certifi.where())

def test_env(self):

self.assertEqual(self.ca_path({"SSL_CERT_FILE": "/ssl-ca.pem"}), "/ssl-ca.pem")

self.assertEqual(

self.ca_path({"REQUESTS_CA_BUNDLE": "/requests-ca.pem"}), "/requests-ca.pem"

)

def test_ca_certs_argument_wins_over_env(self):

self.assertEqual(

self.ca_path({"SSL_CERT_FILE": "/ssl-ca.pem"}, ca_certs="/explicit-ca.pem"),

"/explicit-ca.pem",

)

def test_no_context_without_tls(self):

c = WebSocketClient(api_key="key", feed="localhost:8765", secure=False)

self.assertIsNone(c._ssl_context())

if __name__ == "__main__":

unittest.main()