<s> import importlib <EOL> import inspect <EOL> import json <EOL> import os <EOL> import sys <EOL> import yaml <EOL> from st2common . exceptions . plugins import IncompatiblePluginException <EOL> from st2common import log as logging <EOL> __all__ = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> LOG = logging . getLogger ( __name__ ) <EOL> PYTHON_EXTENSIONS = ( '<STR_LIT>' ) <EOL> def _register_plugin_path ( plugin_dir_abs_path ) : <EOL> if not os . path . isdir ( plugin_dir_abs_path ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> for x in sys . path : <EOL> if plugin_dir_abs_path in ( x , x + os . sep ) : <EOL> return <EOL> sys . path . append ( plugin_dir_abs_path ) <EOL> def _get_plugin_module ( plugin_file_path ) : <EOL> plugin_module = os . path . basename ( plugin_file_path ) <EOL> if plugin_module . endswith ( PYTHON_EXTENSIONS ) : <EOL> plugin_module = plugin_module [ : plugin_module . rfind ( '<STR_LIT>' ) ] <EOL> else : <EOL> plugin_module = None <EOL> return plugin_module <EOL> def _get_classes_in_module ( module ) : <EOL> return [ kls for name , kls in inspect . getmembers ( module , <EOL> lambda member : inspect . isclass ( member ) and member . __module__ == module . __name__ ) ] <EOL> def _get_plugin_classes ( module_name ) : <EOL> return _get_classes_in_module ( module_name ) <EOL> def _get_plugin_methods ( plugin_klass ) : <EOL> """<STR_LIT>""" <EOL> methods = inspect . getmembers ( plugin_klass , inspect . ismethod ) <EOL> method_names = [ ] <EOL> for name , method in methods : <EOL> method_properties = method . __dict__ <EOL> is_abstract = method_properties . get ( '<STR_LIT>' , False ) <EOL> if is_abstract : <EOL> continue <EOL> method_names . append ( name ) <EOL> return method_names <EOL> def _validate_methods ( plugin_base_class , plugin_klass ) : <EOL> '''<STR_LIT>''' <EOL> expected_methods = plugin_base_class . __abstractmethods__ <EOL> plugin_methods = _get_plugin_methods ( plugin_klass ) <EOL> for method in expected_methods : <EOL> if method not in plugin_methods : <EOL> message = '<STR_LIT>' <EOL> raise IncompatiblePluginException ( message % ( plugin_klass . __name__ , method ) ) <EOL> def _register_plugin ( plugin_base_class , plugin_impl ) : <EOL> _validate_methods ( plugin_base_class , plugin_impl ) <EOL> plugin_base_class . register ( plugin_impl ) <EOL> def register_plugin_class ( base_class , file_path , class_name ) : <EOL> """<STR_LIT>""" <EOL> plugin_dir = os . path . dirname ( os . path . realpath ( file_path ) ) <EOL> _register_plugin_path ( plugin_dir ) <EOL> module_name = _get_plugin_module ( file_path ) <EOL> if module_name is None : <EOL> return None <EOL> module = importlib . import_module ( module_name ) <EOL> klass = getattr ( module , class_name , None ) <EOL> if not klass : <EOL> raise Exception ( '<STR_LIT>' % <EOL> ( file_path , class_name ) ) <EOL> _register_plugin ( base_class , klass ) <EOL> return klass <EOL> def register_plugin ( plugin_base_class , plugin_abs_file_path ) : <EOL> registered_plugins = [ ] <EOL> plugin_dir = os . path . dirname ( os . path . realpath ( plugin_abs_file_path ) ) <EOL> _register_plugin_path ( plugin_dir ) <EOL> module_name = _get_plugin_module ( plugin_abs_file_path ) <EOL> if module_name is None : <EOL> return None <EOL> module = importlib . import_module ( module_name ) <EOL> klasses = _get_plugin_classes ( module ) <EOL> for klass in klasses : <EOL> try : <EOL> _register_plugin ( plugin_base_class , klass ) <EOL> registered_plugins . append ( klass ) <EOL> except Exception as e : <EOL> LOG . exception ( e ) <EOL> LOG . debug ( '<STR_LIT>' , klass ) <EOL> continue <EOL> if len ( registered_plugins ) == <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' % <EOL> ( plugin_abs_file_path ) ) <EOL> return registered_plugins <EOL> ALLOWED_EXTS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> PARSER_FUNCS = { '<STR_LIT>' : json . load , '<STR_LIT>' : yaml . safe_load , '<STR_LIT>' : yaml . safe_load } <EOL> def load_meta_file ( file_path ) : <EOL> if not os . path . isfile ( file_path ) : <EOL> raise Exception ( '<STR_LIT>' % file_path ) <EOL> file_name , file_ext = os . path . splitext ( file_path ) <EOL> if file_ext not in ALLOWED_EXTS : <EOL> raise Exception ( '<STR_LIT>' % <EOL> ( file_ext , file_path , ALLOWED_EXTS ) ) <EOL> with open ( file_path , '<STR_LIT:r>' ) as f : <EOL> return PARSER_FUNCS [ file_ext ] ( f ) </s>
