Repository navigation
Expand file tree
/
Copy pathutils.py
More file actions
94 lines (72 loc) · 2.82 KB
/
Copy pathutils.py
File metadata and controls
94 lines (72 loc) · 2.82 KB
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
# -*- coding: utf-8 -*-
"""A collection of useful utilities."""
from collections import Callable
from datetime import datetime, timedelta, tzinfo
from requests.compat import basestring
import re
# with thanks to https://code.google.com/p/jquery-localtime/issues/detail?id=4
ISO_8601 = re.compile("^(-?(?:[1-9][0-9]*)?[0-9]{4})-(1[0-2]|0[1-9])-(3[0-1]|0"
"[1-9]|[1-2][0-9])(T(2[0-3]|[0-1][0-9]):([0-5][0-9]):([0"
"-5][0-9])(\.[0-9]+)?(Z|[+-](?:2[0-3]|[0-1][0-9]):[0-5]["
"0-9])?)?$")
def timestamp_parameter(timestamp, allow_none=True):
"""Function to check the conformance of timestamps passed by users.
This will check that a string is a valid format and allow users to pass a
datetime object which we will then convert to a proper ISO8601 date-time
string.
:param timestamp: string to be validated or datetime object to be
converted.
:param bool allow_none: whether or not to allow timestamp to be None.
Default: ``True``
:returns: valid ISO8601 string
:rtype: str
:raises: ValueError
"""
if timestamp is None:
if allow_none:
return None
raise ValueError("Timestamp value cannot be None")
if isinstance(timestamp, datetime):
return timestamp.isoformat() + 'Z'
if isinstance(timestamp, basestring):
if not ISO_8601.match(timestamp):
raise ValueError(("Invalid timestamp: %s is not a valid ISO-8601"
" formatted date") % timestamp)
return timestamp
raise ValueError("Cannot accept type %s for timestamp" % type(timestamp))
class UTC(tzinfo):
"""Yet another UTC reimplementation, to avoid a dependency on pytz or
dateutil."""
ZERO = timedelta(0)
def __repr__(self):
return 'UTC()'
def dst(self, dt):
return self.ZERO
def tzname(self, dt):
return 'UTC'
def utcoffset(self, dt):
return self.ZERO
def stream_response_to_file(response, path=None):
"""Stream a response body to the specified file.
Either use the ``path`` provided or use the name provided in the
``Content-Disposition`` header.
:param response: A Response object from requests
:type response: requests.models.Response
:param str path: The full path and file name used to save the response
"""
pre_opened = False
fd = None
if path:
if isinstance(getattr(path, 'write', None), Callable):
pre_opened = True
fd = path
else:
fd = open(path, 'wb')
else:
header = response.headers['content-disposition']
i = header.find('filename=') + len('filename=')
fd = open(header[i:], 'wb')
for chunk in response.iter_content(chunk_size=512):
fd.write(chunk)
if not pre_opened:
fd.close()