# pylint: disable=too-many-lines
# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
import datetime
import sys
from typing import Any, Callable, Dict, IO, Iterator, List, Literal, Optional, TypeVar, Union
from azure.core import PipelineClient
from azure.core.exceptions import (
ClientAuthenticationError,
HttpResponseError,
ResourceExistsError,
ResourceNotFoundError,
ResourceNotModifiedError,
StreamClosedError,
StreamConsumedError,
map_error,
)
from azure.core.pipeline import PipelineResponse
from azure.core.rest import HttpRequest, HttpResponse
from azure.core.tracing.decorator import distributed_trace
from azure.core.utils import case_insensitive_dict
from .. import models as _models
from .._configuration import AzureBlobStorageConfiguration
from .._serialization import Deserializer, Serializer
if sys.version_info >= (3, 9):
from collections.abc import MutableMapping
else:
from typing import MutableMapping # type: ignore
T = TypeVar("T")
ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]
_SERIALIZER = Serializer()
_SERIALIZER.client_side_validation = False
def build_create_request(
url: str,
*,
timeout: Optional[int] = None,
metadata: Optional[Dict[str, str]] = None,
access: Optional[Union[str, _models.PublicAccessType]] = None,
request_id_parameter: Optional[str] = None,
default_encryption_scope: Optional[str] = None,
prevent_encryption_scope_override: Optional[bool] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
if metadata is not None:
_headers["x-ms-meta"] = _SERIALIZER.header("metadata", metadata, "{str}")
if access is not None:
_headers["x-ms-blob-public-access"] = _SERIALIZER.header("access", access, "str")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
if default_encryption_scope is not None:
_headers["x-ms-default-encryption-scope"] = _SERIALIZER.header(
"default_encryption_scope", default_encryption_scope, "str"
)
if prevent_encryption_scope_override is not None:
_headers["x-ms-deny-encryption-scope-override"] = _SERIALIZER.header(
"prevent_encryption_scope_override", prevent_encryption_scope_override, "bool"
)
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_get_properties_request(
url: str,
*,
timeout: Optional[int] = None,
lease_id: Optional[str] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
if lease_id is not None:
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_delete_request(
url: str,
*,
timeout: Optional[int] = None,
lease_id: Optional[str] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
if lease_id is not None:
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="DELETE", url=_url, params=_params, headers=_headers, **kwargs)
def build_set_metadata_request(
url: str,
*,
timeout: Optional[int] = None,
lease_id: Optional[str] = None,
metadata: Optional[Dict[str, str]] = None,
if_modified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["metadata"] = kwargs.pop("comp", _params.pop("comp", "metadata"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
if lease_id is not None:
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
if metadata is not None:
_headers["x-ms-meta"] = _SERIALIZER.header("metadata", metadata, "{str}")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_get_access_policy_request(
url: str,
*,
timeout: Optional[int] = None,
lease_id: Optional[str] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["acl"] = kwargs.pop("comp", _params.pop("comp", "acl"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
if lease_id is not None:
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_set_access_policy_request(
url: str,
*,
timeout: Optional[int] = None,
lease_id: Optional[str] = None,
access: Optional[Union[str, _models.PublicAccessType]] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["acl"] = kwargs.pop("comp", _params.pop("comp", "acl"))
content_type: Optional[str] = kwargs.pop("content_type", _headers.pop("Content-Type", None))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
if lease_id is not None:
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
if access is not None:
_headers["x-ms-blob-public-access"] = _SERIALIZER.header("access", access, "str")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
if content_type is not None:
_headers["Content-Type"] = _SERIALIZER.header("content_type", content_type, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, content=content, **kwargs)
def build_restore_request(
url: str,
*,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
deleted_container_name: Optional[str] = None,
deleted_container_version: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["undelete"] = kwargs.pop("comp", _params.pop("comp", "undelete"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
if deleted_container_name is not None:
_headers["x-ms-deleted-container-name"] = _SERIALIZER.header(
"deleted_container_name", deleted_container_name, "str"
)
if deleted_container_version is not None:
_headers["x-ms-deleted-container-version"] = _SERIALIZER.header(
"deleted_container_version", deleted_container_version, "str"
)
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_rename_request(
url: str,
*,
source_container_name: str,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
source_lease_id: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["rename"] = kwargs.pop("comp", _params.pop("comp", "rename"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["x-ms-source-container-name"] = _SERIALIZER.header("source_container_name", source_container_name, "str")
if source_lease_id is not None:
_headers["x-ms-source-lease-id"] = _SERIALIZER.header("source_lease_id", source_lease_id, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_submit_batch_request(
url: str,
*,
content_length: int,
content: IO[bytes],
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["batch"] = kwargs.pop("comp", _params.pop("comp", "batch"))
multipart_content_type: Optional[str] = kwargs.pop("multipart_content_type", _headers.pop("Content-Type", None))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["Content-Length"] = _SERIALIZER.header("content_length", content_length, "int")
if multipart_content_type is not None:
_headers["Content-Type"] = _SERIALIZER.header("multipart_content_type", multipart_content_type, "str")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="POST", url=_url, params=_params, headers=_headers, content=content, **kwargs)
def build_filter_blobs_request(
url: str,
*,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
where: Optional[str] = None,
marker: Optional[str] = None,
maxresults: Optional[int] = None,
include: Optional[List[Union[str, _models.FilterBlobsIncludeItem]]] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["blobs"] = kwargs.pop("comp", _params.pop("comp", "blobs"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
if where is not None:
_params["where"] = _SERIALIZER.query("where", where, "str")
if marker is not None:
_params["marker"] = _SERIALIZER.query("marker", marker, "str")
if maxresults is not None:
_params["maxresults"] = _SERIALIZER.query("maxresults", maxresults, "int", minimum=1)
if include is not None:
_params["include"] = _SERIALIZER.query("include", include, "[str]", div=",")
# Construct headers
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_acquire_lease_request(
url: str,
*,
timeout: Optional[int] = None,
duration: Optional[int] = None,
proposed_lease_id: Optional[str] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["acquire"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "acquire"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-lease-action"] = _SERIALIZER.header("action", action, "str")
if duration is not None:
_headers["x-ms-lease-duration"] = _SERIALIZER.header("duration", duration, "int")
if proposed_lease_id is not None:
_headers["x-ms-proposed-lease-id"] = _SERIALIZER.header("proposed_lease_id", proposed_lease_id, "str")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_release_lease_request(
url: str,
*,
lease_id: str,
timeout: Optional[int] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["release"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "release"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-lease-action"] = _SERIALIZER.header("action", action, "str")
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_renew_lease_request(
url: str,
*,
lease_id: str,
timeout: Optional[int] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["renew"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "renew"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-lease-action"] = _SERIALIZER.header("action", action, "str")
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_break_lease_request(
url: str,
*,
timeout: Optional[int] = None,
break_period: Optional[int] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["break"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "break"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-lease-action"] = _SERIALIZER.header("action", action, "str")
if break_period is not None:
_headers["x-ms-lease-break-period"] = _SERIALIZER.header("break_period", break_period, "int")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_change_lease_request(
url: str,
*,
lease_id: str,
proposed_lease_id: str,
timeout: Optional[int] = None,
if_modified_since: Optional[datetime.datetime] = None,
if_unmodified_since: Optional[datetime.datetime] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["change"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "change"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-lease-action"] = _SERIALIZER.header("action", action, "str")
_headers["x-ms-lease-id"] = _SERIALIZER.header("lease_id", lease_id, "str")
_headers["x-ms-proposed-lease-id"] = _SERIALIZER.header("proposed_lease_id", proposed_lease_id, "str")
if if_modified_since is not None:
_headers["If-Modified-Since"] = _SERIALIZER.header("if_modified_since", if_modified_since, "rfc-1123")
if if_unmodified_since is not None:
_headers["If-Unmodified-Since"] = _SERIALIZER.header("if_unmodified_since", if_unmodified_since, "rfc-1123")
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="PUT", url=_url, params=_params, headers=_headers, **kwargs)
def build_list_blob_flat_segment_request(
url: str,
*,
prefix: Optional[str] = None,
marker: Optional[str] = None,
maxresults: Optional[int] = None,
include: Optional[List[Union[str, _models.ListBlobsIncludeItem]]] = None,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["list"] = kwargs.pop("comp", _params.pop("comp", "list"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if prefix is not None:
_params["prefix"] = _SERIALIZER.query("prefix", prefix, "str")
if marker is not None:
_params["marker"] = _SERIALIZER.query("marker", marker, "str")
if maxresults is not None:
_params["maxresults"] = _SERIALIZER.query("maxresults", maxresults, "int", minimum=1)
if include is not None:
_params["include"] = _SERIALIZER.query("include", include, "[str]", div=",")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_list_blob_hierarchy_segment_request( # pylint: disable=name-too-long
url: str,
*,
delimiter: str,
prefix: Optional[str] = None,
marker: Optional[str] = None,
maxresults: Optional[int] = None,
include: Optional[List[Union[str, _models.ListBlobsIncludeItem]]] = None,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["list"] = kwargs.pop("comp", _params.pop("comp", "list"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if prefix is not None:
_params["prefix"] = _SERIALIZER.query("prefix", prefix, "str")
_params["delimiter"] = _SERIALIZER.query("delimiter", delimiter, "str")
if marker is not None:
_params["marker"] = _SERIALIZER.query("marker", marker, "str")
if maxresults is not None:
_params["maxresults"] = _SERIALIZER.query("maxresults", maxresults, "int", minimum=1)
if include is not None:
_params["include"] = _SERIALIZER.query("include", include, "[str]", div=",")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
def build_get_account_info_request(
url: str, *, timeout: Optional[int] = None, request_id_parameter: Optional[str] = None, **kwargs: Any
) -> HttpRequest:
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["account"] = kwargs.pop("restype", _params.pop("restype", "account"))
comp: Literal["properties"] = kwargs.pop("comp", _params.pop("comp", "properties"))
version: Literal["2025-01-05"] = kwargs.pop("version", _headers.pop("x-ms-version", "2025-01-05"))
accept = _headers.pop("Accept", "application/xml")
# Construct URL
_url = kwargs.pop("template_url", "{url}")
path_format_arguments = {
"url": _SERIALIZER.url("url", url, "str", skip_quote=True),
}
_url: str = _url.format(**path_format_arguments) # type: ignore
# Construct parameters
_params["restype"] = _SERIALIZER.query("restype", restype, "str")
_params["comp"] = _SERIALIZER.query("comp", comp, "str")
if timeout is not None:
_params["timeout"] = _SERIALIZER.query("timeout", timeout, "int", minimum=0)
# Construct headers
_headers["x-ms-version"] = _SERIALIZER.header("version", version, "str")
if request_id_parameter is not None:
_headers["x-ms-client-request-id"] = _SERIALIZER.header("request_id_parameter", request_id_parameter, "str")
_headers["Accept"] = _SERIALIZER.header("accept", accept, "str")
return HttpRequest(method="GET", url=_url, params=_params, headers=_headers, **kwargs)
class ContainerOperations:
"""
.. warning::
**DO NOT** instantiate this class directly.
Instead, you should access the following operations through
:class:`~azure.storage.blob.AzureBlobStorage`'s
:attr:`container` attribute.
"""
models = _models
def __init__(self, *args, **kwargs):
input_args = list(args)
self._client: PipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
self._config: AzureBlobStorageConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
@distributed_trace
def create( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
metadata: Optional[Dict[str, str]] = None,
access: Optional[Union[str, _models.PublicAccessType]] = None,
request_id_parameter: Optional[str] = None,
container_cpk_scope_info: Optional[_models.ContainerCpkScopeInfo] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""creates a new container under the specified account. If the container with the same name
already exists, the operation fails.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param metadata: Optional. Specifies a user-defined name-value pair associated with the blob.
If no name-value pairs are specified, the operation will copy the metadata from the source blob
or file to the destination blob. If one or more name-value pairs are specified, the destination
blob is created with the specified metadata, and metadata is not copied from the source blob or
file. Note that beginning with version 2009-09-19, metadata names must adhere to the naming
rules for C# identifiers. See Naming and Referencing Containers, Blobs, and Metadata for more
information. Default value is None.
:type metadata: dict[str, str]
:param access: Specifies whether data in the container may be accessed publicly and the level
of access. Known values are: "container" and "blob". Default value is None.
:type access: str or ~azure.storage.blob.models.PublicAccessType
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param container_cpk_scope_info: Parameter group. Default value is None.
:type container_cpk_scope_info: ~azure.storage.blob.models.ContainerCpkScopeInfo
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
cls: ClsType[None] = kwargs.pop("cls", None)
_default_encryption_scope = None
_prevent_encryption_scope_override = None
if container_cpk_scope_info is not None:
_default_encryption_scope = container_cpk_scope_info.default_encryption_scope
_prevent_encryption_scope_override = container_cpk_scope_info.prevent_encryption_scope_override
_request = build_create_request(
url=self._config.url,
timeout=timeout,
metadata=metadata,
access=access,
request_id_parameter=request_id_parameter,
default_encryption_scope=_default_encryption_scope,
prevent_encryption_scope_override=_prevent_encryption_scope_override,
restype=restype,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [201]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def get_properties( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
lease_access_conditions: Optional[_models.LeaseAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""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 timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param lease_access_conditions: Parameter group. Default value is None.
:type lease_access_conditions: ~azure.storage.blob.models.LeaseAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
cls: ClsType[None] = kwargs.pop("cls", None)
_lease_id = None
if lease_access_conditions is not None:
_lease_id = lease_access_conditions.lease_id
_request = build_get_properties_request(
url=self._config.url,
timeout=timeout,
lease_id=_lease_id,
request_id_parameter=request_id_parameter,
restype=restype,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-meta"] = self._deserialize("{str}", response.headers.get("x-ms-meta"))
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-lease-duration"] = self._deserialize("str", response.headers.get("x-ms-lease-duration"))
response_headers["x-ms-lease-state"] = self._deserialize("str", response.headers.get("x-ms-lease-state"))
response_headers["x-ms-lease-status"] = self._deserialize("str", response.headers.get("x-ms-lease-status"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
response_headers["x-ms-blob-public-access"] = self._deserialize(
"str", response.headers.get("x-ms-blob-public-access")
)
response_headers["x-ms-has-immutability-policy"] = self._deserialize(
"bool", response.headers.get("x-ms-has-immutability-policy")
)
response_headers["x-ms-has-legal-hold"] = self._deserialize("bool", response.headers.get("x-ms-has-legal-hold"))
response_headers["x-ms-default-encryption-scope"] = self._deserialize(
"str", response.headers.get("x-ms-default-encryption-scope")
)
response_headers["x-ms-deny-encryption-scope-override"] = self._deserialize(
"bool", response.headers.get("x-ms-deny-encryption-scope-override")
)
response_headers["x-ms-immutable-storage-with-versioning-enabled"] = self._deserialize(
"bool", response.headers.get("x-ms-immutable-storage-with-versioning-enabled")
)
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def delete( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
lease_access_conditions: Optional[_models.LeaseAccessConditions] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""operation marks the specified container for deletion. The container and any blobs contained
within it are later deleted during garbage collection.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param lease_access_conditions: Parameter group. Default value is None.
:type lease_access_conditions: ~azure.storage.blob.models.LeaseAccessConditions
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
cls: ClsType[None] = kwargs.pop("cls", None)
_lease_id = None
_if_modified_since = None
_if_unmodified_since = None
if lease_access_conditions is not None:
_lease_id = lease_access_conditions.lease_id
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
_request = build_delete_request(
url=self._config.url,
timeout=timeout,
lease_id=_lease_id,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
restype=restype,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def set_metadata( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
metadata: Optional[Dict[str, str]] = None,
request_id_parameter: Optional[str] = None,
lease_access_conditions: Optional[_models.LeaseAccessConditions] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""operation sets one or more user-defined name-value pairs for the specified container.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param metadata: Optional. Specifies a user-defined name-value pair associated with the blob.
If no name-value pairs are specified, the operation will copy the metadata from the source blob
or file to the destination blob. If one or more name-value pairs are specified, the destination
blob is created with the specified metadata, and metadata is not copied from the source blob or
file. Note that beginning with version 2009-09-19, metadata names must adhere to the naming
rules for C# identifiers. See Naming and Referencing Containers, Blobs, and Metadata for more
information. Default value is None.
:type metadata: dict[str, str]
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param lease_access_conditions: Parameter group. Default value is None.
:type lease_access_conditions: ~azure.storage.blob.models.LeaseAccessConditions
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["metadata"] = kwargs.pop("comp", _params.pop("comp", "metadata"))
cls: ClsType[None] = kwargs.pop("cls", None)
_lease_id = None
_if_modified_since = None
if lease_access_conditions is not None:
_lease_id = lease_access_conditions.lease_id
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_request = build_set_metadata_request(
url=self._config.url,
timeout=timeout,
lease_id=_lease_id,
metadata=metadata,
if_modified_since=_if_modified_since,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def get_access_policy(
self,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
lease_access_conditions: Optional[_models.LeaseAccessConditions] = None,
**kwargs: Any
) -> List[_models.SignedIdentifier]:
# pylint: disable=line-too-long
"""gets the permissions for the specified container. The permissions indicate whether container
data may be accessed publicly.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param lease_access_conditions: Parameter group. Default value is None.
:type lease_access_conditions: ~azure.storage.blob.models.LeaseAccessConditions
:return: list of SignedIdentifier or the result of cls(response)
:rtype: list[~azure.storage.blob.models.SignedIdentifier]
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["acl"] = kwargs.pop("comp", _params.pop("comp", "acl"))
cls: ClsType[List[_models.SignedIdentifier]] = kwargs.pop("cls", None)
_lease_id = None
if lease_access_conditions is not None:
_lease_id = lease_access_conditions.lease_id
_request = build_get_access_policy_request(
url=self._config.url,
timeout=timeout,
lease_id=_lease_id,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-blob-public-access"] = self._deserialize(
"str", response.headers.get("x-ms-blob-public-access")
)
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
deserialized = self._deserialize("[SignedIdentifier]", pipeline_response.http_response)
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@distributed_trace
def set_access_policy( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
access: Optional[Union[str, _models.PublicAccessType]] = None,
request_id_parameter: Optional[str] = None,
lease_access_conditions: Optional[_models.LeaseAccessConditions] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
container_acl: Optional[List[_models.SignedIdentifier]] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""sets the permissions for the specified container. The permissions indicate whether blobs in a
container may be accessed publicly.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param access: Specifies whether data in the container may be accessed publicly and the level
of access. Known values are: "container" and "blob". Default value is None.
:type access: str or ~azure.storage.blob.models.PublicAccessType
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param lease_access_conditions: Parameter group. Default value is None.
:type lease_access_conditions: ~azure.storage.blob.models.LeaseAccessConditions
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:param container_acl: the acls for the container. Default value is None.
:type container_acl: list[~azure.storage.blob.models.SignedIdentifier]
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["acl"] = kwargs.pop("comp", _params.pop("comp", "acl"))
content_type: str = kwargs.pop("content_type", _headers.pop("Content-Type", "application/xml"))
cls: ClsType[None] = kwargs.pop("cls", None)
_lease_id = None
_if_modified_since = None
_if_unmodified_since = None
if lease_access_conditions is not None:
_lease_id = lease_access_conditions.lease_id
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
serialization_ctxt = {"xml": {"name": "SignedIdentifiers", "wrapped": True, "itemsName": "SignedIdentifier"}}
if container_acl is not None:
_content = self._serialize.body(
container_acl, "[SignedIdentifier]", is_xml=True, serialization_ctxt=serialization_ctxt
)
else:
_content = None
_request = build_set_access_policy_request(
url=self._config.url,
timeout=timeout,
lease_id=_lease_id,
access=access,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
content_type=content_type,
version=self._config.version,
content=_content,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def restore( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
deleted_container_name: Optional[str] = None,
deleted_container_version: Optional[str] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""Restores a previously-deleted container.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param deleted_container_name: Optional. Version 2019-12-12 and later. Specifies the name of
the deleted container to restore. Default value is None.
:type deleted_container_name: str
:param deleted_container_version: Optional. Version 2019-12-12 and later. Specifies the
version of the deleted container to restore. Default value is None.
:type deleted_container_version: str
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["undelete"] = kwargs.pop("comp", _params.pop("comp", "undelete"))
cls: ClsType[None] = kwargs.pop("cls", None)
_request = build_restore_request(
url=self._config.url,
timeout=timeout,
request_id_parameter=request_id_parameter,
deleted_container_name=deleted_container_name,
deleted_container_version=deleted_container_version,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [201]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def rename( # pylint: disable=inconsistent-return-statements
self,
source_container_name: str,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
source_lease_id: Optional[str] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""Renames an existing container.
:param source_container_name: Required. Specifies the name of the container to rename.
Required.
:type source_container_name: str
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param source_lease_id: A lease ID for the source path. If specified, the source path must have
an active lease and the lease ID must match. Default value is None.
:type source_lease_id: str
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["rename"] = kwargs.pop("comp", _params.pop("comp", "rename"))
cls: ClsType[None] = kwargs.pop("cls", None)
_request = build_rename_request(
url=self._config.url,
source_container_name=source_container_name,
timeout=timeout,
request_id_parameter=request_id_parameter,
source_lease_id=source_lease_id,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def submit_batch(
self,
content_length: int,
body: IO[bytes],
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> Iterator[bytes]:
# pylint: disable=line-too-long
"""The Batch operation allows multiple API calls to be embedded into a single HTTP request.
:param content_length: The length of the request. Required.
:type content_length: int
:param body: Initial data. Required.
:type body: IO[bytes]
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:return: Iterator[bytes] or the result of cls(response)
:rtype: Iterator[bytes]
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["batch"] = kwargs.pop("comp", _params.pop("comp", "batch"))
multipart_content_type: str = kwargs.pop(
"multipart_content_type", _headers.pop("Content-Type", "application/xml")
)
cls: ClsType[Iterator[bytes]] = kwargs.pop("cls", None)
_content = body
_request = build_submit_batch_request(
url=self._config.url,
content_length=content_length,
timeout=timeout,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
multipart_content_type=multipart_content_type,
version=self._config.version,
content=_content,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_decompress = kwargs.pop("decompress", True)
_stream = True
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [202]:
try:
response.read() # Load the body in memory and close the socket
except (StreamConsumedError, StreamClosedError):
pass
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["Content-Type"] = self._deserialize("str", response.headers.get("Content-Type"))
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
deserialized = response.stream_download(self._client._pipeline, decompress=_decompress)
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@distributed_trace
def filter_blobs(
self,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
where: Optional[str] = None,
marker: Optional[str] = None,
maxresults: Optional[int] = None,
include: Optional[List[Union[str, _models.FilterBlobsIncludeItem]]] = None,
**kwargs: Any
) -> _models.FilterBlobSegment:
# pylint: disable=line-too-long
"""The Filter Blobs operation enables callers to list blobs in a container whose tags match a
given search expression. Filter blobs searches within the given container.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param where: Filters the results to return only to return only blobs whose tags match the
specified expression. Default value is None.
:type where: str
:param marker: A string value that identifies the portion of the list of containers to be
returned with the next listing operation. The operation returns the NextMarker value within the
response body if the listing operation did not return all containers remaining to be listed
with the current page. The NextMarker value can be used as the value for the marker parameter
in a subsequent call to request the next page of list items. The marker value is opaque to the
client. Default value is None.
:type marker: str
:param maxresults: Specifies the maximum number of containers to return. If the request does
not specify maxresults, or specifies a value greater than 5000, the server will return up to
5000 items. Note that if the listing operation crosses a partition boundary, then the service
will return a continuation token for retrieving the remainder of the results. For this reason,
it is possible that the service will return fewer results than specified by maxresults, or than
the default of 5000. Default value is None.
:type maxresults: int
:param include: Include this parameter to specify one or more datasets to include in the
response. Default value is None.
:type include: list[str or ~azure.storage.blob.models.FilterBlobsIncludeItem]
:return: FilterBlobSegment or the result of cls(response)
:rtype: ~azure.storage.blob.models.FilterBlobSegment
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["blobs"] = kwargs.pop("comp", _params.pop("comp", "blobs"))
cls: ClsType[_models.FilterBlobSegment] = kwargs.pop("cls", None)
_request = build_filter_blobs_request(
url=self._config.url,
timeout=timeout,
request_id_parameter=request_id_parameter,
where=where,
marker=marker,
maxresults=maxresults,
include=include,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
deserialized = self._deserialize("FilterBlobSegment", pipeline_response.http_response)
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@distributed_trace
def acquire_lease( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
duration: Optional[int] = None,
proposed_lease_id: Optional[str] = None,
request_id_parameter: Optional[str] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""[Update] establishes and manages a lock on a container for delete operations. The lock duration
can be 15 to 60 seconds, or can be infinite.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param duration: Specifies the duration of the lease, in seconds, or negative one (-1) for a
lease that never expires. A non-infinite lease can be between 15 and 60 seconds. A lease
duration cannot be changed using renew or change. Default value is None.
:type duration: int
:param proposed_lease_id: Proposed lease ID, in a GUID string format. The Blob service returns
400 (Invalid request) if the proposed lease ID is not in the correct format. See Guid
Constructor (String) for a list of valid GUID string formats. Default value is None.
:type proposed_lease_id: str
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["acquire"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "acquire"))
cls: ClsType[None] = kwargs.pop("cls", None)
_if_modified_since = None
_if_unmodified_since = None
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
_request = build_acquire_lease_request(
url=self._config.url,
timeout=timeout,
duration=duration,
proposed_lease_id=proposed_lease_id,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
comp=comp,
restype=restype,
action=action,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [201]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-lease-id"] = self._deserialize("str", response.headers.get("x-ms-lease-id"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def release_lease( # pylint: disable=inconsistent-return-statements
self,
lease_id: str,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""[Update] establishes and manages a lock on a container for delete operations. The lock duration
can be 15 to 60 seconds, or can be infinite.
:param lease_id: Specifies the current lease ID on the resource. Required.
:type lease_id: str
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["release"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "release"))
cls: ClsType[None] = kwargs.pop("cls", None)
_if_modified_since = None
_if_unmodified_since = None
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
_request = build_release_lease_request(
url=self._config.url,
lease_id=lease_id,
timeout=timeout,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
comp=comp,
restype=restype,
action=action,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def renew_lease( # pylint: disable=inconsistent-return-statements
self,
lease_id: str,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""[Update] establishes and manages a lock on a container for delete operations. The lock duration
can be 15 to 60 seconds, or can be infinite.
:param lease_id: Specifies the current lease ID on the resource. Required.
:type lease_id: str
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["renew"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "renew"))
cls: ClsType[None] = kwargs.pop("cls", None)
_if_modified_since = None
_if_unmodified_since = None
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
_request = build_renew_lease_request(
url=self._config.url,
lease_id=lease_id,
timeout=timeout,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
comp=comp,
restype=restype,
action=action,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-lease-id"] = self._deserialize("str", response.headers.get("x-ms-lease-id"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def break_lease( # pylint: disable=inconsistent-return-statements
self,
timeout: Optional[int] = None,
break_period: Optional[int] = None,
request_id_parameter: Optional[str] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""[Update] establishes and manages a lock on a container for delete operations. The lock duration
can be 15 to 60 seconds, or can be infinite.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param break_period: For a break operation, proposed duration the lease should continue before
it is broken, in seconds, between 0 and 60. This break period is only used if it is shorter
than the time remaining on the lease. If longer, the time remaining on the lease is used. A new
lease will not be available before the break period has expired, but the lease may be held for
longer than the break period. If this header does not appear with a break operation, a
fixed-duration lease breaks after the remaining lease period elapses, and an infinite lease
breaks immediately. Default value is None.
:type break_period: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["break"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "break"))
cls: ClsType[None] = kwargs.pop("cls", None)
_if_modified_since = None
_if_unmodified_since = None
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
_request = build_break_lease_request(
url=self._config.url,
timeout=timeout,
break_period=break_period,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
comp=comp,
restype=restype,
action=action,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [202]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-lease-time"] = self._deserialize("int", response.headers.get("x-ms-lease-time"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def change_lease( # pylint: disable=inconsistent-return-statements
self,
lease_id: str,
proposed_lease_id: str,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
modified_access_conditions: Optional[_models.ModifiedAccessConditions] = None,
**kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""[Update] establishes and manages a lock on a container for delete operations. The lock duration
can be 15 to 60 seconds, or can be infinite.
:param lease_id: Specifies the current lease ID on the resource. Required.
:type lease_id: str
:param proposed_lease_id: Proposed lease ID, in a GUID string format. The Blob service returns
400 (Invalid request) if the proposed lease ID is not in the correct format. See Guid
Constructor (String) for a list of valid GUID string formats. Required.
:type proposed_lease_id: str
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:param modified_access_conditions: Parameter group. Default value is None.
:type modified_access_conditions: ~azure.storage.blob.models.ModifiedAccessConditions
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = case_insensitive_dict(kwargs.pop("headers", {}) or {})
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
comp: Literal["lease"] = kwargs.pop("comp", _params.pop("comp", "lease"))
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
action: Literal["change"] = kwargs.pop("action", _headers.pop("x-ms-lease-action", "change"))
cls: ClsType[None] = kwargs.pop("cls", None)
_if_modified_since = None
_if_unmodified_since = None
if modified_access_conditions is not None:
_if_modified_since = modified_access_conditions.if_modified_since
_if_unmodified_since = modified_access_conditions.if_unmodified_since
_request = build_change_lease_request(
url=self._config.url,
lease_id=lease_id,
proposed_lease_id=proposed_lease_id,
timeout=timeout,
if_modified_since=_if_modified_since,
if_unmodified_since=_if_unmodified_since,
request_id_parameter=request_id_parameter,
comp=comp,
restype=restype,
action=action,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["ETag"] = self._deserialize("str", response.headers.get("ETag"))
response_headers["Last-Modified"] = self._deserialize("rfc-1123", response.headers.get("Last-Modified"))
response_headers["x-ms-lease-id"] = self._deserialize("str", response.headers.get("x-ms-lease-id"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore
@distributed_trace
def list_blob_flat_segment(
self,
prefix: Optional[str] = None,
marker: Optional[str] = None,
maxresults: Optional[int] = None,
include: Optional[List[Union[str, _models.ListBlobsIncludeItem]]] = None,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> _models.ListBlobsFlatSegmentResponse:
# pylint: disable=line-too-long
"""[Update] The List Blobs operation returns a list of the blobs under the specified container.
:param prefix: Filters the results to return only containers whose name begins with the
specified prefix. Default value is None.
:type prefix: str
:param marker: A string value that identifies the portion of the list of containers to be
returned with the next listing operation. The operation returns the NextMarker value within the
response body if the listing operation did not return all containers remaining to be listed
with the current page. The NextMarker value can be used as the value for the marker parameter
in a subsequent call to request the next page of list items. The marker value is opaque to the
client. Default value is None.
:type marker: str
:param maxresults: Specifies the maximum number of containers to return. If the request does
not specify maxresults, or specifies a value greater than 5000, the server will return up to
5000 items. Note that if the listing operation crosses a partition boundary, then the service
will return a continuation token for retrieving the remainder of the results. For this reason,
it is possible that the service will return fewer results than specified by maxresults, or than
the default of 5000. Default value is None.
:type maxresults: int
:param include: Include this parameter to specify one or more datasets to include in the
response. Default value is None.
:type include: list[str or ~azure.storage.blob.models.ListBlobsIncludeItem]
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:return: ListBlobsFlatSegmentResponse or the result of cls(response)
:rtype: ~azure.storage.blob.models.ListBlobsFlatSegmentResponse
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["list"] = kwargs.pop("comp", _params.pop("comp", "list"))
cls: ClsType[_models.ListBlobsFlatSegmentResponse] = kwargs.pop("cls", None)
_request = build_list_blob_flat_segment_request(
url=self._config.url,
prefix=prefix,
marker=marker,
maxresults=maxresults,
include=include,
timeout=timeout,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["Content-Type"] = self._deserialize("str", response.headers.get("Content-Type"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
deserialized = self._deserialize("ListBlobsFlatSegmentResponse", pipeline_response.http_response)
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@distributed_trace
def list_blob_hierarchy_segment(
self,
delimiter: str,
prefix: Optional[str] = None,
marker: Optional[str] = None,
maxresults: Optional[int] = None,
include: Optional[List[Union[str, _models.ListBlobsIncludeItem]]] = None,
timeout: Optional[int] = None,
request_id_parameter: Optional[str] = None,
**kwargs: Any
) -> _models.ListBlobsHierarchySegmentResponse:
# pylint: disable=line-too-long
"""[Update] The List Blobs operation returns a list of the blobs under the specified container.
:param delimiter: When the request includes this parameter, the operation returns a BlobPrefix
element in the response body that acts as a placeholder for all blobs whose names begin with
the same substring up to the appearance of the delimiter character. The delimiter may be a
single character or a string. Required.
:type delimiter: str
:param prefix: Filters the results to return only containers whose name begins with the
specified prefix. Default value is None.
:type prefix: str
:param marker: A string value that identifies the portion of the list of containers to be
returned with the next listing operation. The operation returns the NextMarker value within the
response body if the listing operation did not return all containers remaining to be listed
with the current page. The NextMarker value can be used as the value for the marker parameter
in a subsequent call to request the next page of list items. The marker value is opaque to the
client. Default value is None.
:type marker: str
:param maxresults: Specifies the maximum number of containers to return. If the request does
not specify maxresults, or specifies a value greater than 5000, the server will return up to
5000 items. Note that if the listing operation crosses a partition boundary, then the service
will return a continuation token for retrieving the remainder of the results. For this reason,
it is possible that the service will return fewer results than specified by maxresults, or than
the default of 5000. Default value is None.
:type maxresults: int
:param include: Include this parameter to specify one or more datasets to include in the
response. Default value is None.
:type include: list[str or ~azure.storage.blob.models.ListBlobsIncludeItem]
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:return: ListBlobsHierarchySegmentResponse or the result of cls(response)
:rtype: ~azure.storage.blob.models.ListBlobsHierarchySegmentResponse
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["container"] = kwargs.pop("restype", _params.pop("restype", "container"))
comp: Literal["list"] = kwargs.pop("comp", _params.pop("comp", "list"))
cls: ClsType[_models.ListBlobsHierarchySegmentResponse] = kwargs.pop("cls", None)
_request = build_list_blob_hierarchy_segment_request(
url=self._config.url,
delimiter=delimiter,
prefix=prefix,
marker=marker,
maxresults=maxresults,
include=include,
timeout=timeout,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["Content-Type"] = self._deserialize("str", response.headers.get("Content-Type"))
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
deserialized = self._deserialize("ListBlobsHierarchySegmentResponse", pipeline_response.http_response)
if cls:
return cls(pipeline_response, deserialized, response_headers) # type: ignore
return deserialized # type: ignore
@distributed_trace
def get_account_info( # pylint: disable=inconsistent-return-statements
self, timeout: Optional[int] = None, request_id_parameter: Optional[str] = None, **kwargs: Any
) -> None:
# pylint: disable=line-too-long
"""Returns the sku name and account kind.
:param timeout: The timeout parameter is expressed in seconds. For more information, see
:code:`Setting
Timeouts for Blob Service Operations.`. Default value is None.
:type timeout: int
:param request_id_parameter: Provides a client-generated, opaque value with a 1 KB character
limit that is recorded in the analytics logs when storage analytics logging is enabled. Default
value is None.
:type request_id_parameter: str
:return: None or the result of cls(response)
:rtype: None
:raises ~azure.core.exceptions.HttpResponseError:
"""
error_map: MutableMapping = {
401: ClientAuthenticationError,
404: ResourceNotFoundError,
409: ResourceExistsError,
304: ResourceNotModifiedError,
}
error_map.update(kwargs.pop("error_map", {}) or {})
_headers = kwargs.pop("headers", {}) or {}
_params = case_insensitive_dict(kwargs.pop("params", {}) or {})
restype: Literal["account"] = kwargs.pop("restype", _params.pop("restype", "account"))
comp: Literal["properties"] = kwargs.pop("comp", _params.pop("comp", "properties"))
cls: ClsType[None] = kwargs.pop("cls", None)
_request = build_get_account_info_request(
url=self._config.url,
timeout=timeout,
request_id_parameter=request_id_parameter,
restype=restype,
comp=comp,
version=self._config.version,
headers=_headers,
params=_params,
)
_request.url = self._client.format_url(_request.url)
_stream = False
pipeline_response: PipelineResponse = self._client._pipeline.run( # pylint: disable=protected-access
_request, stream=_stream, **kwargs
)
response = pipeline_response.http_response
if response.status_code not in [200]:
map_error(status_code=response.status_code, response=response, error_map=error_map)
error = self._deserialize.failsafe_deserialize(_models.StorageError, pipeline_response)
raise HttpResponseError(response=response, model=error)
response_headers = {}
response_headers["x-ms-client-request-id"] = self._deserialize(
"str", response.headers.get("x-ms-client-request-id")
)
response_headers["x-ms-request-id"] = self._deserialize("str", response.headers.get("x-ms-request-id"))
response_headers["x-ms-version"] = self._deserialize("str", response.headers.get("x-ms-version"))
response_headers["Date"] = self._deserialize("rfc-1123", response.headers.get("Date"))
response_headers["x-ms-sku-name"] = self._deserialize("str", response.headers.get("x-ms-sku-name"))
response_headers["x-ms-account-kind"] = self._deserialize("str", response.headers.get("x-ms-account-kind"))
response_headers["x-ms-is-hns-enabled"] = self._deserialize("bool", response.headers.get("x-ms-is-hns-enabled"))
if cls:
return cls(pipeline_response, None, response_headers) # type: ignore