<s> """<STR_LIT>""" <EOL> import os <EOL> import random <EOL> import uuid <EOL> from urllib . parse import quote <EOL> import requests <EOL> OAUTH_ENDPOINT_TEMPLATE = '<STR_LIT>' <EOL> MANAGEMENT_ROOT = '<STR_LIT>' <EOL> SUBSCRIPTION_ENDPOINT = '<STR_LIT>' <EOL> RESOURCE_ENDPOINT = '<STR_LIT>' <EOL> DEPLOYMENT_URI = '<STR_LIT>' <EOL> RESOURCES = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def setup_requests_session ( token ) : <EOL> session = requests . Session ( ) <EOL> session . params = { '<STR_LIT>' : AzureClient . API_VERSION } <EOL> session . headers = { '<STR_LIT>' : '<STR_LIT>' . format ( token ) } <EOL> return session <EOL> class AzureClientException ( Exception ) : <EOL> pass <EOL> class AzureClientTemplateException ( AzureClientException ) : <EOL> def __init__ ( self , template , errors ) : <EOL> self . message = '<STR_LIT>' . format ( template , <EOL> errors ) <EOL> self . template = template <EOL> self . errors = errors <EOL> super ( AzureClientTemplateException , self ) . __init__ ( self . message ) <EOL> class AzureClient ( object ) : <EOL> API_VERSION = '<STR_LIT>' <EOL> @ staticmethod <EOL> def create_from_env_creds ( ** kwargs ) : <EOL> subscription_id = os . environ . get ( '<STR_LIT>' ) <EOL> tenant_id = os . environ . get ( '<STR_LIT>' ) <EOL> client_id = os . environ . get ( '<STR_LIT>' ) <EOL> client_secret = os . environ . get ( '<STR_LIT>' ) <EOL> return AzureClient ( subscription_id = subscription_id , <EOL> tenant_id = tenant_id , <EOL> client_id = client_id , <EOL> client_secret = client_secret , <EOL> ** kwargs <EOL> ) <EOL> def __init__ ( self , subscription_id , tenant_id , client_id , client_secret , <EOL> random_resource_group_name_prefix = '<STR_LIT>' , ** kwargs ) : <EOL> self . _subscription_id = subscription_id <EOL> self . _tenant_id = tenant_id <EOL> self . _client_id = client_id <EOL> self . _client_secret = client_secret <EOL> self . _token = None <EOL> self . _random_resource_group_name_prefix = random_resource_group_name_prefix <EOL> self . _resource_group_name = self . get_random_resource_group_name ( ) <EOL> self . _deployment_name = self . get_random_deployment_name ( ) <EOL> self . _session = setup_requests_session ( self . get_auth_token ( ) ) <EOL> self . _url_space = self . _generate_api_urls ( ) <EOL> @ property <EOL> def subscription_endpoint ( self ) : <EOL> return self . _get_subscription_endpoint ( self . _subscription_id ) <EOL> def _get_subscription_endpoint ( self , sub_id ) : <EOL> return SUBSCRIPTION_ENDPOINT . format ( <EOL> base_url = MANAGEMENT_ROOT , <EOL> sub_id = sub_id <EOL> ) <EOL> @ property <EOL> def resource_endpoint ( self ) : <EOL> return self . _get_resource_endpoint ( <EOL> self . subscription_endpoint , <EOL> self . _resource_group_name ) <EOL> def _get_resource_endpoint ( self , sub_url , resource_group ) : <EOL> return RESOURCE_ENDPOINT . format ( <EOL> sub_url = sub_url , <EOL> resource_grp = resource_group <EOL> ) <EOL> @ property <EOL> def deployment_endpoint ( self ) : <EOL> return DEPLOYMENT_URI . format ( <EOL> res_grp_url = self . resource_endpoint , <EOL> deploy_name = self . _deployment_name <EOL> ) <EOL> def _generate_api_urls ( self ) : <EOL> '''<STR_LIT>''' <EOL> def url_format ( u ) : <EOL> return '<STR_LIT>' . format ( self . deployment_endpoint , u ) <EOL> urls = dict ( zip ( RESOURCES , map ( url_format , RESOURCES ) ) ) <EOL> print ( '<STR_LIT>' . format ( urls ) ) <EOL> return urls <EOL> def get_token_from_client_credentials ( self , endpoint , client_id , client_secret ) : <EOL> payload = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : client_id , <EOL> '<STR_LIT>' : client_secret , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> response = requests . post ( endpoint , data = payload ) . json ( ) <EOL> return response [ '<STR_LIT>' ] <EOL> def get_auth_token ( self ) : <EOL> self . _token = self . _token or self . get_token_from_client_credentials ( <EOL> endpoint = OAUTH_ENDPOINT_TEMPLATE . format ( self . _tenant_id ) , <EOL> client_id = self . _client_id , <EOL> client_secret = self . _client_secret <EOL> ) <EOL> return self . _token <EOL> def verify ( self , template_body_json , template_parameters ) : <EOL> endpoint = self . _url_space [ '<STR_LIT>' ] <EOL> template_json = self . get_template_json ( template_body_json = template_body_json , <EOL> template_parameters = template_parameters ) <EOL> r = self . _session . post ( endpoint , json = template_json ) <EOL> errors = r . json ( ) . get ( '<STR_LIT:error>' ) <EOL> if errors : <EOL> raise AzureClientTemplateException ( template_json , errors ) <EOL> return r <EOL> def get_template_deployment ( self ) : <EOL> return self . _session . get ( self . _url_space [ '<STR_LIT>' ] ) . json ( ) <EOL> def get_template_deployment_status ( self ) : <EOL> template_deployment = self . get_template_deployment ( ) <EOL> return template_deployment . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' ) <EOL> def create_template_deployment ( self , template_body_json , template_parameters ) : <EOL> template_json = self . get_template_json ( template_body_json = template_body_json , <EOL> template_parameters = template_parameters ) <EOL> r = self . _session . put ( self . _url_space [ '<STR_LIT>' ] , json = template_json ) <EOL> return r . json ( ) <EOL> def cancel_template_deployment ( self ) : <EOL> endpoint = self . _url_space [ '<STR_LIT>' ] <EOL> r = self . _session . post ( endpoint ) <EOL> return r <EOL> def delete_template_deployment ( self ) : <EOL> r = self . _session . delete ( self . _url_space [ '<STR_LIT>' ] ) <EOL> return r <EOL> def get_template_json ( self , template_body_json , template_parameters ) : <EOL> return { <EOL> "<STR_LIT>" : <EOL> { <EOL> "<STR_LIT>" : template_body_json , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : template_parameters <EOL> } <EOL> } <EOL> def get_resource_group ( self , resource_group_name = None ) : <EOL> resource_group_name = resource_group_name or self . _resource_group_name <EOL> endpoint = self . _get_resource_endpoint ( self . subscription_endpoint , resource_group_name ) <EOL> return self . _session . get ( endpoint ) . json ( ) <EOL> def create_resource_group ( self , location = '<STR_LIT>' , resource_group_name = None ) : <EOL> resource_group_name = resource_group_name or self . _resource_group_name <EOL> endpoint = self . _get_resource_endpoint ( self . subscription_endpoint , resource_group_name ) <EOL> r = self . _session . put ( endpoint , json = { '<STR_LIT:location>' : location } ) <EOL> if r . ok : <EOL> self . _resource_group_name = resource_group_name <EOL> self . _url_space = self . _generate_api_urls ( ) <EOL> return r . json ( ) <EOL> def delete_resource_group ( self , resource_group_name = None ) : <EOL> resource_group_name = resource_group_name or self . _resource_group_name <EOL> endpoint = self . _get_resource_endpoint ( self . subscription_endpoint , resource_group_name ) <EOL> r = self . _session . delete ( endpoint ) <EOL> return r <EOL> def status_delete_resource_group ( self , poll_location ) : <EOL> '''<STR_LIT>''' <EOL> r = self . _session . get ( poll_location ) <EOL> return r <EOL> def get_random_resource_group_name ( self ) : <EOL> '''<STR_LIT>''' <EOL> prefix = self . _random_resource_group_name_prefix [ : <NUM_LIT:8> ] <EOL> rand_string = '<STR_LIT>' . join ( random . choice ( '<STR_LIT>' ) for n in range ( <NUM_LIT:10> ) ) <EOL> return '<STR_LIT>' . format ( prefix , rand_string ) <EOL> def get_random_deployment_name ( self ) : <EOL> return '<STR_LIT>' . format ( uuid . uuid4 ( ) . hex ) <EOL> def list_template_deployment_operations ( self , provisioning_state_filter = None ) : <EOL> url = self . _url_space [ '<STR_LIT>' ] <EOL> if provisioning_state_filter : <EOL> url += '<STR_LIT>' + quote ( provisioning_state_filter ) + '<STR_LIT>' <EOL> r = self . _session . get ( url ) <EOL> return r . json ( ) <EOL> class TemplateProvisioningState ( object ) : <EOL> ACCEPTED = '<STR_LIT>' <EOL> READY = '<STR_LIT>' <EOL> CANCELED = '<STR_LIT>' <EOL> FAILED = '<STR_LIT>' <EOL> DELETED = '<STR_LIT>' <EOL> SUCCEEDED = '<STR_LIT>' <EOL> RUNNING = '<STR_LIT>' <EOL> DELETING = '<STR_LIT>' </s>
