<s> import datetime <EOL> import functools <EOL> from elasticsearch_dsl import Q <EOL> from pyramid . view import view_config <EOL> from pyramid_rpc . xmlrpc import exception_view as _exception_view , xmlrpc_method <EOL> from sqlalchemy import func , select <EOL> from sqlalchemy . orm . exc import NoResultFound <EOL> from warehouse . accounts . models import User <EOL> from warehouse . classifiers . models import Classifier <EOL> from warehouse . packaging . interfaces import IDownloadStatService <EOL> from warehouse . packaging . models import ( <EOL> Role , Project , Release , File , JournalEntry , release_classifiers , <EOL> ) <EOL> pypi_xmlrpc = functools . partial ( <EOL> xmlrpc_method , <EOL> endpoint = "<STR_LIT>" , <EOL> require_csrf = False , <EOL> require_methods = [ "<STR_LIT:POST>" ] , <EOL> ) <EOL> @ view_config ( route_name = "<STR_LIT>" , context = Exception , renderer = "<STR_LIT>" ) <EOL> def exception_view ( exc , request ) : <EOL> return _exception_view ( exc , request ) <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def search ( request , spec , operator = "<STR_LIT>" ) : <EOL> if operator not in { "<STR_LIT>" , "<STR_LIT>" } : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> spec = { <EOL> k : [ v ] if isinstance ( v , str ) else v <EOL> for k , v in spec . items ( ) <EOL> if v and k in { <EOL> "<STR_LIT:name>" , "<STR_LIT:version>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT:description>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> } <EOL> } <EOL> queries = [ ] <EOL> for field , value in sorted ( spec . items ( ) ) : <EOL> q = None <EOL> for item in value : <EOL> if q is None : <EOL> q = Q ( "<STR_LIT>" , ** { field : item } ) <EOL> else : <EOL> q |= Q ( "<STR_LIT>" , ** { field : item } ) <EOL> queries . append ( q ) <EOL> if operator == "<STR_LIT>" : <EOL> query = request . es . query ( "<STR_LIT:bool>" , must = queries ) <EOL> else : <EOL> query = request . es . query ( "<STR_LIT:bool>" , should = queries ) <EOL> results = query [ : <NUM_LIT:1000> ] . execute ( ) <EOL> return [ <EOL> { "<STR_LIT:name>" : r . name , "<STR_LIT>" : r . summary , "<STR_LIT:version>" : v } <EOL> for r in results <EOL> for v in r . version <EOL> if v in spec . get ( "<STR_LIT:version>" , [ v ] ) <EOL> ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def list_packages ( request ) : <EOL> names = request . db . query ( Project . name ) . order_by ( Project . name ) . all ( ) <EOL> return [ n [ <NUM_LIT:0> ] for n in names ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def list_packages_with_serial ( request ) : <EOL> serials = ( <EOL> request . db . query ( JournalEntry . name , func . max ( JournalEntry . id ) ) <EOL> . join ( Project , JournalEntry . name == Project . name ) <EOL> . group_by ( JournalEntry . name ) <EOL> ) <EOL> return dict ( ( serial [ <NUM_LIT:0> ] , serial [ <NUM_LIT:1> ] ) for serial in serials ) <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def package_hosting_mode ( request , package_name ) : <EOL> try : <EOL> project = ( <EOL> request . db . query ( Project ) <EOL> . filter ( Project . normalized_name == <EOL> func . normalize_pep426_name ( package_name ) ) <EOL> . one ( ) <EOL> ) <EOL> except NoResultFound : <EOL> return None <EOL> else : <EOL> return project . hosting_mode <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def user_packages ( request , username ) : <EOL> roles = ( <EOL> request . db . query ( Role ) <EOL> . join ( User , Project ) <EOL> . filter ( User . username == username ) <EOL> . order_by ( Role . role_name . desc ( ) , Project . name ) <EOL> . all ( ) <EOL> ) <EOL> return [ ( r . role_name , r . project . name ) for r in roles ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def top_packages ( request , num = None ) : <EOL> fdownloads = func . sum ( File . downloads ) . label ( "<STR_LIT>" ) <EOL> downloads = ( <EOL> request . db . query ( File . name , fdownloads ) <EOL> . group_by ( File . name ) <EOL> . order_by ( fdownloads . desc ( ) ) <EOL> ) <EOL> if num is not None : <EOL> downloads = downloads . limit ( num ) <EOL> return [ ( d [ <NUM_LIT:0> ] , d [ <NUM_LIT:1> ] ) for d in downloads . all ( ) ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def package_releases ( request , package_name , show_hidden = False ) : <EOL> versions = ( <EOL> request . db . query ( Release . version ) <EOL> . join ( Project ) <EOL> . filter ( Project . normalized_name == <EOL> func . normalize_pep426_name ( package_name ) ) <EOL> . order_by ( Release . _pypi_ordering ) <EOL> . all ( ) <EOL> ) <EOL> return [ v [ <NUM_LIT:0> ] for v in versions ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def release_data ( request , package_name , version ) : <EOL> try : <EOL> release = ( <EOL> request . db . query ( Release ) <EOL> . join ( Project ) <EOL> . filter ( ( Project . normalized_name == <EOL> func . normalize_pep426_name ( package_name ) ) & <EOL> ( Release . version == version ) ) <EOL> . one ( ) <EOL> ) <EOL> except NoResultFound : <EOL> return { } <EOL> stats_svc = request . find_service ( IDownloadStatService ) <EOL> return { <EOL> "<STR_LIT:name>" : release . project . name , <EOL> "<STR_LIT:version>" : release . version , <EOL> "<STR_LIT>" : release . project . stable_version , <EOL> "<STR_LIT>" : release . project . bugtrack_url , <EOL> "<STR_LIT>" : request . route_url ( <EOL> "<STR_LIT>" , <EOL> name = release . project . name , <EOL> ) , <EOL> "<STR_LIT>" : request . route_url ( <EOL> "<STR_LIT>" , <EOL> name = release . project . name , <EOL> version = release . version , <EOL> ) , <EOL> "<STR_LIT>" : release . project . documentation_url , <EOL> "<STR_LIT>" : release . home_page , <EOL> "<STR_LIT>" : release . download_url , <EOL> "<STR_LIT>" : list ( release . project_urls ) , <EOL> "<STR_LIT>" : release . author , <EOL> "<STR_LIT>" : release . author_email , <EOL> "<STR_LIT>" : release . maintainer , <EOL> "<STR_LIT>" : release . maintainer_email , <EOL> "<STR_LIT>" : release . summary , <EOL> "<STR_LIT:description>" : release . description , <EOL> "<STR_LIT>" : release . license , <EOL> "<STR_LIT>" : release . keywords , <EOL> "<STR_LIT>" : release . platform , <EOL> "<STR_LIT>" : list ( release . classifiers ) , <EOL> "<STR_LIT>" : list ( release . requires ) , <EOL> "<STR_LIT>" : list ( release . requires_dist ) , <EOL> "<STR_LIT>" : list ( release . provides ) , <EOL> "<STR_LIT>" : list ( release . provides_dist ) , <EOL> "<STR_LIT>" : list ( release . obsoletes ) , <EOL> "<STR_LIT>" : list ( release . obsoletes_dist ) , <EOL> "<STR_LIT>" : release . requires_python , <EOL> "<STR_LIT>" : list ( release . requires_external ) , <EOL> "<STR_LIT>" : release . _pypi_ordering , <EOL> "<STR_LIT>" : release . _pypi_hidden , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : stats_svc . get_daily_stats ( release . project . name ) , <EOL> "<STR_LIT>" : stats_svc . get_weekly_stats ( release . project . name ) , <EOL> "<STR_LIT>" : stats_svc . get_monthly_stats ( release . project . name ) , <EOL> } , <EOL> } <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def release_urls ( request , package_name , version ) : <EOL> files = ( <EOL> request . db . query ( File ) <EOL> . join ( Release , Project ) <EOL> . filter ( ( Project . normalized_name == <EOL> func . normalize_pep426_name ( package_name ) ) & <EOL> ( Release . version == version ) ) <EOL> . all ( ) <EOL> ) <EOL> return [ <EOL> { <EOL> "<STR_LIT:filename>" : f . filename , <EOL> "<STR_LIT>" : f . packagetype , <EOL> "<STR_LIT>" : f . python_version , <EOL> "<STR_LIT:size>" : f . size , <EOL> "<STR_LIT>" : f . md5_digest , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : f . md5_digest , <EOL> "<STR_LIT>" : f . sha256_digest , <EOL> } , <EOL> "<STR_LIT>" : f . has_signature , <EOL> "<STR_LIT>" : f . upload_time , <EOL> "<STR_LIT>" : f . comment_text , <EOL> "<STR_LIT>" : f . downloads , <EOL> "<STR_LIT:url>" : request . route_url ( "<STR_LIT>" , path = f . path ) , <EOL> } <EOL> for f in files <EOL> ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def package_roles ( request , package_name ) : <EOL> roles = ( <EOL> request . db . query ( Role ) <EOL> . join ( User , Project ) <EOL> . filter ( Project . normalized_name == <EOL> func . normalize_pep426_name ( package_name ) ) <EOL> . order_by ( Role . role_name . desc ( ) , User . username ) <EOL> . all ( ) <EOL> ) <EOL> return [ ( r . role_name , r . user . username ) for r in roles ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def changelog_last_serial ( request ) : <EOL> return request . db . query ( func . max ( JournalEntry . id ) ) . scalar ( ) <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def changelog_since_serial ( request , serial ) : <EOL> entries = ( <EOL> request . db . query ( JournalEntry ) <EOL> . filter ( JournalEntry . id > serial ) <EOL> . order_by ( JournalEntry . id ) <EOL> . all ( ) <EOL> ) <EOL> return [ <EOL> ( <EOL> e . name , <EOL> e . version , <EOL> int ( <EOL> e . submitted_date <EOL> . replace ( tzinfo = datetime . timezone . utc ) <EOL> . timestamp ( ) <EOL> ) , <EOL> e . action , <EOL> e . id , <EOL> ) <EOL> for e in entries <EOL> ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def changelog ( request , since , with_ids = False ) : <EOL> since = datetime . datetime . utcfromtimestamp ( since ) <EOL> entries = ( <EOL> request . db . query ( JournalEntry ) <EOL> . filter ( JournalEntry . submitted_date > since ) <EOL> . order_by ( JournalEntry . id ) <EOL> . all ( ) <EOL> ) <EOL> results = ( <EOL> ( <EOL> e . name , <EOL> e . version , <EOL> int ( <EOL> e . submitted_date <EOL> . replace ( tzinfo = datetime . timezone . utc ) <EOL> . timestamp ( ) <EOL> ) , <EOL> e . action , <EOL> e . id , <EOL> ) <EOL> for e in entries <EOL> ) <EOL> if with_ids : <EOL> return list ( results ) <EOL> else : <EOL> return [ r [ : - <NUM_LIT:1> ] for r in results ] <EOL> @ pypi_xmlrpc ( method = "<STR_LIT>" ) <EOL> def browse ( request , classifiers ) : <EOL> classifiers_q = ( <EOL> request . db . query ( Classifier ) <EOL> . filter ( Classifier . classifier . in_ ( classifiers ) ) <EOL> . subquery ( ) <EOL> ) <EOL> release_classifiers_q = ( <EOL> select ( [ release_classifiers ] ) <EOL> . where ( release_classifiers . c . trove_id == classifiers_q . c . id ) <EOL> . alias ( "<STR_LIT>" ) <EOL> ) <EOL> releases = ( <EOL> request . db . query ( Release . name , Release . version ) <EOL> . join ( release_classifiers_q , <EOL> ( Release . name == release_classifiers_q . c . name ) & <EOL> ( Release . version == release_classifiers_q . c . version ) ) <EOL> . group_by ( Release . name , Release . version ) <EOL> . having ( func . count ( ) == len ( classifiers ) ) <EOL> . order_by ( Release . name , Release . version ) <EOL> . all ( ) <EOL> ) <EOL> return [ ( r . name , r . version ) for r in releases ] </s>
