<s> import logging <EOL> import requests <EOL> from sqlalchemy import sql <EOL> from acoustid import tables as schema <EOL> from acoustid . data . account import ( <EOL> insert_account , lookup_account_id_by_apikey , <EOL> get_account_details_by_mbuser , <EOL> ) <EOL> from acoustid . api import errors <EOL> from acoustid . api . v2 import APIHandler , APIHandlerParams <EOL> from acoustid . handler import Handler , Response <EOL> from acoustid . utils import is_uuid , is_foreignid <EOL> logger = logging . getLogger ( __name__ ) <EOL> class TrackListByMBIDHandlerParams ( APIHandlerParams ) : <EOL> def parse ( self , values , conn ) : <EOL> super ( TrackListByMBIDHandlerParams , self ) . parse ( values , conn ) <EOL> self . disabled = values . get ( '<STR_LIT>' , type = int ) <EOL> self . batch = values . get ( '<STR_LIT>' , type = int ) <EOL> self . mbids = values . getlist ( '<STR_LIT>' ) <EOL> if not self . mbids : <EOL> raise errors . MissingParameterError ( '<STR_LIT>' ) <EOL> if not all ( map ( is_uuid , self . mbids ) ) : <EOL> raise errors . InvalidUUIDError ( '<STR_LIT>' + suffix ) <EOL> class TrackListByMBIDHandler ( APIHandler ) : <EOL> params_class = TrackListByMBIDHandlerParams <EOL> def _handle_internal ( self , params ) : <EOL> response = { } <EOL> condition = schema . track_mbid . c . mbid . in_ ( params . mbids ) <EOL> if not params . disabled : <EOL> condition = sql . and_ ( condition , schema . track_mbid . c . disabled == False ) <EOL> query = sql . select ( [ schema . track_mbid . c . mbid , schema . track_mbid . c . disabled , schema . track . c . gid ] , <EOL> condition , schema . track_mbid . join ( schema . track ) ) <EOL> tracks_map = { } <EOL> for mbid , disabled , track_gid in self . conn . execute ( query ) : <EOL> track = { '<STR_LIT:id>' : track_gid } <EOL> if params . disabled and disabled : <EOL> track [ '<STR_LIT>' ] = disabled <EOL> tracks_map . setdefault ( mbid , [ ] ) . append ( track ) <EOL> if not params . batch : <EOL> response [ '<STR_LIT>' ] = tracks_map . get ( params . mbids [ <NUM_LIT:0> ] , [ ] ) <EOL> else : <EOL> response [ '<STR_LIT>' ] = mbids = [ ] <EOL> for mbid in params . mbids : <EOL> mbids . append ( { <EOL> '<STR_LIT>' : mbid , <EOL> '<STR_LIT>' : tracks_map . get ( mbid , [ ] ) , <EOL> } ) <EOL> return response <EOL> class TrackListByPUIDHandlerParams ( APIHandlerParams ) : <EOL> def parse ( self , values , conn ) : <EOL> super ( TrackListByPUIDHandlerParams , self ) . parse ( values , conn ) <EOL> self . puid = values . get ( '<STR_LIT>' ) <EOL> if not self . puid : <EOL> raise errors . MissingParameterError ( '<STR_LIT>' ) <EOL> if not is_uuid ( self . puid ) : <EOL> raise errors . InvalidUUIDError ( '<STR_LIT>' ) <EOL> class TrackListByPUIDHandler ( APIHandler ) : <EOL> params_class = TrackListByPUIDHandlerParams <EOL> def _handle_internal ( self , params ) : <EOL> response = { } <EOL> response [ '<STR_LIT>' ] = tracks = [ ] <EOL> return response <EOL> class UserCreateMusicBrainzHandlerParams ( APIHandlerParams ) : <EOL> def parse ( self , values , conn ) : <EOL> super ( UserCreateMusicBrainzHandlerParams , self ) . parse ( values , conn ) <EOL> self . access_token = values . get ( '<STR_LIT>' ) <EOL> if not self . access_token : <EOL> raise errors . MissingParameterError ( '<STR_LIT>' ) <EOL> class UserCreateMusicBrainzHandler ( APIHandler ) : <EOL> params_class = UserCreateMusicBrainzHandlerParams <EOL> def _handle_internal ( self , params ) : <EOL> if not self . is_secure : <EOL> raise errors . InsecureRequestError ( ) <EOL> resp = requests . get ( '<STR_LIT>' , <EOL> params = { '<STR_LIT>' : params . access_token } ) <EOL> if resp . status_code != requests . codes . ok : <EOL> raise errors . InvalidMusicBrainzAccessTokenError ( ) <EOL> mbuser = resp . json ( ) [ '<STR_LIT>' ] <EOL> account = get_account_details_by_mbuser ( self . conn , mbuser ) <EOL> if account is not None : <EOL> api_key = account [ '<STR_LIT>' ] <EOL> else : <EOL> id , api_key = insert_account ( self . conn , { <EOL> '<STR_LIT:name>' : mbuser , <EOL> '<STR_LIT>' : mbuser , <EOL> '<STR_LIT>' : self . user_ip , <EOL> } ) <EOL> return { '<STR_LIT:user>' : { '<STR_LIT>' : api_key } } <EOL> class UserCreateAnonymousHandlerParams ( APIHandlerParams ) : <EOL> def parse ( self , values , conn ) : <EOL> super ( UserCreateAnonymousHandlerParams , self ) . parse ( values , conn ) <EOL> self . _parse_client ( values , conn ) <EOL> class UserCreateAnonymousHandler ( APIHandler ) : <EOL> params_class = UserCreateAnonymousHandlerParams <EOL> def _handle_internal ( self , params ) : <EOL> print { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . user_ip , <EOL> '<STR_LIT>' : params . application_id , <EOL> '<STR_LIT>' : params . application_version , <EOL> } <EOL> id , api_key = insert_account ( self . conn , { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . user_ip , <EOL> '<STR_LIT>' : params . application_id , <EOL> '<STR_LIT>' : params . application_version , <EOL> } ) <EOL> return { '<STR_LIT:user>' : { '<STR_LIT>' : api_key } } <EOL> class UserLookupHandlerParams ( APIHandlerParams ) : <EOL> def parse ( self , values , conn ) : <EOL> super ( UserLookupHandlerParams , self ) . parse ( values , conn ) <EOL> account_apikey = values . get ( '<STR_LIT:user>' ) <EOL> if not account_apikey : <EOL> raise errors . MissingParameterError ( '<STR_LIT:user>' ) <EOL> self . account_id = lookup_account_id_by_apikey ( conn , account_apikey ) <EOL> if not self . account_id : <EOL> raise errors . InvalidUserAPIKeyError ( ) <EOL> self . account_apikey = account_apikey <EOL> class UserLookupHandler ( APIHandler ) : <EOL> params_class = UserLookupHandlerParams <EOL> def _handle_internal ( self , params ) : <EOL> return { '<STR_LIT:user>' : { '<STR_LIT>' : params . account_apikey } } </s>
