How to use the azure-storage-blob.azure.storage.blob._utils.get_modification_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 / lease.py View on Github external
def release(
            self, if_modified_since=None,  # type: Optional[datetime]
            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: (...) -> None
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        response = self._client.release_lease(
            lease_id=self.id,
            timeout=timeout,
            modified_access_conditions=mod_conditions,
            cls=return_response_headers,
            **kwargs)
        self.etag = response.get('ETag')  # type: str
        self.id = response.get('x-ms-lease-id')  # type: str
        self.last_modified = response.get('Last-Modified')   # type: datetime 
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
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 / container_client.py View on Github external
If timezone is included, any non-UTC datetimes will be converted to UTC.
            If a date is passed in without timezone info, it is assumed to be UTC. 
            Specify this header to perform the operation only
            if the resource has been modified since the specified time.
        :param datetime if_unmodified_since:
            A DateTime value. Azure expects the date value passed in to be UTC.
            If timezone is included, any non-UTC datetimes will be converted to UTC.
            If a date is passed in without timezone info, it is assumed to be UTC.
            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.
        :returns: A Lease object, that can be run in a context manager.
        :rtype: ~azure.storage.blob.lease.Lease
        """
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        try:
            response = self._client.container.acquire_lease(
                timeout=timeout,
                duration=lease_duration,
                proposed_lease_id=lease_id,
                modified_access_conditions=mod_conditions,
                cls=return_response_headers,
                error_map=basic_error_map(),
                **kwargs)
        except StorageErrorException as error:
            process_storage_error(error)
        return Lease(self._client.container, **response)
github Azure / azure-sdk-for-python / azure-storage-storage / blob / blob_client.py View on Github external
try:
                length = get_length(page)
            except TypeError:
                raise ValueError("Please specifiy content length.")
        if start_range is None or start_range % 512 != 0:
            raise ValueError("start_range must be an integer that aligns with 512 page size")
        if end_range is None or end_range % 512 != 511:
            raise ValueError("end_range must be an integer that aligns with 512 page size")
        content_range = 'bytes={0}-{1}'.format(start_range, end_range)
        access_conditions = get_access_conditions(lease)
        seq_conditions = get_sequence_conditions(
            if_sequence_number_lte=if_sequence_number_lte,
            if_sequence_number_lt=if_sequence_number_lt,
            if_sequence_number_eq=if_sequence_number_eq
        )
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        try:
            return self._client.page_blob.upload_pages(
                page[:length],
                content_length=length,
                transactional_content_md5=None,
                timeout=timeout,
                range=content_range,
                lease_access_conditions=access_conditions,
                sequence_number_access_conditions=seq_conditions,
                modified_access_conditions=mod_conditions,
                validate_content=validate_content,
                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
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,
            key_resolver_function=self.key_resolver_function,
github Azure / azure-sdk-for-python / azure-storage-storage / blob / container_client.py View on Github external
If timezone is included, any non-UTC datetimes will be converted to UTC.
            If a date is passed in without timezone info, it is assumed to be UTC. 
            Specify this header to perform the operation only
            if the resource has been modified since the specified time.
        :param datetime if_unmodified_since:
            A DateTime value. Azure expects the date value passed in to be UTC.
            If timezone is included, any non-UTC datetimes will be converted to UTC.
            If a date is passed in without timezone info, it is assumed to be UTC.
            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
def acquire_lease(
            self, lease_duration=-1,  # type: int
            lease_id=None,  # type: Optional[str]
            if_modified_since=None,  # type: Optional[datetime]
            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: (...) -> Lease
        """
        :returns: A Lease object.
        """
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        try:
            response = self._client.blob.acquire_lease(
                timeout=timeout,
                duration=lease_duration,
                proposed_lease_id=lease_id,
                modified_access_conditions=mod_conditions,
                cls=return_response_headers,
                **kwargs)
        except StorageErrorException as error:
            process_storage_error(error)
        return Lease(self._client.blob, **response)
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
def get_blob_properties(
            self, lease=None,  # type: Optional[Union[Lease, str]]
            if_modified_since=None,  # type: Optional[datetime]
            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: (...) -> 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 / lease.py View on Github external
def renew(
            self, if_modified_since=None,  # type: Optional[datetime]
            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: (...) -> None
        mod_conditions = get_modification_conditions(
            if_modified_since, if_unmodified_since, if_match, if_none_match)
        response = self._client.renew_lease(
            lease_id=self.id,
            timeout=timeout,
            modified_access_conditions=mod_conditions,
            cls=return_response_headers,
            **kwargs)
        self.etag = response.get('ETag')  # type: str
        self.id = response.get('x-ms-lease-id')  # type: str
        self.last_modified = response.get('Last-Modified')   # type: datetime 

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