data-link-python/nasdaqdatalink/model/authorized_session.py at e87eee2d202242bea94cf09a60658388d070bf31 · Nasdaq/data-link-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import nasdaqdatalink
from nasdaqdatalink.api_config import ApiConfig
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import requests
import urllib


def get_retries(api_config=nasdaqdatalink.ApiConfig):
    retries = None
    if not api_config.use_retries:
        return Retry(total=0)

    Retry.BACKOFF_MAX = api_config.max_wait_between_retries
    retries = Retry(total=api_config.number_of_retries,
                    connect=api_config.number_of_retries,
                    read=api_config.number_of_retries,
                    status_forcelist=api_config.retry_status_codes,
                    backoff_factor=api_config.retry_backoff_factor,
                    raise_on_status=False)
    return retries


class AuthorizedSession:
    def __init__(self, api_config=ApiConfig) -> None:
        super(AuthorizedSession, self).__init__()
        if not isinstance(api_config, ApiConfig):
            api_config = ApiConfig
        self._api_config = api_config
        self._auth_session = requests.Session()
        retries = get_retries(self._api_config)
        adapter = HTTPAdapter(max_retries=retries)
        self._auth_session.mount(api_config.api_protocol, adapter)

        proxies = urllib.request.getproxies()
        if proxies is not None:
            self._auth_session.proxies.update(proxies)

    def get(self, dataset, **kwargs):
        nasdaqdatalink.get(dataset, session=self._auth_session,
                           api_config=self._api_config, **kwargs)

    def bulkdownload(self, database, **kwargs):
        nasdaqdatalink.bulkdownload(database, session=self._auth_session,
                                    api_config=self._api_config, **kwargs)

    def export_table(self, datatable_code, **kwargs):
        nasdaqdatalink.export_table(datatable_code, session=self._auth_session,
                                    api_config=self._api_config, **kwargs)

    def get_table(self, datatable_code, **options):
        nasdaqdatalink.get_table(datatable_code, session=self._auth_session,
                                 api_config=self._api_config, **options)

    def get_point_in_time(self, datatable_code, **options):
        nasdaqdatalink.get_point_in_time(datatable_code, session=self._auth_session,
                                         api_config=self._api_config, **options)