<s> """<STR_LIT>""" <EOL> from __future__ import print_function , division , absolute_import , unicode_literals <EOL> from future import standard_library <EOL> standard_library . install_aliases ( ) <EOL> from builtins import * <EOL> from six import raise_from <EOL> from collections import namedtuple <EOL> from google . protobuf . descriptor import FieldDescriptor <EOL> import json <EOL> from gmusicapi . exceptions import ( <EOL> CallFailure , ParseException , ValidationException , <EOL> ) <EOL> from gmusicapi . utils import utils <EOL> import requests <EOL> from future . utils import with_metaclass <EOL> log = utils . DynamicClientLogger ( __name__ ) <EOL> _auth_names = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> """<STR_LIT>""" <EOL> AuthTypes = namedtuple ( '<STR_LIT>' , _auth_names ) <EOL> def authtypes ( ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> for name in _auth_names : <EOL> if name not in kwargs : <EOL> kwargs [ name ] = False <EOL> return AuthTypes ( ** kwargs ) <EOL> class BuildRequestMeta ( type ) : <EOL> """<STR_LIT>""" <EOL> def __new__ ( cls , name , bases , dct ) : <EOL> new_cls = super ( BuildRequestMeta , cls ) . __new__ ( cls , name , bases , dct ) <EOL> merge_keys = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> all_keys = ( '<STR_LIT>' , '<STR_LIT:url>' , '<STR_LIT>' , '<STR_LIT:data>' , '<STR_LIT>' , '<STR_LIT>' ) + merge_keys <EOL> config = { } <EOL> dyn = lambda key : '<STR_LIT>' + key <EOL> stat = lambda key : '<STR_LIT>' + key <EOL> has_key = lambda key : hasattr ( new_cls , key ) <EOL> get_key = lambda key : getattr ( new_cls , key ) <EOL> for key in all_keys : <EOL> if not has_key ( dyn ( key ) ) and not has_key ( stat ( key ) ) : <EOL> continue <EOL> if has_key ( dyn ( key ) ) : <EOL> config [ key ] = get_key ( dyn ( key ) ) <EOL> else : <EOL> config [ key ] = get_key ( stat ( key ) ) <EOL> for key in merge_keys : <EOL> if has_key ( dyn ( key ) ) and has_key ( stat ( key ) ) : <EOL> def key_closure ( stat_val = get_key ( stat ( key ) ) , dyn_func = get_key ( dyn ( key ) ) ) : <EOL> def build_key ( * args , ** kwargs ) : <EOL> dyn_val = dyn_func ( * args , ** kwargs ) <EOL> stat_val . update ( dyn_val ) <EOL> return stat_val <EOL> return build_key <EOL> config [ key ] = key_closure ( ) <EOL> def req_closure ( config = config ) : <EOL> def build_request ( cls , * args , ** kwargs ) : <EOL> req_kwargs = { } <EOL> for key , val in config . items ( ) : <EOL> if hasattr ( val , '<STR_LIT>' ) : <EOL> val = val ( * args , ** kwargs ) <EOL> req_kwargs [ key ] = val <EOL> return req_kwargs <EOL> return build_request <EOL> new_cls . build_request = classmethod ( req_closure ( ) ) <EOL> return new_cls <EOL> class Call ( with_metaclass ( BuildRequestMeta , object ) ) : <EOL> """<STR_LIT>""" <EOL> gets_logged = True <EOL> fail_on_non_200 = True <EOL> required_auth = authtypes ( ) <EOL> @ classmethod <EOL> def parse_response ( cls , response ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ classmethod <EOL> def validate ( cls , response , msg ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ classmethod <EOL> def check_success ( cls , response , msg ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> @ classmethod <EOL> def filter_response ( cls , msg ) : <EOL> """<STR_LIT>""" <EOL> return msg <EOL> @ classmethod <EOL> def perform ( cls , session , validate , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> call_name = cls . __name__ <EOL> if cls . gets_logged : <EOL> log . debug ( "<STR_LIT>" , <EOL> call_name , <EOL> [ utils . truncate ( a ) for a in args ] , <EOL> dict ( ( k , utils . truncate ( v ) ) for ( k , v ) in kwargs . items ( ) ) <EOL> ) <EOL> else : <EOL> log . debug ( "<STR_LIT>" , call_name ) <EOL> req_kwargs = cls . build_request ( * args , ** kwargs ) <EOL> response = session . send ( req_kwargs , cls . required_auth ) <EOL> safe_req_kwargs = req_kwargs . copy ( ) <EOL> if safe_req_kwargs . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' , None ) is not None : <EOL> safe_req_kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if cls . fail_on_non_200 : <EOL> try : <EOL> response . raise_for_status ( ) <EOL> except requests . HTTPError as e : <EOL> err_msg = str ( e ) <EOL> if cls . gets_logged : <EOL> err_msg += "<STR_LIT>" % ( safe_req_kwargs ) <EOL> err_msg += "<STR_LIT>" % response . text <EOL> raise CallFailure ( err_msg , call_name ) <EOL> try : <EOL> parsed_response = cls . parse_response ( response ) <EOL> except ParseException : <EOL> err_msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if cls . gets_logged : <EOL> err_msg += "<STR_LIT>" % ( safe_req_kwargs ) <EOL> err_msg += "<STR_LIT>" % response . text <EOL> log . exception ( "<STR_LIT>" , call_name , response . text ) <EOL> else : <EOL> log . exception ( "<STR_LIT>" , call_name ) <EOL> raise CallFailure ( err_msg , call_name ) <EOL> if cls . gets_logged : <EOL> log . debug ( cls . filter_response ( parsed_response ) ) <EOL> try : <EOL> cls . check_success ( response , parsed_response ) <EOL> if validate : <EOL> cls . validate ( response , parsed_response ) <EOL> except CallFailure as e : <EOL> if not cls . gets_logged : <EOL> raise <EOL> err_msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( <EOL> e_message = str ( e ) , <EOL> req_kwargs = safe_req_kwargs , <EOL> content = response . text ) <EOL> raise_from ( CallFailure ( err_msg , e . callname ) , e ) <EOL> except ValidationException as e : <EOL> err_msg = "<STR_LIT>" % call_name <EOL> err_msg += "<STR_LIT>" % e <EOL> if cls . gets_logged : <EOL> raw_response = response . text <EOL> if len ( raw_response ) > <NUM_LIT> : <EOL> raw_response = raw_response [ : <NUM_LIT> ] + '<STR_LIT>' <EOL> err_msg += ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) % ( safe_req_kwargs , raw_response ) <EOL> log . exception ( err_msg ) <EOL> return parsed_response <EOL> @ staticmethod <EOL> def _parse_json ( text ) : <EOL> try : <EOL> return json . loads ( text ) <EOL> except ValueError as e : <EOL> raise_from ( ParseException ( str ( e ) ) , e ) <EOL> @ staticmethod <EOL> def _filter_proto ( msg , make_copy = True ) : <EOL> """<STR_LIT>""" <EOL> filtered = msg <EOL> if make_copy : <EOL> filtered = msg . __class__ ( ) <EOL> filtered . CopyFrom ( msg ) <EOL> fields = filtered . ListFields ( ) <EOL> for field_name , val in ( ( fd . name , val ) for fd , val in fields <EOL> if fd . type == FieldDescriptor . TYPE_BYTES ) : <EOL> setattr ( filtered , field_name , bytes ( "<STR_LIT>" % len ( val ) , '<STR_LIT:utf8>' ) ) <EOL> for field in ( val for fd , val in fields <EOL> if fd . type == FieldDescriptor . TYPE_MESSAGE ) : <EOL> is_repeated = hasattr ( field , '<STR_LIT>' ) <EOL> if not is_repeated : <EOL> Call . _filter_proto ( field , make_copy = False ) <EOL> else : <EOL> for i in range ( len ( field ) ) : <EOL> old_fields = [ f for f in field ] <EOL> del field [ : ] <EOL> field . extend ( [ Call . _filter_proto ( f , make_copy = False ) <EOL> for f in old_fields ] ) <EOL> return filtered </s>
