How to use the azure-storage-blob.azure.storage.blob._utils.basic_error_map function in azure-storage-blob

To help you get started, we’ve selected a few azure-storage-blob examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
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')
        }
github Azure / azure-sdk-for-python / azure-storage-storage / blob / polling.py View on Github external
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
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
# 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)
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
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)
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
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')
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
):
        # 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
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_service_client.py View on Github external
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)
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
: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
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
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')
        }
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
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', [])
        }

azure-storage-blob

Microsoft Azure Blob Storage Client Library for Python

MIT
Latest version published 1 month ago

Package Health Score

93 / 100
Full package analysis