1# ------------------------------------------------------------------------- 2# Copyright (c) Microsoft Corporation. All rights reserved. 3# Licensed under the MIT License. See License.txt in the project root for 4# license information. 5# -------------------------------------------------------------------------- 6# pylint: disable=no-self-use 7from typing import ( # pylint: disable=unused-import 8 Tuple, Dict, List, 9 TYPE_CHECKING 10) 11 12from ._models import BlobType, CopyProperties, ContentSettings, LeaseProperties, BlobProperties 13from ._shared.models import get_enum_value 14 15from ._shared.response_handlers import deserialize_metadata 16from ._models import ContainerProperties, BlobAnalyticsLogging, Metrics, CorsRule, RetentionPolicy, \ 17 StaticWebsite, ObjectReplicationPolicy, ObjectReplicationRule 18 19if TYPE_CHECKING: 20 from ._generated.models import PageList 21 22 23def deserialize_pipeline_response_into_cls(cls_method, response, obj, headers): 24 try: 25 deserialized_response = response.http_response 26 except AttributeError: 27 deserialized_response = response 28 return cls_method(deserialized_response, obj, headers) 29 30 31def deserialize_blob_properties(response, obj, headers): 32 blob_properties = BlobProperties( 33 metadata=deserialize_metadata(response, obj, headers), 34 object_replication_source_properties=deserialize_ors_policies(response.http_response.headers), 35 **headers 36 ) 37 if 'Content-Range' in headers: 38 if 'x-ms-blob-content-md5' in headers: 39 blob_properties.content_settings.content_md5 = headers['x-ms-blob-content-md5'] 40 else: 41 blob_properties.content_settings.content_md5 = None 42 return blob_properties 43 44 45def deserialize_ors_policies(policy_dictionary): 46 47 if policy_dictionary is None: 48 return None 49 # For source blobs (blobs that have policy ids and rule ids applied to them), 50 # the header will be formatted as "x-ms-or-<policy_id>_<rule_id>: {Complete, Failed}". 51 # The value of this header is the status of the replication. 52 or_policy_status_headers = {key: val for key, val in policy_dictionary.items() 53 if 'or-' in key and key != 'x-ms-or-policy-id'} 54 55 parsed_result = {} 56 57 for key, val in or_policy_status_headers.items(): 58 # list blobs gives or-policy_rule and get blob properties gives x-ms-or-policy_rule 59 policy_and_rule_ids = key.split('or-')[1].split('_') 60 policy_id = policy_and_rule_ids[0] 61 rule_id = policy_and_rule_ids[1] 62 63 # If we are seeing this policy for the first time, create a new list to store rule_id -> result 64 parsed_result[policy_id] = parsed_result.get(policy_id) or list() 65 parsed_result[policy_id].append(ObjectReplicationRule(rule_id=rule_id, status=val)) 66 67 result_list = [ObjectReplicationPolicy(policy_id=k, rules=v) for k, v in parsed_result.items()] 68 69 return result_list 70 71 72def deserialize_blob_stream(response, obj, headers): 73 blob_properties = deserialize_blob_properties(response, obj, headers) 74 obj.properties = blob_properties 75 return response.http_response.location_mode, obj 76 77 78def deserialize_container_properties(response, obj, headers): 79 metadata = deserialize_metadata(response, obj, headers) 80 container_properties = ContainerProperties( 81 metadata=metadata, 82 **headers 83 ) 84 return container_properties 85 86 87def get_page_ranges_result(ranges): 88 # type: (PageList) -> Tuple[List[Dict[str, int]], List[Dict[str, int]]] 89 page_range = [] # type: ignore 90 clear_range = [] # type: List 91 if ranges.page_range: 92 page_range = [{'start': b.start, 'end': b.end} for b in ranges.page_range] # type: ignore 93 if ranges.clear_range: 94 clear_range = [{'start': b.start, 'end': b.end} for b in ranges.clear_range] 95 return page_range, clear_range # type: ignore 96 97 98def service_stats_deserialize(generated): 99 """Deserialize a ServiceStats objects into a dict. 100 """ 101 return { 102 'geo_replication': { 103 'status': generated.geo_replication.status, 104 'last_sync_time': generated.geo_replication.last_sync_time, 105 } 106 } 107 108 109def service_properties_deserialize(generated): 110 """Deserialize a ServiceProperties objects into a dict. 111 """ 112 return { 113 'analytics_logging': BlobAnalyticsLogging._from_generated(generated.logging), # pylint: disable=protected-access 114 'hour_metrics': Metrics._from_generated(generated.hour_metrics), # pylint: disable=protected-access 115 'minute_metrics': Metrics._from_generated(generated.minute_metrics), # pylint: disable=protected-access 116 'cors': [CorsRule._from_generated(cors) for cors in generated.cors], # pylint: disable=protected-access 117 'target_version': generated.default_service_version, # pylint: disable=protected-access 118 'delete_retention_policy': RetentionPolicy._from_generated(generated.delete_retention_policy), # pylint: disable=protected-access 119 'static_website': StaticWebsite._from_generated(generated.static_website), # pylint: disable=protected-access 120 } 121 122 123def get_blob_properties_from_generated_code(generated): 124 blob = BlobProperties() 125 blob.name = generated.name 126 blob_type = get_enum_value(generated.properties.blob_type) 127 blob.blob_type = BlobType(blob_type) if blob_type else None 128 blob.etag = generated.properties.etag 129 blob.deleted = generated.deleted 130 blob.snapshot = generated.snapshot 131 blob.is_append_blob_sealed = generated.properties.is_sealed 132 blob.metadata = generated.metadata.additional_properties if generated.metadata else {} 133 blob.encrypted_metadata = generated.metadata.encrypted if generated.metadata else None 134 blob.lease = LeaseProperties._from_generated(generated) # pylint: disable=protected-access 135 blob.copy = CopyProperties._from_generated(generated) # pylint: disable=protected-access 136 blob.last_modified = generated.properties.last_modified 137 blob.creation_time = generated.properties.creation_time 138 blob.content_settings = ContentSettings._from_generated(generated) # pylint: disable=protected-access 139 blob.size = generated.properties.content_length 140 blob.page_blob_sequence_number = generated.properties.blob_sequence_number 141 blob.server_encrypted = generated.properties.server_encrypted 142 blob.encryption_scope = generated.properties.encryption_scope 143 blob.deleted_time = generated.properties.deleted_time 144 blob.remaining_retention_days = generated.properties.remaining_retention_days 145 blob.blob_tier = generated.properties.access_tier 146 blob.rehydrate_priority = generated.properties.rehydrate_priority 147 blob.blob_tier_inferred = generated.properties.access_tier_inferred 148 blob.archive_status = generated.properties.archive_status 149 blob.blob_tier_change_time = generated.properties.access_tier_change_time 150 blob.version_id = generated.version_id 151 blob.is_current_version = generated.is_current_version 152 blob.tag_count = generated.properties.tag_count 153 blob.tags = parse_tags(generated.blob_tags) # pylint: disable=protected-access 154 blob.object_replication_source_properties = deserialize_ors_policies(generated.object_replication_metadata) 155 blob.last_accessed_on = generated.properties.last_accessed_on 156 return blob 157 158 159def parse_tags(generated_tags): 160 # type: (Optional[List[BlobTag]]) -> Union[Dict[str, str], None] 161 """Deserialize a list of BlobTag objects into a dict. 162 """ 163 if generated_tags: 164 tag_dict = {t.key: t.value for t in generated_tags.blob_tag_set} 165 return tag_dict 166 return None 167