Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

from __future__ import absolute_import 

from base64 import b64encode 

 

from ..packages.six import b, integer_types 

from ..exceptions import UnrewindableBodyError 

 

ACCEPT_ENCODING = 'gzip,deflate' 

_FAILEDTELL = object() 

 

 

def make_headers(keep_alive=None, accept_encoding=None, user_agent=None, 

basic_auth=None, proxy_basic_auth=None, disable_cache=None): 

""" 

Shortcuts for generating request headers. 

 

:param keep_alive: 

If ``True``, adds 'connection: keep-alive' header. 

 

:param accept_encoding: 

Can be a boolean, list, or string. 

``True`` translates to 'gzip,deflate'. 

List will get joined by comma. 

String will be used as provided. 

 

:param user_agent: 

String representing the user-agent you want, such as 

"python-urllib3/0.6" 

 

:param basic_auth: 

Colon-separated username:password string for 'authorization: basic ...' 

auth header. 

 

:param proxy_basic_auth: 

Colon-separated username:password string for 'proxy-authorization: basic ...' 

auth header. 

 

:param disable_cache: 

If ``True``, adds 'cache-control: no-cache' header. 

 

Example:: 

 

>>> make_headers(keep_alive=True, user_agent="Batman/1.0") 

{'connection': 'keep-alive', 'user-agent': 'Batman/1.0'} 

>>> make_headers(accept_encoding=True) 

{'accept-encoding': 'gzip,deflate'} 

""" 

headers = {} 

if accept_encoding: 

if isinstance(accept_encoding, str): 

pass 

elif isinstance(accept_encoding, list): 

accept_encoding = ','.join(accept_encoding) 

else: 

accept_encoding = ACCEPT_ENCODING 

headers['accept-encoding'] = accept_encoding 

 

if user_agent: 

headers['user-agent'] = user_agent 

 

if keep_alive: 

headers['connection'] = 'keep-alive' 

 

if basic_auth: 

headers['authorization'] = 'Basic ' + \ 

b64encode(b(basic_auth)).decode('utf-8') 

 

if proxy_basic_auth: 

headers['proxy-authorization'] = 'Basic ' + \ 

b64encode(b(proxy_basic_auth)).decode('utf-8') 

 

if disable_cache: 

headers['cache-control'] = 'no-cache' 

 

return headers 

 

 

def set_file_position(body, pos): 

""" 

If a position is provided, move file to that point. 

Otherwise, we'll attempt to record a position for future use. 

""" 

if pos is not None: 

rewind_body(body, pos) 

elif getattr(body, 'tell', None) is not None: 

try: 

pos = body.tell() 

except (IOError, OSError): 

# This differentiates from None, allowing us to catch 

# a failed `tell()` later when trying to rewind the body. 

pos = _FAILEDTELL 

 

return pos 

 

 

def rewind_body(body, body_pos): 

""" 

Attempt to rewind body to a certain position. 

Primarily used for request redirects and retries. 

 

:param body: 

File-like object that supports seek. 

 

:param int pos: 

Position to seek to in file. 

""" 

body_seek = getattr(body, 'seek', None) 

if body_seek is not None and isinstance(body_pos, integer_types): 

try: 

body_seek(body_pos) 

except (IOError, OSError): 

raise UnrewindableBodyError("An error occurred when rewinding request " 

"body for redirect/retry.") 

elif body_pos is _FAILEDTELL: 

raise UnrewindableBodyError("Unable to record file position for rewinding " 

"request body during a redirect/retry.") 

else: 

raise ValueError("body_pos must be of type integer, " 

"instead it was %s." % type(body_pos))