How to use the azure-storage-blob.azure.storage.blob._utils.get_access_conditions 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_container_properties(self, lease=None, timeout=None, **kwargs):
        # type: (Optional[Union[Lease, str]], Optional[int], **Any) -> ContainerProperties
        """
        Returns all user-defined metadata and system properties for the specified
        container. The data returned does not include the container's list of blobs.

        :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 / blob_client.py View on Github external
if isinstance(data, six.text_type):
            data = data.encode(encoding)
        if not length:
            try:
                length = get_length(data)
            except TypeError:
                raise ValueError("Please specifiy content length.")
        if length == 0:
            return {}
        append_conditions = None
        if maxsize_condition or appendpos_condition:
            append_conditions = AppendPositionAccessConditions(
                max_size=maxsize_condition,
                append_condition=appendpos_condition
            )
        access_conditions = get_access_conditions(lease)
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        try:
            return self._client.append_blob.append_block(
                data[:length],
                content_length=length,
                timeout=None,
                transactional_content_md5=None,
                lease_access_conditions=access_conditions,
                append_position_access_conditions=append_conditions,
                modified_access_conditions=mod_conditions,
                cls=return_response_headers,
                **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
source.strip('/')
            )

        headers = kwargs.pop('headers', {})
        headers.update(add_metadata_headers(metadata))
        if source_lease:
            try:
                headers['x-ms-source-lease-id'] = source_lease.id
            except AttributeError:
                headers['x-ms-source-lease-id'] = source_lease
        if premium_page_blob_tier:
            headers['x-ms-access-tier'] = premium_page_blob_tier.value
        if requires_sync:
            headers['x-ms-requires-sync'] = str(requires_sync)

        dest_access_conditions = get_access_conditions(destination_lease)
        dest_mod_conditions = get_modification_conditions(
            destination_if_modified_since,
            destination_if_unmodified_since,
            destination_if_match,
            destination_if_none_match)
        try:
            if incremental_copy:
                start_copy = self._client.page_blob.copy_incremental(
                    source_url,
                    timeout=None,
                    modified_access_conditions=dest_mod_conditions,
                    error_map=basic_error_map(),
                    headers=headers,
                    cls=return_response_headers,
                    **kwargs)
            else:
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
raise ValueError(_ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION)
        block_lookup = BlockLookupList(committed=[], uncommitted=[], latest=[])
        for block in block_list:
            try:
                if block.state.value == 'committed':
                    block_lookup.committed.append(encode_base64(str(block.id)))
                elif block.state.value == 'uncommitted':
                    block_lookup.uncommitted.append(encode_base64(str(block.id)))
                else:
                    block_lookup.latest.append(encode_base64(str(block.id)))
            except AttributeError:
                block_lookup.latest.append(encode_base64(str(block)))
        headers = kwargs.pop('headers', {})
        headers.update(add_metadata_headers(metadata))
        blob_headers = None
        access_conditions = get_access_conditions(lease)
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        if content_settings:
            blob_headers = BlobHTTPHeaders(
                blob_cache_control=content_settings.cache_control,
                blob_content_type=content_settings.content_type,
                blob_content_md5=bytearray(content_settings.content_md5) if content_settings.content_md5 else None,
                blob_content_encoding=content_settings.content_encoding,
                blob_content_language=content_settings.content_language,
                blob_content_disposition=content_settings.content_disposition
            )
        try:
            return self._client.block_blob.commit_block_list(
                block_lookup,
                blob_http_headers=blob_headers,
                lease_access_conditions=access_conditions,
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
if_unmodified_since=None,  # type: Optional[datetime]
            if_match=None,  # type: Optional[str]
            if_none_match=None,  # type: Optional[str]
            timeout=None,  # type: Optional[int]
            **kwargs
        ):
        # type: (...) -> Iterable[bytes]
        """
        :returns: A iterable data generator (stream)
        """
        if self.require_encryption and not self.key_encryption_key:
            raise ValueError("Encryption required but no key was provided.")
        if length is not None and offset is None:
            raise ValueError("Offset value must not be None is length is set.")

        access_conditions = get_access_conditions(lease)
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)

        return StorageStreamDownloader(
            name=self.name,
            container=self.container,
            service=self._client.blob,
            config=self._config.blob_settings,
            offset=offset,
            length=length,
            validate_content=validate_content,
            access_conditions=access_conditions,
            mod_conditions=mod_conditions,
            timeout=timeout,
            require_encryption=self.require_encryption,
            key_encryption_key=self.key_encryption_key,
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
the operation only if the resource does not exist, and fail the
            operation if it does exist.
        :param int timeout:
            The timeout parameter is expressed in seconds.
        :param ~azure.storage.blob.common.PremiumPageBlobTier premium_page_blob_tier:
            Only for Page blobs. A page blob tier value to set the blob to. The tier correlates to the size of the
            blob and number of allowed IOPS. This is only applicable to page blobs on
            premium storage accounts.
        :returns: Blob-updated property dict (Etag and last modified).
        :rtype: dict(str, Any)
        """
        if self.require_encryption or (self.key_encryption_key is not None):
            raise ValueError(_ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION)
        headers = kwargs.pop('headers', {})
        headers.update(add_metadata_headers(metadata))
        access_conditions = get_access_conditions(lease)
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        blob_headers = None
        if content_settings:
            blob_headers = BlobHTTPHeaders(
                blob_cache_control=content_settings.cache_control,
                blob_content_type=content_settings.content_type,
                blob_content_md5=bytearray(content_settings.content_md5) if content_settings.content_md5 else None,
                blob_content_encoding=content_settings.content_encoding,
                blob_content_language=content_settings.content_language,
                blob_content_disposition=content_settings.content_disposition
            )
        try:
            if self.blob_type == BlobType.PageBlob:
                if content_length is None:
                    raise ValueError("A content length must be specified for a Page Blob.")
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
def set_container_metadata(
            self, metadata=None,  # type: Optional[Dict[str, str]]
            lease=None,  # type: Optional[Union[str, Lease]]
            if_modified_since=None,  # type: Optional[datetime]
            timeout=None,  # type: Optional[int]
            **kwargs
    ):
        # 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
