<s> from biicode . client . rest . rest_api import RestApiClient <EOL> from biicode . common . utils . serializer import Serializer , ListDeserializer <EOL> from biicode . common . exception import BiiServiceException <EOL> from biicode . common . model . symbolic . block_version_table import BlockVersionTable <EOL> from biicode . common . find . finder_result import FinderResult <EOL> from biicode . common . utils . bii_logging import logger <EOL> from biicode . common . model . symbolic . block_version import BlockVersion <EOL> from biicode . common . model . renames import Renames <EOL> from biicode . common . model . symbolic . reference import ReferencedResources <EOL> from biicode . common . rest . rest_return_mapping import getExceptionFromHttpError <EOL> from biicode . common . api . biiapi import BiiAPI <EOL> from biicode . common . settings . osinfo import OSInfo <EOL> from biicode . common . model . server_info import ServerInfo <EOL> from biicode . common . model . cells import CellDeserializer <EOL> from biicode . common . model . content import ContentDeserializer <EOL> from biicode . common . model . brl . block_cell_name import BlockCellName <EOL> from biicode . common . diffmerge . changes import ChangesDeserializer <EOL> from biicode . common . model . brl . cell_name import CellName <EOL> from biicode . common . model . resource import ResourceDeserializer <EOL> from biicode . common . model . block_delta import BlockDelta <EOL> from biicode . common . model . block_info import BlockInfo <EOL> from biicode . client . rest . rest_api import JWTAuth <EOL> from requests . auth import HTTPBasicAuth <EOL> from biicode . common . utils . bson_encoding import decode_bson , encode_bson <EOL> from biicode . common . api . ui import BiiResponse <EOL> class BiiRestApiClient ( RestApiClient , BiiAPI ) : <EOL> '''<STR_LIT>''' <EOL> version = "<STR_LIT>" <EOL> authorized_functions = { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:GET>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT:GET>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:GET>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:POST>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:GET>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:GET>" } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : "<STR_LIT:GET>" } , <EOL> } <EOL> def __init__ ( self , base_url ) : <EOL> self . base_url = base_url <EOL> self . token = None <EOL> self . custom_headers = { } <EOL> logger . debug ( "<STR_LIT>" % self . base_url ) <EOL> super ( BiiRestApiClient , self ) . __init__ ( <EOL> self . base_url + "<STR_LIT:/>" + BiiRestApiClient . version , <EOL> self . authorized_functions ) <EOL> def get_published_resources ( self , references ) : <EOL> serialized_data = Serializer ( ) . build ( ( "<STR_LIT:data>" , references ) ) <EOL> return self . bson_jwt_call ( '<STR_LIT>' , data = serialized_data , <EOL> deserializer = ReferencedResources ) <EOL> def publish ( self , publish_request ) : <EOL> data = Serializer ( ) . build ( ( "<STR_LIT:data>" , publish_request ) ) <EOL> return self . bson_jwt_call ( '<STR_LIT>' , data = data , deserializer = BlockVersion ) <EOL> def get_dep_table ( self , block_version ) : <EOL> block , time , _ = block_version <EOL> owner_name = block . owner <EOL> block_name = block . block_name <EOL> branch_name = block . branch <EOL> params = { "<STR_LIT>" : owner_name , "<STR_LIT>" : block_name , <EOL> "<STR_LIT>" : branch_name , "<STR_LIT:version>" : time } <EOL> deserializer = BlockVersionTable <EOL> return self . bson_jwt_call ( '<STR_LIT>' , url_params = params , deserializer = deserializer ) <EOL> def get_cells_snapshot ( self , block_version ) : <EOL> data = Serializer ( ) . build ( ( "<STR_LIT:data>" , block_version ) ) <EOL> return self . bson_jwt_call ( '<STR_LIT>' , data = data , <EOL> deserializer = ListDeserializer ( CellName ) ) <EOL> def find ( self , finder_request , response ) : <EOL> data = Serializer ( ) . build ( ( "<STR_LIT:data>" , finder_request ) ) <EOL> return self . bson_jwt_call ( '<STR_LIT>' , data = data , deserializer = FinderResult , response = response ) <EOL> def diff ( self , base , other ) : <EOL> data = Serializer ( ) . build ( ( "<STR_LIT>" , base ) , <EOL> ( "<STR_LIT>" , other ) ) <EOL> values_deserializer = ResourceDeserializer ( CellDeserializer ( BlockCellName ) , <EOL> ContentDeserializer ( BlockCellName ) ) <EOL> deserializer = ChangesDeserializer ( CellName , values_deserializer ) <EOL> return self . bson_jwt_call ( '<STR_LIT>' , data = data , deserializer = deserializer ) <EOL> def get_renames ( self , brl_block , t1 , t2 ) : <EOL> data = Serializer ( ) . build ( ( "<STR_LIT>" , brl_block ) , <EOL> ( "<STR_LIT>" , t1 ) , <EOL> ( "<STR_LIT>" , t2 ) ) <EOL> return self . bson_jwt_call ( '<STR_LIT>' , data = data , deserializer = Renames ) <EOL> def get_block_info ( self , brl_block ) : <EOL> owner_name = brl_block . owner <EOL> block_name = brl_block . block_name <EOL> branch_name = brl_block . branch <EOL> url_params = { "<STR_LIT>" : owner_name , <EOL> "<STR_LIT>" : block_name , <EOL> "<STR_LIT>" : branch_name } <EOL> return self . bson_jwt_call ( '<STR_LIT>' , url_params = url_params , deserializer = BlockInfo ) <EOL> def get_version_delta_info ( self , block_version ) : <EOL> """<STR_LIT>""" <EOL> brl , time , _ = block_version <EOL> url_params = { "<STR_LIT>" : brl . owner , <EOL> "<STR_LIT>" : brl . block_name , <EOL> "<STR_LIT>" : brl . branch , <EOL> "<STR_LIT:version>" : time <EOL> } <EOL> return self . bson_jwt_call ( '<STR_LIT>' , <EOL> url_params = url_params , deserializer = BlockDelta ) <EOL> def get_version_by_tag ( self , brl_block , version_tag ) : <EOL> """<STR_LIT>""" <EOL> assert version_tag is not None <EOL> url_params = { "<STR_LIT>" : brl_block . owner , <EOL> "<STR_LIT>" : brl_block . block_name , <EOL> "<STR_LIT>" : brl_block . branch , <EOL> "<STR_LIT>" : version_tag <EOL> } <EOL> return self . bson_jwt_call ( '<STR_LIT>' , <EOL> url_params = url_params , deserializer = BlockVersion ) <EOL> def get_server_info ( self ) : <EOL> """<STR_LIT>""" <EOL> os_info = OSInfo . capture ( ) <EOL> from biicode . common import __version__ <EOL> data = ( os_info , str ( __version__ ) ) <EOL> serialized_data = Serializer ( ) . build ( ( "<STR_LIT:data>" , data ) ) <EOL> info = self . bson_jwt_call ( '<STR_LIT>' , data = serialized_data , <EOL> deserializer = ServerInfo , timeout = <NUM_LIT:1> ) <EOL> return info <EOL> def require_auth ( self ) : <EOL> info = self . bson_jwt_call ( '<STR_LIT>' ) <EOL> return info <EOL> def authenticate ( self , user , password ) : <EOL> '''<STR_LIT>''' <EOL> token = self . basic_auth_call ( user , password , "<STR_LIT>" ) <EOL> return token <EOL> def bson_jwt_call ( self , function_name , deserializer = None , url_params = { } , data = None , <EOL> headers = None , response = None , timeout = None ) : <EOL> logger . debug ( "<STR_LIT>" % str ( function_name ) ) <EOL> auth = JWTAuth ( self . token ) if self . token else None <EOL> headers = headers or { } <EOL> headers . update ( self . custom_headers ) <EOL> headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT>' <EOL> if data is not None : <EOL> data = str ( encode_bson ( data ) ) <EOL> return self . call ( function_name , url_params = url_params , data = data , headers = headers , <EOL> auth = auth , deserializer = deserializer , response = response , timeout = timeout ) <EOL> def basic_auth_call ( self , user , password , function_name , url_params = { } , <EOL> data = None , headers = None , deserializer = None ) : <EOL> auth = HTTPBasicAuth ( user , password ) <EOL> return self . call ( function_name , url_params = url_params , <EOL> data = data , headers = headers , auth = auth , <EOL> deserializer = deserializer ) <EOL> def call ( self , * args , ** kwargs ) : <EOL> deserializer = kwargs . pop ( "<STR_LIT>" , None ) <EOL> response = kwargs . pop ( "<STR_LIT>" , None ) <EOL> ret = super ( BiiRestApiClient , self ) . call ( * args , ** kwargs ) <EOL> return BiiRestApiClient . deserialize_return ( ret , deserializer , response ) <EOL> @ staticmethod <EOL> def decode_return_content ( res , response = None ) : <EOL> if '<STR_LIT>' in res . headers and res . headers [ '<STR_LIT>' ] == "<STR_LIT>" : <EOL> tmp = decode_bson ( res . content ) <EOL> if response is not None : <EOL> response_server = BiiResponse . deserialize ( tmp [ "<STR_LIT:info>" ] ) <EOL> response_server . biiout ( response ) <EOL> return tmp [ "<STR_LIT>" ] <EOL> else : <EOL> return res . content <EOL> @ staticmethod <EOL> def deserialize_return ( res , deserializer = None , response = None ) : <EOL> '''<STR_LIT>''' <EOL> exc_kls = getExceptionFromHttpError ( res . status_code ) <EOL> logger . debug ( "<STR_LIT>" % str ( exc_kls ) ) <EOL> logger . debug ( "<STR_LIT>" % str ( res . headers . get ( '<STR_LIT>' , "<STR_LIT>" ) ) ) <EOL> data = BiiRestApiClient . decode_return_content ( res , response ) <EOL> if exc_kls is None : <EOL> if deserializer is not None : <EOL> try : <EOL> return deserializer . deserialize ( data ) <EOL> except KeyError : <EOL> raise BiiServiceException ( "<STR_LIT>" ) <EOL> else : <EOL> return data <EOL> else : <EOL> if '<STR_LIT>' in res . headers and "<STR_LIT>" in res . headers [ '<STR_LIT>' ] : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> raise exc_kls ( data ) </s>
