FilesExpand file tree

master

/

uploader.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

48 lines (43 loc) · 2.02 KB

master

/

uploader.py

Copy path

Top

File metadata and controls

• Code
• Blame

48 lines (43 loc) · 2.02 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

# coding: utf-8

from __future__ import print_function, unicode_literals

import os

from typing import Any, Union

import msgpack

import requests

import requests.utils

class Fake500(object):

def __init__(self, exception: Exception) -> None:

self.status_code = 500

self.exception = exception

class Uploader(object):

def __init__(self, source_token: str, host: str, timeout: Union[float, tuple[float, float], None]) -> None:

self.source_token = source_token

self.host = host

self.timeout = timeout

# requests would otherwise ask the environment, and on macOS SystemConfiguration, for

# proxy settings on every request. Resolving them once here keeps the flush thread out

# of CoreFoundation, which refuses to run in a child forked from a multithreaded

# process, such as an RQ work horse. The CA bundle variables follow the same rule.

self.proxies = requests.utils.get_environ_proxies(host)

self.verify = os.environ.get('REQUESTS_CA_BUNDLE') or os.environ.get('CURL_CA_BUNDLE') or True

self.headers = {

'Authorization': 'Bearer %s' % source_token,

'Content-Type': 'application/msgpack',

}

self.session = self._new_session()

def __call__(self, frame: list[dict[str, Any]]) -> Union[requests.Response, Fake500]:

data = msgpack.packb(frame, use_bin_type=True)

try:

return self.session.post(self.host, data=data, headers=self.headers, timeout=self.timeout)

except requests.RequestException as e:

return Fake500(e)

def reset(self) -> None:

# A forked child shares the parent's pooled socket, so it needs a session of its own,

# built from the settings resolved above rather than looked up again.

self.session = self._new_session()

def _new_session(self) -> requests.Session:

session = requests.Session()

session.trust_env = False

session.proxies = self.proxies

session.verify = self.verify

return session