<s> from __future__ import unicode_literals <EOL> import json <EOL> import requests <EOL> import six <EOL> from datetime import datetime <EOL> from six . moves . urllib . parse import parse_qs <EOL> from xml . etree . ElementTree import tostring , SubElement , Element <EOL> from . exceptions import ( <EOL> XeroBadRequest , XeroExceptionUnknown , XeroForbidden , XeroInternalError , <EOL> XeroNotAvailable , XeroNotFound , XeroNotImplemented , XeroRateLimitExceeded , <EOL> XeroUnauthorized <EOL> ) <EOL> from . utils import singular , isplural , json_load_object_hook <EOL> class BaseManager ( object ) : <EOL> DECORATED_METHODS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:all>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> DATETIME_FIELDS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ) <EOL> DATE_FIELDS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> BOOLEAN_FIELDS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> DECIMAL_FIELDS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> INTEGER_FIELDS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> NO_SEND_FIELDS = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> OPERATOR_MAPPINGS = { <EOL> '<STR_LIT>' : '<STR_LIT:>>' , <EOL> '<STR_LIT>' : '<STR_LIT:<>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> def __init__ ( self ) : <EOL> pass <EOL> def dict_to_xml ( self , root_elm , data ) : <EOL> for key in data . keys ( ) : <EOL> if key in self . NO_SEND_FIELDS : <EOL> continue <EOL> sub_data = data [ key ] <EOL> elm = SubElement ( root_elm , key ) <EOL> if isinstance ( sub_data , dict ) : <EOL> self . dict_to_xml ( elm , sub_data ) <EOL> elif isinstance ( sub_data , list ) or isinstance ( sub_data , tuple ) : <EOL> if isplural ( key ) : <EOL> for d in sub_data : <EOL> self . dict_to_xml ( SubElement ( elm , singular ( key ) ) , d ) <EOL> else : <EOL> for d in sub_data : <EOL> self . dict_to_xml ( elm , d ) <EOL> else : <EOL> if key in self . BOOLEAN_FIELDS : <EOL> val = '<STR_LIT:true>' if sub_data else '<STR_LIT:false>' <EOL> elif key in self . DATE_FIELDS : <EOL> val = sub_data . strftime ( '<STR_LIT>' ) <EOL> else : <EOL> val = six . text_type ( sub_data ) <EOL> elm . text = val <EOL> return root_elm <EOL> def _prepare_data_for_save ( self , data ) : <EOL> if isinstance ( data , list ) or isinstance ( data , tuple ) : <EOL> root_elm = Element ( self . name ) <EOL> for d in data : <EOL> sub_elm = SubElement ( root_elm , self . singular ) <EOL> self . dict_to_xml ( sub_elm , d ) <EOL> else : <EOL> root_elm = self . dict_to_xml ( Element ( self . singular ) , data ) <EOL> return six . u ( tostring ( root_elm ) ) <EOL> def _parse_api_response ( self , response , resource_name ) : <EOL> data = json . loads ( response . text , object_hook = json_load_object_hook ) <EOL> assert data [ '<STR_LIT>' ] == '<STR_LIT:OK>' , "<STR_LIT>" % data [ '<STR_LIT>' ] <EOL> try : <EOL> return data [ resource_name ] <EOL> except KeyError : <EOL> return data <EOL> def _get_data ( self , func ) : <EOL> """<STR_LIT>""" <EOL> def wrapper ( * args , ** kwargs ) : <EOL> timeout = kwargs . pop ( '<STR_LIT>' , None ) <EOL> uri , params , method , body , headers , singleobject = func ( * args , ** kwargs ) <EOL> cert = getattr ( self . credentials , '<STR_LIT>' , None ) <EOL> if headers is None : <EOL> headers = { } <EOL> if '<STR_LIT>' not in headers : <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT:application/json>' <EOL> headers [ '<STR_LIT>' ] = self . user_agent <EOL> response = getattr ( requests , method ) ( <EOL> uri , data = body , headers = headers , auth = self . credentials . oauth , <EOL> params = params , cert = cert , timeout = timeout ) <EOL> if response . status_code == <NUM_LIT:200> : <EOL> if not response . headers [ '<STR_LIT>' ] . startswith ( '<STR_LIT:application/json>' ) : <EOL> return response . content <EOL> return self . _parse_api_response ( response , self . name ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise XeroBadRequest ( response ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise XeroUnauthorized ( response ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise XeroForbidden ( response ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise XeroNotFound ( response ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise XeroInternalError ( response ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> raise XeroNotImplemented ( response ) <EOL> elif response . status_code == <NUM_LIT> : <EOL> payload = parse_qs ( response . text ) <EOL> if payload : <EOL> raise XeroRateLimitExceeded ( response , payload ) <EOL> else : <EOL> raise XeroNotAvailable ( response ) <EOL> else : <EOL> raise XeroExceptionUnknown ( response ) <EOL> return wrapper <EOL> def _get ( self , id , headers = None , params = None ) : <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name , id ] ) <EOL> uri_params = self . extra_params . copy ( ) <EOL> uri_params . update ( params if params else { } ) <EOL> return uri , uri_params , '<STR_LIT>' , None , headers , True <EOL> def _get_attachments ( self , id ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name , id , '<STR_LIT>' ] ) + '<STR_LIT:/>' <EOL> return uri , { } , '<STR_LIT>' , None , None , False <EOL> def _get_attachment_data ( self , id , filename ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name , id , '<STR_LIT>' , filename ] ) <EOL> return uri , { } , '<STR_LIT>' , None , None , False <EOL> def get_attachment ( self , id , filename , file ) : <EOL> """<STR_LIT>""" <EOL> data = self . get_attachment_data ( id , filename ) <EOL> file . write ( data ) <EOL> return len ( data ) <EOL> def save_or_put ( self , data , method = '<STR_LIT>' , headers = None , summarize_errors = True ) : <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name ] ) <EOL> body = { '<STR_LIT>' : self . _prepare_data_for_save ( data ) } <EOL> params = self . extra_params . copy ( ) <EOL> if not summarize_errors : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT:false>' <EOL> return uri , params , method , body , headers , False <EOL> def _save ( self , data ) : <EOL> return self . save_or_put ( data , method = '<STR_LIT>' ) <EOL> def _put ( self , data , summarize_errors = True ) : <EOL> return self . save_or_put ( data , method = '<STR_LIT>' , summarize_errors = summarize_errors ) <EOL> def _put_attachment_data ( self , id , filename , data , content_type , include_online = False ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name , id , '<STR_LIT>' , filename ] ) <EOL> params = { '<STR_LIT>' : '<STR_LIT:true>' } if include_online else { } <EOL> headers = { '<STR_LIT:Content-Type>' : content_type , '<STR_LIT>' : len ( data ) } <EOL> return uri , params , '<STR_LIT>' , data , headers , False <EOL> def put_attachment ( self , id , filename , file , content_type , include_online = False ) : <EOL> """<STR_LIT>""" <EOL> self . put_attachment_data ( id , filename , file . read ( ) , content_type , <EOL> include_online = include_online ) <EOL> def prepare_filtering_date ( self , val ) : <EOL> if isinstance ( val , datetime ) : <EOL> val = val . strftime ( '<STR_LIT>' ) <EOL> else : <EOL> val = '<STR_LIT>' % val <EOL> return { '<STR_LIT>' : val } <EOL> def _filter ( self , ** kwargs ) : <EOL> params = self . extra_params . copy ( ) <EOL> headers = None <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name ] ) <EOL> if kwargs : <EOL> if '<STR_LIT>' in kwargs : <EOL> val = kwargs [ '<STR_LIT>' ] <EOL> headers = self . prepare_filtering_date ( val ) <EOL> del kwargs [ '<STR_LIT>' ] <EOL> def get_filter_params ( key , value ) : <EOL> last_key = key . split ( '<STR_LIT:_>' ) [ - <NUM_LIT:1> ] <EOL> if last_key . upper ( ) . endswith ( '<STR_LIT>' ) : <EOL> return '<STR_LIT>' % six . text_type ( value ) <EOL> if key in self . BOOLEAN_FIELDS : <EOL> return '<STR_LIT:true>' if value else '<STR_LIT:false>' <EOL> elif key in self . DATE_FIELDS : <EOL> return '<STR_LIT>' % ( value . year , value . month , value . day ) <EOL> elif key in self . DATETIME_FIELDS : <EOL> return value . isoformat ( ) <EOL> else : <EOL> return '<STR_LIT>' % six . text_type ( value ) <EOL> def generate_param ( key , value ) : <EOL> parts = key . split ( "<STR_LIT>" ) <EOL> field = key . replace ( '<STR_LIT:_>' , '<STR_LIT:.>' ) <EOL> fmt = '<STR_LIT>' <EOL> if len ( parts ) == <NUM_LIT:2> : <EOL> if parts [ <NUM_LIT:1> ] in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> field = parts [ <NUM_LIT:0> ] <EOL> fmt = '<STR_LIT>' . join ( [ '<STR_LIT>' , parts [ <NUM_LIT:1> ] , '<STR_LIT>' ] ) <EOL> elif parts [ <NUM_LIT:1> ] in self . OPERATOR_MAPPINGS : <EOL> field = parts [ <NUM_LIT:0> ] <EOL> key = field <EOL> fmt = '<STR_LIT:%s>' + self . OPERATOR_MAPPINGS [ parts [ <NUM_LIT:1> ] ] + '<STR_LIT:%s>' <EOL> elif parts [ <NUM_LIT:1> ] in [ "<STR_LIT>" ] : <EOL> sign = '<STR_LIT:=>' if value else '<STR_LIT:!>' <EOL> return '<STR_LIT>' % ( parts [ <NUM_LIT:0> ] , sign ) <EOL> field = field . replace ( '<STR_LIT:_>' , '<STR_LIT:.>' ) <EOL> return fmt % ( <EOL> field , <EOL> get_filter_params ( key , value ) <EOL> ) <EOL> KNOWN_PARAMETERS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for param in KNOWN_PARAMETERS : <EOL> if param in kwargs : <EOL> params [ param ] = kwargs . pop ( param ) <EOL> filter_params = [ ] <EOL> if '<STR_LIT>' in kwargs : <EOL> raw = kwargs . pop ( '<STR_LIT>' ) <EOL> filter_params . append ( raw ) <EOL> sortedkwargs = sorted ( six . iteritems ( kwargs ) , <EOL> key = lambda item : - <NUM_LIT:1> if '<STR_LIT>' in item [ <NUM_LIT:0> ] else <NUM_LIT:0> ) <EOL> for key , value in sortedkwargs : <EOL> filter_params . append ( generate_param ( key , value ) ) <EOL> if filter_params : <EOL> params [ '<STR_LIT>' ] = '<STR_LIT>' . join ( filter_params ) <EOL> return uri , params , '<STR_LIT>' , None , headers , False <EOL> def _all ( self ) : <EOL> uri = '<STR_LIT:/>' . join ( [ self . base_url , self . name ] ) <EOL> return uri , { } , '<STR_LIT>' , None , None , False </s>
