Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def list_service_certificates(self, service_name):
'''
Lists all of the service certificates associated with the specified
hosted service.
service_name: Name of the hosted service.
'''
_validate_not_none('service_name', service_name)
return self._perform_get(
'/' + self.subscription_id + '/services/hostedservices/' +
_str(service_name) + '/certificates',
Certificates)
content_md5:
Optional. An MD5 hash of the block content. This hash is used to
verify the integrity of the blob during transport. When this
header is specified, the storage service checks the hash that has
arrived with the one that was sent.
x_ms_lease_id: Required if the blob has an active lease.
'''
_validate_not_none('container_name', container_name)
_validate_not_none('blob_name', blob_name)
_validate_not_none('block', block)
_validate_not_none('blockid', blockid)
request = HTTPRequest()
request.method = 'PUT'
request.host = self._get_host()
request.path = '/' + \
_str(container_name) + '/' + _str(blob_name) + '?comp=block'
request.headers = [
('Content-MD5', _str_or_none(content_md5)),
('x-ms-lease-id', _str_or_none(x_ms_lease_id))
]
request.query = [('blockid', _encode_base64(_str_or_none(blockid)))]
request.body = _get_request_body_bytes_only('block', block)
request.path, request.query = _update_request_uri_query_local_storage(
request, self.use_local_storage)
request.headers = _update_storage_blob_header(
request, self.account_name, self.account_key)
self._perform_request(request)
def get_blob_properties(self, container_name, blob_name,
x_ms_lease_id=None):
'''
Returns all user-defined metadata, standard HTTP properties, and
system properties for the blob.
container_name: Name of existing container.
blob_name: Name of existing blob.
x_ms_lease_id: Required if the blob has an active lease.
'''
_validate_not_none('container_name', container_name)
_validate_not_none('blob_name', blob_name)
request = HTTPRequest()
request.method = 'HEAD'
request.host = self._get_host()
request.path = '/' + _str(container_name) + '/' + _str(blob_name) + ''
request.headers = [('x-ms-lease-id', _str_or_none(x_ms_lease_id))]
request.path, request.query = _update_request_uri_query_local_storage(
request, self.use_local_storage)
request.headers = _update_storage_blob_header(
request, self.account_name, self.account_key)
response = self._perform_request(request)
return _parse_response_for_dict(response)
def extended_properties_dict_to_xml_fragment(extended_properties):
xml = ''
if extended_properties is not None and len(extended_properties) > 0:
xml += ''
for key, val in extended_properties.items():
xml += ''.join(['',
'',
_str(key),
'',
'',
_str(val),
'',
''])
xml += ''
return xml
blob_name: Name of destination blob.
x_ms_copy_id:
Copy identifier provided in the x-ms-copy-id of the original
copy_blob operation.
x_ms_lease_id:
Required if the destination blob has an active infinite lease.
'''
_validate_not_none('container_name', container_name)
_validate_not_none('blob_name', blob_name)
_validate_not_none('x_ms_copy_id', x_ms_copy_id)
request = HTTPRequest()
request.method = 'PUT'
request.host = self._get_host()
request.path = '/' + _str(container_name) + '/' + \
_str(blob_name) + '?comp=copy©id=' + \
_str(x_ms_copy_id)
request.headers = [
('x-ms-lease-id', _str_or_none(x_ms_lease_id)),
('x-ms-copy-action', 'abort'),
]
request.path, request.query = _update_request_uri_query_local_storage(
request, self.use_local_storage)
request.headers = _update_storage_blob_header(
request, self.account_name, self.account_key)
self._perform_request(request)
topic_name: Name of the topic.
subscription_name: Name of the subscription.
rule_name:
Name of the rule to delete. DEFAULT_RULE_NAME=$Default.
Use DEFAULT_RULE_NAME to delete default rule for the subscription.
fail_not_exist:
Specify whether throw exception when rule doesn't exist.
'''
_validate_not_none('topic_name', topic_name)
_validate_not_none('subscription_name', subscription_name)
_validate_not_none('rule_name', rule_name)
request = HTTPRequest()
request.method = 'DELETE'
request.host = self._get_host()
request.path = '/' + _str(topic_name) + '/subscriptions/' + \
_str(subscription_name) + \
'/rules/' + _str(rule_name) + ''
request.path, request.query = _update_request_uri_query(request)
request.headers = self._update_service_bus_header(request)
if not fail_not_exist:
try:
self._perform_request(request)
return True
except WindowsAzureError as ex:
_dont_fail_not_exist(ex)
return False
else:
self._perform_request(request)
return True
'''
Sets user-defined metadata for the specified blob as one or more
name-value pairs.
container_name: Name of existing container.
blob_name: Name of existing blob.
x_ms_meta_name_values: Dict containing name and value pairs.
x_ms_lease_id: Required if the blob has an active lease.
'''
_validate_not_none('container_name', container_name)
_validate_not_none('blob_name', blob_name)
request = HTTPRequest()
request.method = 'PUT'
request.host = self._get_host()
request.path = '/' + \
_str(container_name) + '/' + _str(blob_name) + '?comp=metadata'
request.headers = [
('x-ms-meta-name-values', x_ms_meta_name_values),
('x-ms-lease-id', _str_or_none(x_ms_lease_id))
]
request.path, request.query = _update_request_uri_query_local_storage(
request, self.use_local_storage)
request.headers = _update_storage_blob_header(
request, self.account_name, self.account_key)
self._perform_request(request)
if_match:
Optional. Specifies the condition for which the delete should be
performed. To force an unconditional delete, set to the wildcard
character (*).
'''
_validate_not_none('table_name', table_name)
_validate_not_none('partition_key', partition_key)
_validate_not_none('row_key', row_key)
_validate_not_none('content_type', content_type)
_validate_not_none('if_match', if_match)
request = HTTPRequest()
request.method = 'DELETE'
request.host = self._get_host()
request.path = '/' + \
_str(table_name) + '(PartitionKey=\'' + \
_str(partition_key) + '\',RowKey=\'' + _str(row_key) + '\')'
request.headers = [
('Content-Type', _str_or_none(content_type)),
('If-Match', _str_or_none(if_match))
]
request.path, request.query = _update_request_uri_query_local_storage(
request, self.use_local_storage)
request.headers = _update_storage_table_header(request)
self._perform_request(request)
def extended_properties_dict_to_xml_fragment(extended_properties):
xml = ''
if extended_properties is not None and len(extended_properties) > 0:
xml += ''
for key, val in extended_properties.items():
xml += ''.join(['',
'',
_str(key),
'',
'',
_str(val),
'',
''])
xml += ''
return xml
'''
Sets user-defined metadata for the specified blob as one or more
name-value pairs.
container_name: Name of existing container.
blob_name: Name of existing blob.
x_ms_meta_name_values: Dict containing name and value pairs.
x_ms_lease_id: Required if the blob has an active lease.
'''
_validate_not_none('container_name', container_name)
_validate_not_none('blob_name', blob_name)
request = HTTPRequest()
request.method = 'PUT'
request.host = self._get_host()
request.path = '/' + \
_str(container_name) + '/' + _str(blob_name) + '?comp=metadata'
request.headers = [
('x-ms-meta-name-values', x_ms_meta_name_values),
('x-ms-lease-id', _str_or_none(x_ms_lease_id))
]
request.path, request.query = _update_request_uri_query_local_storage(
request, self.use_local_storage)
request.headers = _update_storage_blob_header(
request, self.account_name, self.account_key)
self._perform_request(request)