azure-cli/src/azure-cli/azure/cli/command_modules/acr/replication.py at e6b12f0099b63d531335bc9e8fe3e63c0401550f · lizMSFT/azure-cli · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
# --------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# --------------------------------------------------------------------------------------------

from knack.log import get_logger
from azure.cli.core.util import CLIError

from ._utils import (
    get_resource_group_name_by_registry_name,
    validate_premium_registry
)

logger = get_logger(__name__)


REPLICATIONS_NOT_SUPPORTED = 'Replications are only supported for managed registries in Premium SKU.'


def acr_replication_list(cmd, client, registry_name, resource_group_name=None):
    _, resource_group_name = validate_premium_registry(
        cmd, registry_name, resource_group_name, REPLICATIONS_NOT_SUPPORTED)
    return client.list(resource_group_name, registry_name)


def acr_replication_create(cmd,
                           client,
                           location,
                           registry_name,
                           resource_group_name=None,
                           replication_name=None,
                           region_endpoint_enabled=None,
                           global_endpoint_routing=None,
                           zone_redundancy=None,
                           tags=None):
    registry, resource_group_name = validate_premium_registry(
        cmd, registry_name, resource_group_name, REPLICATIONS_NOT_SUPPORTED)

    # --global-endpoint-routing takes precedence over deprecated --region-endpoint-enabled
    if global_endpoint_routing is not None:
        region_endpoint_enabled = global_endpoint_routing

    normalized_location = "".join(location.split()).lower()
    if registry.location == normalized_location:
        raise CLIError('Replication could not be created in the same location as the registry.')

    from msrest.exceptions import ValidationError
    ReplicationType = cmd.get_models('Replication')

    replication_name = replication_name or normalized_location
    replication_properties = ReplicationType(
        location=location,
        region_endpoint_enabled=region_endpoint_enabled,
        zone_redundancy=zone_redundancy,
        tags=tags)

    try:
        return client.begin_create(
            resource_group_name=resource_group_name,
            registry_name=registry_name,
            replication_name=replication_name,
            replication=replication_properties
        )
    except ValidationError as e:
        raise CLIError(e)


def acr_replication_delete(cmd,
                           client,
                           replication_name,
                           registry_name,
                           resource_group_name=None):
    _, resource_group_name = validate_premium_registry(
        cmd, registry_name, resource_group_name, REPLICATIONS_NOT_SUPPORTED)
    return client.begin_delete(resource_group_name, registry_name, replication_name)


def acr_replication_show(cmd,
                         client,
                         replication_name,
                         registry_name,
                         resource_group_name=None):
    _, resource_group_name = validate_premium_registry(
        cmd, registry_name, resource_group_name, REPLICATIONS_NOT_SUPPORTED)
    return client.get(resource_group_name, registry_name, replication_name)


def acr_replication_update_custom(instance, region_endpoint_enabled=None, global_endpoint_routing=None, tags=None):
    if tags is not None:
        instance.tags = tags

    # --global-endpoint-routing takes precedence over deprecated --region-endpoint-enabled
    effective_value = global_endpoint_routing if global_endpoint_routing is not None \
        else region_endpoint_enabled
    if effective_value is not None:
        instance.region_endpoint_enabled = effective_value

    return instance


def acr_replication_update_get(cmd):
    """Returns an empty ReplicationUpdateParameters object.
    """
    ReplicationUpdateParameters = cmd.get_models('ReplicationUpdateParameters')
    return ReplicationUpdateParameters()


def acr_replication_update_set(cmd,
                               client,
                               replication_name,
                               registry_name,
                               resource_group_name=None,
                               parameters=None):

    resource_group_name = get_resource_group_name_by_registry_name(
        cmd.cli_ctx, registry_name, resource_group_name)
    ReplicationUpdateParameters = cmd.get_models('ReplicationUpdateParameters')
    replication_update_parameters = ReplicationUpdateParameters(
        region_endpoint_enabled=parameters.region_endpoint_enabled,
        tags=parameters.tags
    )
    return client.begin_update(
        resource_group_name=resource_group_name,
        registry_name=registry_name,
        replication_name=replication_name,
        replication_update_parameters=replication_update_parameters)