Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_storage_key(
cli_ctx,
storage_account,
resource_group_name,
use_secondary_key):
'''
Gets storage account key by querying storage ARM API.
'''
from azure.mgmt.storage import StorageManagementClient
from azure.cli.core.commands.client_factory import get_mgmt_service_client
# Get storage keys
client = get_mgmt_service_client(cli_ctx, StorageManagementClient)
keys = client.storage_accounts.list_keys(
resource_group_name=resource_group_name,
account_name=storage_account)
# Choose storage key
index = 1 if use_secondary_key else 0
return keys.keys[index].value # pylint: disable=no-member
def _validator(cmd, namespace):
from msrestazure.tools import parse_resource_id
location = namespace.location
network_client = get_mgmt_service_client(cmd.cli_ctx, CUSTOM_NW_CONNECTION_MONITOR).network_watchers
watcher = next((x for x in network_client.list_all() if x.location.lower() == location.lower()), None)
if not watcher:
raise CLIError("network watcher is not enabled for region '{}'.".format(location))
id_parts = parse_resource_id(watcher.id)
setattr(namespace, rg_name, id_parts['resource_group'])
setattr(namespace, watcher_name, id_parts['name'])
if remove:
del namespace.location
def frontdoor_client_factory(cli_ctx, aux_subscriptions=None, **_):
from azure.cli.core.commands.client_factory import get_mgmt_service_client
from azext_front_door.vendored_sdks import FrontDoorManagementClient
return get_mgmt_service_client(cli_ctx, FrontDoorManagementClient, aux_subscriptions=aux_subscriptions)
def web_client_factory(cli_ctx, **_):
from azure.mgmt.web import WebSiteManagementClient
from azure.cli.core.commands.client_factory import get_mgmt_service_client
return get_mgmt_service_client(cli_ctx, WebSiteManagementClient)
def create_privatedns_link(cmd, resource_group_name, private_zone_name, virtual_network_link_name, virtual_network, registration_enabled, tags=None):
from azext_privatedns.vendored_sdks import PrivateDnsManagementClient
from azext_privatedns.vendored_sdks.models import VirtualNetworkLink
link = VirtualNetworkLink(location='global', tags=tags)
if registration_enabled is not None:
link.registration_enabled = registration_enabled
if virtual_network is not None:
link.virtual_network = virtual_network
aux_subscription = parse_resource_id(virtual_network.id)['subscription']
client = get_mgmt_service_client(cmd.cli_ctx, PrivateDnsManagementClient, aux_subscriptions=[aux_subscription]).virtual_network_links
return client.create_or_update(resource_group_name, private_zone_name, virtual_network_link_name, link, if_none_match='*')
def web_client_factory(cli_ctx, **_):
from azure.mgmt.web import WebSiteManagementClient
mgmt_client = get_mgmt_service_client(cli_ctx, WebSiteManagementClient)
return mgmt_client
def dms_client_factory(cli_ctx, *_):
from azure.cli.core.commands.client_factory import get_mgmt_service_client
from azure.mgmt.datamigration import DataMigrationServiceClient
return get_mgmt_service_client(cli_ctx, DataMigrationServiceClient)
if client_id:
from azure.common.credentials import ServicePrincipalCredentials
credentials = ServicePrincipalCredentials(
client_id=client_id,
secret=getenv(CLIENT_SECRET),
tenant=getenv(TENANT_ID))
else:
from msrest.authentication import Authentication # pylint: disable=import-error
credentials = Authentication()
return MySQLManagementClient(
subscription_id=getenv(SUB_ID_OVERRIDE),
base_url=rm_uri_override,
credentials=credentials)
# Normal production scenario.
return get_mgmt_service_client(cli_ctx, MySQLManagementClient)
def cf_resource_groups(cli_ctx, subscription_id=None):
return get_mgmt_service_client(cli_ctx, ResourceType.MGMT_RESOURCE_RESOURCES,
subscription_id=subscription_id).resource_groups