Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_account_information(self, timeout=None, **kwargs):
# type: (Optional[int]) -> Dict[str, str]
"""
:returns: A dict of account information (SKU and account type).
"""
try:
response = self._client.container.get_account_info(
cls=return_response_headers,
timeout=timeout,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
return {
'SKU': response.get('x-ms-sku-name'),
'AccountType': response.get('x-ms-account-kind')
}
def _update_status(self):
try:
self.blob = self._client._client.blob.get_properties(
cls=deserialize_blob_properties,
error_map=basic_error_map(),
**self.kwargs)
except StorageErrorException as error:
process_storage_error(error)
self._status = self.blob.copy.status
self.etag = self.blob.etag
self.last_modified = self.blob.last_modified
# type: (...) -> Dict[str, Union[str, datetime]]
"""
:returns: Container-updated property dict (Etag and last modified).
"""
headers = kwargs.pop('headers', {})
headers.update(add_metadata_headers(metadata))
access_conditions = get_access_conditions(lease)
mod_conditions = get_modification_conditions(if_modified_since)
try:
return self._client.container.set_metadata(
timeout=timeout,
lease_access_conditions=access_conditions,
modified_access_conditions=mod_conditions,
cls=return_response_headers,
headers=headers,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
operation if it does exist.
:param int timeout:
The timeout parameter is expressed in seconds.
:rtype: None
"""
access_conditions = get_access_conditions(lease)
mod_conditions = get_modification_conditions(
if_modified_since, if_unmodified_since, if_match, if_none_match)
try:
self._client.blob.delete(
timeout=timeout,
snapshot=self.snapshot,
delete_snapshots=delete_snapshots,
lease_access_conditions=access_conditions,
modified_access_conditions=mod_conditions,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
Specify this header to perform the operation only if
the resource has not been modified since the specified date/time.
:param int timeout:
The timeout parameter is expressed in seconds.
:return: Approximate time remaining in the lease period, in seconds.
:rtype: int
"""
mod_conditions = get_modification_conditions(
if_modified_since, if_unmodified_since)
try:
response = self._client.container.break_lease(
timeout=timeout,
break_period=lease_break_period,
modified_access_conditions=mod_conditions,
cls=return_response_headers,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
return response.get('x-ms-lease-time')
):
# type: (...) -> BlobProperties
"""
:returns: BlobProperties
"""
access_conditions = get_access_conditions(lease)
mod_conditions = get_modification_conditions(
if_modified_since, if_unmodified_since, if_match, if_none_match)
try:
blob_props = self._client.blob.get_properties(
timeout=timeout,
snapshot=self.snapshot,
lease_access_conditions=access_conditions,
modified_access_conditions=mod_conditions,
cls=deserialize_blob_properties,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
blob_props.name = self.name
blob_props.container = self.container
return blob_props
An opaque continuation token. This value can be retrieved from the
next_marker field of a previous generator object. If specified,
this generator will begin returning results from this point.
:param int timeout:
The timeout parameter is expressed in seconds.
:returns: An iterable (auto-paging) of ContainerProperties.
:rtype: ~azure.core.blob.models.ContainerPropertiesPaged
"""
include = 'metadata' if include_metadata else None
results_per_page = kwargs.pop('results_per_page', None)
command = functools.partial(
self._client.service.list_containers_segment,
prefix=starts_with,
include=include,
timeout=timeout,
error_map=basic_error_map(),
**kwargs)
return ContainerPropertiesPaged(
command, prefix=starts_with, results_per_page=results_per_page, marker=marker)
:param ~azure.storage.blob.lease.Lease lease:
If specified, get_container_properties only succeeds if the
container's lease is active and matches this ID.
:param int timeout:
The timeout parameter is expressed in seconds.
:return: properties for the specified container within a container object.
:rtype: ~azure.storage.blob.models.ContainerProperties
"""
access_conditions = get_access_conditions(lease)
try:
response = self._client.container.get_properties(
timeout=timeout,
lease_access_conditions=access_conditions,
cls=deserialize_container_properties,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
response.name = self.name
return response
for key, value in signed_identifiers.items():
identifiers.append(SignedIdentifier(id=key, access_policy=value))
signed_identifiers = identifiers
mod_conditions = get_modification_conditions(
if_modified_since, if_unmodified_since)
access_conditions = get_access_conditions(lease)
try:
response = self._client.container.set_access_policy(
container_acl=signed_identifiers or None,
timeout=timeout,
access=public_access,
lease_access_conditions=access_conditions,
modified_access_conditions=mod_conditions,
cls=return_response_headers,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
return {
'ETag': response.get('ETag'),
'Last-Modified': response.get('Last-Modified')
}
def get_container_acl(self, lease=None, timeout=None, **kwargs):
# type: (Optional[Union[Lease, str]], Optional[int]) -> Dict[str, str]
"""
:returns: Access policy information in a dict.
"""
access_conditions = get_access_conditions(lease)
try:
response = self._client.container.get_access_policy(
timeout=timeout,
lease_access_conditions=access_conditions,
cls=return_response_and_deserialized,
error_map=basic_error_map(),
**kwargs)
except StorageErrorException as error:
process_storage_error(error)
return {
'public_access': response.get('header').get('x-ms-blob-public-access'),
'signed_identifiers': response.get('deserialized', [])
}