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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
from xml.sax.saxutils import escape as xml_escape
from datetime import date
try:
from xml.etree import cElementTree as ETree
except ImportError:
from xml.etree import ElementTree as ETree
from ..common._common_conversion import (
_encode_base64,
_str,
)
from ..common._serialization import (
_to_utc_datetime,
)
from ..common._error import (
_validate_not_none,
_ERROR_START_END_NEEDED_FOR_MD5,
_ERROR_RANGE_TOO_LARGE_FOR_MD5,
)
from ._error import (
_ERROR_PAGE_BLOB_START_ALIGNMENT,
_ERROR_PAGE_BLOB_END_ALIGNMENT,
_ERROR_INVALID_BLOCK_ID,
)
from io import BytesIO
def _get_path(container_name=None, blob_name=None):
'''
Creates the path to access a blob resource.
container_name:
Name of container.
blob_name:
The path to the blob.
'''
if container_name and blob_name:
return '/{0}/{1}'.format(
_str(container_name),
_str(blob_name))
elif container_name:
return '/{0}'.format(_str(container_name))
else:
return '/'
def _validate_and_format_range_headers(request, start_range, end_range, start_range_required=True,
end_range_required=True, check_content_md5=False, align_to_page=False,
range_header_name='x-ms-range'):
# If end range is provided, start range must be provided
if start_range_required or end_range is not None:
_validate_not_none('start_range', start_range)
if end_range_required:
_validate_not_none('end_range', end_range)
# Page ranges must be 512 aligned
if align_to_page:
if start_range is not None and start_range % 512 != 0:
raise ValueError(_ERROR_PAGE_BLOB_START_ALIGNMENT)
if end_range is not None and end_range % 512 != 511:
raise ValueError(_ERROR_PAGE_BLOB_END_ALIGNMENT)
# Format based on whether end_range is present
request.headers = request.headers or {}
if end_range is not None:
request.headers[range_header_name] = 'bytes={0}-{1}'.format(start_range, end_range)
elif start_range is not None:
request.headers[range_header_name] = "bytes={0}-".format(start_range)
# Content MD5 can only be provided for a complete range less than 4MB in size
if check_content_md5:
if start_range is None or end_range is None:
raise ValueError(_ERROR_START_END_NEEDED_FOR_MD5)
if end_range - start_range > 4 * 1024 * 1024:
raise ValueError(_ERROR_RANGE_TOO_LARGE_FOR_MD5)
request.headers['x-ms-range-get-content-md5'] = 'true'
def _convert_block_list_to_xml(block_id_list):
'''
<?xml version="1.0" encoding="utf-8"?>
<BlockList>
<Committed>first-base64-encoded-block-id</Committed>
<Uncommitted>second-base64-encoded-block-id</Uncommitted>
<Latest>third-base64-encoded-block-id</Latest>
</BlockList>
Convert a block list to xml to send.
block_id_list:
A list of BlobBlock containing the block ids and block state that are used in put_block_list.
Only get block from latest blocks.
'''
if block_id_list is None:
return ''
block_list_element = ETree.Element('BlockList')
# Enabled
for block in block_id_list:
if block.id is None:
raise ValueError(_ERROR_INVALID_BLOCK_ID)
id = xml_escape(_str(format(_encode_base64(block.id))))
ETree.SubElement(block_list_element, block.state).text = id
# Add xml declaration and serialize
try:
stream = BytesIO()
ETree.ElementTree(block_list_element).write(stream, xml_declaration=True, encoding='utf-8', method='xml')
except:
raise
finally:
output = stream.getvalue()
stream.close()
# return xml value
return output
def _convert_delegation_key_info_to_xml(start_time, expiry_time):
"""
<?xml version="1.0" encoding="utf-8"?>
<KeyInfo>
<Start> String, formatted ISO Date </Start>
<Expiry> String, formatted ISO Date </Expiry>
</KeyInfo>
Convert key info to xml to send.
"""
if start_time is None or expiry_time is None:
raise ValueError("delegation key start/end times are required")
key_info_element = ETree.Element('KeyInfo')
ETree.SubElement(key_info_element, 'Start').text = \
_to_utc_datetime(start_time) if isinstance(start_time, date) else start_time
ETree.SubElement(key_info_element, 'Expiry').text = \
_to_utc_datetime(expiry_time) if isinstance(expiry_time, date) else expiry_time
# Add xml declaration and serialize
try:
stream = BytesIO()
ETree.ElementTree(key_info_element).write(stream, xml_declaration=True, encoding='utf-8', method='xml')
finally:
output = stream.getvalue()
stream.close()
# return xml value
return output
|
