<s> from distutils . version import LooseVersion <EOL> from itertools import groupby <EOL> import operator <EOL> import six <EOL> from nailgun . db import db <EOL> from nailgun . db . sqlalchemy import models <EOL> from nailgun . objects import DeploymentGraph <EOL> from nailgun . objects import NailgunCollection <EOL> from nailgun . objects import NailgunObject <EOL> from nailgun . objects . serializers . plugin import PluginSerializer <EOL> from nailgun . plugins . adapters import wrap_plugin <EOL> class Plugin ( NailgunObject ) : <EOL> model = models . Plugin <EOL> serializer = PluginSerializer <EOL> @ classmethod <EOL> def create ( cls , data ) : <EOL> deployment_tasks = data . pop ( '<STR_LIT>' , [ ] ) <EOL> new_plugin = super ( Plugin , cls ) . create ( data ) <EOL> DeploymentGraph . create_for_model ( <EOL> { '<STR_LIT>' : deployment_tasks } , new_plugin ) <EOL> plugin_adapter = wrap_plugin ( new_plugin ) <EOL> cls . update ( new_plugin , plugin_adapter . get_metadata ( ) ) <EOL> ClusterPlugins . add_compatible_clusters ( new_plugin ) <EOL> return new_plugin <EOL> @ classmethod <EOL> def get_by_name_version ( cls , name , version ) : <EOL> return db ( ) . query ( cls . model ) . filter_by ( name = name , version = version ) . first ( ) <EOL> class PluginCollection ( NailgunCollection ) : <EOL> single = Plugin <EOL> @ classmethod <EOL> def all_newest ( cls ) : <EOL> """<STR_LIT>""" <EOL> newest_plugins = [ ] <EOL> get_name = operator . attrgetter ( '<STR_LIT:name>' ) <EOL> grouped_by_name = groupby ( sorted ( cls . all ( ) , key = get_name ) , get_name ) <EOL> for name , plugins in grouped_by_name : <EOL> newest_plugin = max ( plugins , key = lambda p : LooseVersion ( p . version ) ) <EOL> newest_plugins . append ( newest_plugin ) <EOL> return newest_plugins <EOL> @ classmethod <EOL> def get_by_uids ( cls , plugin_ids ) : <EOL> """<STR_LIT>""" <EOL> return cls . filter_by_id_list ( cls . all ( ) , plugin_ids ) <EOL> @ classmethod <EOL> def get_by_release ( cls , release ) : <EOL> """<STR_LIT>""" <EOL> release_plugins = set ( ) <EOL> release_os = release . operating_system . lower ( ) <EOL> release_version = release . version <EOL> for plugin in PluginCollection . all ( ) : <EOL> for plugin_release in plugin . releases : <EOL> if ( release_os == plugin_release . get ( '<STR_LIT>' ) and <EOL> release_version == plugin_release . get ( '<STR_LIT:version>' ) ) : <EOL> release_plugins . add ( plugin ) <EOL> return sorted ( release_plugins , key = lambda plugin : plugin . name ) <EOL> class ClusterPlugins ( NailgunObject ) : <EOL> model = models . ClusterPlugins <EOL> @ classmethod <EOL> def is_compatible ( cls , cluster , plugin ) : <EOL> """<STR_LIT>""" <EOL> plugin_adapter = wrap_plugin ( plugin ) <EOL> return plugin_adapter . validate_compatibility ( cluster ) <EOL> @ classmethod <EOL> def get_compatible_plugins ( cls , cluster ) : <EOL> """<STR_LIT>""" <EOL> return list ( six . moves . filter ( <EOL> lambda p : cls . is_compatible ( cluster , p ) , <EOL> PluginCollection . all ( ) ) ) <EOL> @ classmethod <EOL> def add_compatible_plugins ( cls , cluster ) : <EOL> """<STR_LIT>""" <EOL> for plugin in cls . get_compatible_plugins ( cluster ) : <EOL> plugin_attributes = dict ( plugin . attributes_metadata ) <EOL> plugin_attributes . pop ( '<STR_LIT>' , None ) <EOL> cls . create ( { <EOL> '<STR_LIT>' : cluster . id , <EOL> '<STR_LIT>' : plugin . id , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : plugin_attributes <EOL> } ) <EOL> @ classmethod <EOL> def get_compatible_clusters ( cls , plugin ) : <EOL> """<STR_LIT>""" <EOL> return list ( six . moves . filter ( <EOL> lambda c : cls . is_compatible ( c , plugin ) , <EOL> db ( ) . query ( models . Cluster ) ) ) <EOL> @ classmethod <EOL> def add_compatible_clusters ( cls , plugin ) : <EOL> """<STR_LIT>""" <EOL> plugin_attributes = dict ( plugin . attributes_metadata ) <EOL> plugin_attributes . pop ( '<STR_LIT>' , None ) <EOL> for cluster in cls . get_compatible_clusters ( plugin ) : <EOL> cls . create ( { <EOL> '<STR_LIT>' : cluster . id , <EOL> '<STR_LIT>' : plugin . id , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : plugin_attributes <EOL> } ) <EOL> @ classmethod <EOL> def set_attributes ( cls , cluster_id , plugin_id , enabled = None , attrs = None ) : <EOL> """<STR_LIT>""" <EOL> params = { } <EOL> if enabled is not None : <EOL> params [ '<STR_LIT>' ] = enabled <EOL> if attrs is not None : <EOL> params [ '<STR_LIT>' ] = attrs <EOL> db ( ) . query ( cls . model ) . filter_by ( plugin_id = plugin_id , cluster_id = cluster_id ) . update ( params , synchronize_session = '<STR_LIT>' ) <EOL> db ( ) . flush ( ) <EOL> @ classmethod <EOL> def get_connected_plugins_data ( cls , cluster_id ) : <EOL> """<STR_LIT>""" <EOL> return db ( ) . query ( <EOL> models . Plugin . id , <EOL> models . Plugin . name , <EOL> models . Plugin . title , <EOL> models . Plugin . version , <EOL> models . Plugin . is_hotpluggable , <EOL> models . Plugin . attributes_metadata , <EOL> cls . model . enabled , <EOL> cls . model . attributes <EOL> ) . join ( cls . model ) . filter ( cls . model . cluster_id == cluster_id ) . order_by ( models . Plugin . name , models . Plugin . version ) <EOL> @ classmethod <EOL> def get_connected_plugins ( cls , cluster , plugin_ids = None ) : <EOL> """<STR_LIT>""" <EOL> plugins = db ( ) . query ( <EOL> models . Plugin <EOL> ) . join ( cls . model ) . filter ( cls . model . cluster_id == cluster . id ) . order_by ( models . Plugin . name , models . Plugin . version ) <EOL> if plugin_ids : <EOL> plugins = plugins . filter ( cls . model . plugin_id . in_ ( plugin_ids ) ) <EOL> return plugins <EOL> @ classmethod <EOL> def get_connected_clusters ( cls , plugin_id ) : <EOL> """<STR_LIT>""" <EOL> return db ( ) . query ( models . Cluster ) . join ( cls . model ) . filter ( cls . model . plugin_id == plugin_id ) . order_by ( models . Cluster . name ) <EOL> @ classmethod <EOL> def get_enabled ( cls , cluster_id ) : <EOL> """<STR_LIT>""" <EOL> return db ( ) . query ( models . Plugin ) . join ( cls . model ) . filter ( cls . model . cluster_id == cluster_id ) . filter ( cls . model . enabled . is_ ( True ) ) . order_by ( models . Plugin . id ) <EOL> @ classmethod <EOL> def is_plugin_used ( cls , plugin_id ) : <EOL> """<STR_LIT>""" <EOL> q = db ( ) . query ( cls . model ) . filter ( cls . model . plugin_id == plugin_id ) . filter ( cls . model . enabled . is_ ( True ) ) <EOL> return db ( ) . query ( q . exists ( ) ) . scalar ( ) </s>
