<s> import boto <EOL> from boto . compat import json <EOL> from boto . connection import AWSQueryConnection <EOL> from boto . regioninfo import RegionInfo <EOL> from boto . exception import JSONResponseError <EOL> from boto . cognito . identity import exceptions <EOL> class CognitoIdentityConnection ( AWSQueryConnection ) : <EOL> """<STR_LIT>""" <EOL> APIVersion = "<STR_LIT>" <EOL> DefaultRegionName = "<STR_LIT>" <EOL> DefaultRegionEndpoint = "<STR_LIT>" <EOL> ServiceName = "<STR_LIT>" <EOL> TargetPrefix = "<STR_LIT>" <EOL> ResponseError = JSONResponseError <EOL> _faults = { <EOL> "<STR_LIT>" : exceptions . LimitExceededException , <EOL> "<STR_LIT>" : exceptions . ResourceConflictException , <EOL> "<STR_LIT>" : exceptions . DeveloperUserAlreadyRegisteredException , <EOL> "<STR_LIT>" : exceptions . TooManyRequestsException , <EOL> "<STR_LIT>" : exceptions . InvalidParameterException , <EOL> "<STR_LIT>" : exceptions . ResourceNotFoundException , <EOL> "<STR_LIT>" : exceptions . InternalErrorException , <EOL> "<STR_LIT>" : exceptions . NotAuthorizedException , <EOL> } <EOL> def __init__ ( self , ** kwargs ) : <EOL> region = kwargs . pop ( '<STR_LIT>' , None ) <EOL> if not region : <EOL> region = RegionInfo ( self , self . DefaultRegionName , <EOL> self . DefaultRegionEndpoint ) <EOL> if '<STR_LIT:host>' not in kwargs or kwargs [ '<STR_LIT:host>' ] is None : <EOL> kwargs [ '<STR_LIT:host>' ] = region . endpoint <EOL> super ( CognitoIdentityConnection , self ) . __init__ ( ** kwargs ) <EOL> self . region = region <EOL> def _required_auth_capability ( self ) : <EOL> return [ '<STR_LIT>' ] <EOL> def create_identity_pool ( self , identity_pool_name , <EOL> allow_unauthenticated_identities , <EOL> supported_login_providers = None , <EOL> developer_provider_name = None , <EOL> open_id_connect_provider_ar_ns = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : identity_pool_name , <EOL> '<STR_LIT>' : allow_unauthenticated_identities , <EOL> } <EOL> if supported_login_providers is not None : <EOL> params [ '<STR_LIT>' ] = supported_login_providers <EOL> if developer_provider_name is not None : <EOL> params [ '<STR_LIT>' ] = developer_provider_name <EOL> if open_id_connect_provider_ar_ns is not None : <EOL> params [ '<STR_LIT>' ] = open_id_connect_provider_ar_ns <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def delete_identity_pool ( self , identity_pool_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : identity_pool_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def describe_identity_pool ( self , identity_pool_id ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : identity_pool_id , } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def get_id ( self , account_id , identity_pool_id , logins = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : account_id , <EOL> '<STR_LIT>' : identity_pool_id , <EOL> } <EOL> if logins is not None : <EOL> params [ '<STR_LIT>' ] = logins <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def get_open_id_token ( self , identity_id , logins = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : identity_id , } <EOL> if logins is not None : <EOL> params [ '<STR_LIT>' ] = logins <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def get_open_id_token_for_developer_identity ( self , identity_pool_id , <EOL> logins , identity_id = None , <EOL> token_duration = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : identity_pool_id , <EOL> '<STR_LIT>' : logins , <EOL> } <EOL> if identity_id is not None : <EOL> params [ '<STR_LIT>' ] = identity_id <EOL> if token_duration is not None : <EOL> params [ '<STR_LIT>' ] = token_duration <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def list_identities ( self , identity_pool_id , max_results , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : identity_pool_id , <EOL> '<STR_LIT>' : max_results , <EOL> } <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def list_identity_pools ( self , max_results , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : max_results , } <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def lookup_developer_identity ( self , identity_pool_id , identity_id = None , <EOL> developer_user_identifier = None , <EOL> max_results = None , next_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT>' : identity_pool_id , } <EOL> if identity_id is not None : <EOL> params [ '<STR_LIT>' ] = identity_id <EOL> if developer_user_identifier is not None : <EOL> params [ '<STR_LIT>' ] = developer_user_identifier <EOL> if max_results is not None : <EOL> params [ '<STR_LIT>' ] = max_results <EOL> if next_token is not None : <EOL> params [ '<STR_LIT>' ] = next_token <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def merge_developer_identities ( self , source_user_identifier , <EOL> destination_user_identifier , <EOL> developer_provider_name , identity_pool_id ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : source_user_identifier , <EOL> '<STR_LIT>' : destination_user_identifier , <EOL> '<STR_LIT>' : developer_provider_name , <EOL> '<STR_LIT>' : identity_pool_id , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def unlink_developer_identity ( self , identity_id , identity_pool_id , <EOL> developer_provider_name , <EOL> developer_user_identifier ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : identity_id , <EOL> '<STR_LIT>' : identity_pool_id , <EOL> '<STR_LIT>' : developer_provider_name , <EOL> '<STR_LIT>' : developer_user_identifier , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def unlink_identity ( self , identity_id , logins , logins_to_remove ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : identity_id , <EOL> '<STR_LIT>' : logins , <EOL> '<STR_LIT>' : logins_to_remove , <EOL> } <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def update_identity_pool ( self , identity_pool_id , identity_pool_name , <EOL> allow_unauthenticated_identities , <EOL> supported_login_providers = None , <EOL> developer_provider_name = None , <EOL> open_id_connect_provider_ar_ns = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : identity_pool_id , <EOL> '<STR_LIT>' : identity_pool_name , <EOL> '<STR_LIT>' : allow_unauthenticated_identities , <EOL> } <EOL> if supported_login_providers is not None : <EOL> params [ '<STR_LIT>' ] = supported_login_providers <EOL> if developer_provider_name is not None : <EOL> params [ '<STR_LIT>' ] = developer_provider_name <EOL> if open_id_connect_provider_ar_ns is not None : <EOL> params [ '<STR_LIT>' ] = open_id_connect_provider_ar_ns <EOL> return self . make_request ( action = '<STR_LIT>' , <EOL> body = json . dumps ( params ) ) <EOL> def make_request ( self , action , body ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' % ( self . TargetPrefix , action ) , <EOL> '<STR_LIT>' : self . region . endpoint , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( body ) ) , <EOL> } <EOL> http_request = self . build_base_http_request ( <EOL> method = '<STR_LIT:POST>' , path = '<STR_LIT:/>' , auth_path = '<STR_LIT:/>' , params = { } , <EOL> headers = headers , data = body ) <EOL> response = self . _mexe ( http_request , sender = None , <EOL> override_num_retries = <NUM_LIT:10> ) <EOL> response_body = response . read ( ) . decode ( '<STR_LIT:utf-8>' ) <EOL> boto . log . debug ( response_body ) <EOL> if response . status == <NUM_LIT:200> : <EOL> if response_body : <EOL> return json . loads ( response_body ) <EOL> else : <EOL> json_body = json . loads ( response_body ) <EOL> fault_name = json_body . get ( '<STR_LIT>' , None ) <EOL> exception_class = self . _faults . get ( fault_name , self . ResponseError ) <EOL> raise exception_class ( response . status , response . reason , <EOL> body = json_body ) </s>