timeout=None,  # type: Optional[int]
            **kwargs
        ):
        # type: (...) -> None
        """
        :raises: TypeError when blob client type is not BlockBlob.
        :returns: None
        """
        if self.blob_type != BlobType.BlockBlob:
            raise TypeError("This operation is only available for BlockBlob type blobs.")
        if self.require_encryption or (self.key_encryption_key is not None):
            raise ValueError(_ERROR_UNSUPPORTED_METHOD_FOR_ENCRYPTION)
        block_id = encode_base64(str(block_id))
        if isinstance(data, six.text_type):
            data = data.encode(encoding)
        access_conditions = get_access_conditions(lease)
        if not length:
            try:
                length = len(data)
            except TypeError:
                raise ValueError("Please specify content length.")
        try:
            self._client.block_blob.stage_block(
                block_id,
                length,
                data,
                transactional_content_md5=None,
                timeout=timeout,
                lease_access_conditions=access_conditions,
                validate_content=validate_content,
                **kwargs)
        except StorageErrorException as error:
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_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 str if_match:
            An ETag value, or the wildcard character (*). Specify this header to perform
            the operation only if the resource's ETag matches the value specified.
        :param str if_none_match:
            An ETag value, or the wildcard character (*). Specify this header
            to perform the operation only if the resource's ETag does not match
            the value specified. Specify the wildcard character (*) to perform
            the operation only if the resource does not exist, and fail the
            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 / blob_client.py View on Github external
:param PremiumPageBlobTier premium_page_blob_tier:
            A page blob tier value to set the blob to. The tier correlates to the size of the
            blob and number of allowed IOPS. This is only applicable to page blobs on
            premium storage accounts.
        :param int timeout:
            The timeout parameter is expressed in seconds. This method may make
            multiple calls to the Azure service and the timeout will apply to
            each call individually.
        :param str lease:
            Required if the blob has an active lease.
        :raises: TypeError when blob client type is not PageBlob.
        :returns: None
        """
        if self.blob_type != BlobType.PageBlob:
            raise TypeError("This operation is only available for PageBlob type blobs.")
        access_conditions = get_access_conditions(lease)
        if premium_page_blob_tier is None:
            raise ValueError("A PremiumPageBlobTiermust be specified")
        try:
            self._client.blob.set_tier(
                tier=premium_page_blob_tier,
                timeout=timeout,
                lease_access_conditions=access_conditions,
                **kwargs)
        except StorageErrorException as error:
            process_storage_error(error)

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