This repository was archived by the owner on May 22, 2021. It is now read-only.
Repository navigation
Expand file tree
/
Copy pathutils.py
More file actions
90 lines (74 loc) · 2.68 KB
/
Copy pathutils.py
File metadata and controls
90 lines (74 loc) · 2.68 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
# -*- coding: utf-8 -*-
"""A collection of useful utilities."""
import collections
import datetime
import re
from requests import compat
# with thanks to https://code.google.com/p/jquery-localtime/issues/detail?id=4
ISO_8601 = re.compile(
r"^(-?(?:[1-9][0-9]*)?[0-9]{4})-(1[0-2]|0[1-9])-(3[0-1]|0"
r"[1-9]|[1-2][0-9])(T(2[0-3]|[0-1][0-9]):([0-5][0-9]):([0"
r"-5][0-9])(\.[0-9]+)?(Z|[+-](?:2[0-3]|[0-1][0-9]):[0-5]["
r"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.datetime):
return timestamp.isoformat() + "Z"
if isinstance(timestamp, compat.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))
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
:return: path to the file
:rtype: str
"""
pre_opened = False
fd = None
filename = None
if path:
if isinstance(getattr(path, "write", None), collections.Callable):
pre_opened = True
fd = path
filename = getattr(fd, "name", None)
else:
fd = open(path, "wb")
filename = path
else:
header = response.headers["content-disposition"]
i = header.find("filename=") + len("filename=")
filename = header[i:]
fd = open(filename, "wb")
for chunk in response.iter_content(chunk_size=512):
fd.write(chunk)
if not pre_opened:
fd.close()
return filename