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