elodie.plugins.plugins.Plugins.run_batch() - Code Metrics - Inspection of "Add a plugin to upload photos to Google Photos" - jmathai/elodie - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#319)

by Jaisen

created 2019-07-10 07:09 UTC

elodie.plugins.plugins.Plugins.run_batch() A

↳ Parent: elodie.plugins.plugins

Complexity

Conditions

Size

Total Lines	19
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
eloc	14
nop	1
dl	0
loc	19
rs	9.7
c	0
b	0
f	0

"""
Plugin object.

.. moduleauthor:: Jaisen Mathai <[email protected]>
"""
from __future__ import print_function
from builtins import object

from json import dumps, loads
from importlib import import_module
from os.path import dirname, dirname, isdir, isfile
from os import mkdir
from sys import exc_info
from traceback import format_exc

from elodie.config import load_config_for_plugin, load_plugin_config
from elodie.constants import application_directory
from elodie import log

class ElodiePluginError(Exception):
    pass


class PluginBase(object):

    __name__ = 'PluginBase'

    def __init__(self):
        self.config_for_plugin = load_config_for_plugin(self.__name__)
        self.db = PluginDb(self.__name__)

    def after(self, file_path, destination_folder, final_file_path, media):
        pass

    def batch(self):
        pass

    def before(self, file_path, destination_folder, media):
        pass

    def log(self, msg):
        log.info(msg)

    def display(self, msg):
        log.all(dumps(
            {self.__name__: msg}
        ))

class PluginDb(object):

    def __init__(self, plugin_name):
        self.db_file = '{}/plugins/{}.json'.format(
            application_directory,
            plugin_name.lower()
        )

        # If the plugin db directory does not exist, create it
        if(not isdir(dirname(self.db_file))):
            mkdir(dirname(self.db_file))

        # If the db file does not exist we initialize it
        if(not isfile(self.db_file)):
            with open(self.db_file, 'w+') as f:
                f.write(dumps({}))


    def get(self, key):
        with open(self.db_file, 'r') as f:
            db = loads(f.read())

        if(key not in db):
            return None

        return db[key]

    def set(self, key, value):
        with open(self.db_file, 'r') as f:
            db = loads(f.read())

        db[key] = value
        with open(self.db_file, 'w') as f:
            f.write(dumps(db))

    def get_all(self):
        with open(self.db_file, 'r') as f:
            db = loads(f.read())
        return db

    def delete(self, key):
        with open(self.db_file, 'r') as f:
            db = loads(f.read())

        # delete key without throwing an exception
        db.pop(key, None)
        with open(self.db_file, 'w') as f:
            f.write(dumps(db))


class Plugins(object):
    """A class to execute plugin actions."""

    def __init__(self):
        self.plugins = []
        self.classes = {}
        self.loaded = False

    def load(self):
        """Load plugins from config file.
        """
        # If plugins have been loaded then return
        if self.loaded == True:
            return

        plugin_list = load_plugin_config()
        for plugin in plugin_list:
            plugin_lower = plugin.lower()
            try:
                # We attempt to do the following.
                #  1. Load the module of the plugin.
                #  2. Instantiate an object of the plugin's class.
                #  3. Add the plugin to the list of plugins.
                #  
                #  #3 should only happen if #2 doesn't throw an error
                this_module = import_module('elodie.plugins.{}.{}'.format(plugin_lower, plugin_lower))
                self.classes[plugin] = getattr(this_module, plugin)()
                # We only append to self.plugins if we're able to load the class
                self.plugins.append(plugin)
            except:
                log.error('An error occurred initiating plugin {}'.format(plugin))
                log.error(format_exc())

        self.loaded = True

    def run_all_before(self, file_path, destination_folder, media):
        """Process `before` methods of each plugin that was loaded.
        """
        self.load()
        pass_status = True
        for cls in self.classes:
            this_method = getattr(self.classes[cls], 'before')
            # We try to call the plugin's `before()` method.
            # If the method explicitly raises an ElodiePluginError we'll fail the import
            #  by setting pass_status to False.
            # If any other error occurs we log the message and proceed as usual.
            # By default, plugins don't change behavior.
            try:
                this_method(file_path, destination_folder, media)
            except ElodiePluginError as err:
                log.warn('Plugin {} raised an exception: {}'.format(cls, err))
                log.error(format_exc())
                pass_status = False
            except:
                log.error(format_exc())
        return pass_status

    def run_all_after(self, file_path, destination_folder, final_file_path, media):
        """Process `before` methods of each plugin that was loaded.
        """
        self.load()
        pass_status = True
        for cls in self.classes:
            this_method = getattr(self.classes[cls], 'before')
            # We try to call the plugin's `before()` method.
            # If the method explicitly raises an ElodiePluginError we'll fail the import
            #  by setting pass_status to False.
            # If any other error occurs we log the message and proceed as usual.
            # By default, plugins don't change behavior.
            try:
                this_method(file_path, destination_folder, media)
            except ElodiePluginError as err:
                log.warn('Plugin {} raised an exception: {}'.format(cls, err))
                log.error(format_exc())
                pass_status = False
            except:
                log.error(format_exc())
        return pass_status

    def run_batch(self):
        self.load()
        pass_status = True
        for cls in self.classes:
            this_method = getattr(self.classes[cls], 'batch')
            # We try to call the plugin's `before()` method.
            # If the method explicitly raises an ElodiePluginError we'll fail the import
            #  by setting pass_status to False.
            # If any other error occurs we log the message and proceed as usual.
            # By default, plugins don't change behavior.
            try:
                this_method()
            except ElodiePluginError as err:
                log.warn('Plugin {} raised an exception: {}'.format(cls, err))
                log.error(format_exc())
                pass_status = False
            except:
                log.error(format_exc())
        return pass_status


1			"""
2			Plugin object.
3
4			.. moduleauthor:: Jaisen Mathai <[email protected]>
5			"""
6			from __future__ import print_function
7			from builtins import object
8
9			from json import dumps, loads
10			from importlib import import_module
11			from os.path import dirname, dirname, isdir, isfile
12			from os import mkdir
13			from sys import exc_info
14			from traceback import format_exc
15
16			from elodie.config import load_config_for_plugin, load_plugin_config
17			from elodie.constants import application_directory
18			from elodie import log
19
20			class ElodiePluginError(Exception):
21			pass
22
23
24			class PluginBase(object):
25
26			__name__ = 'PluginBase'
27
28			def __init__(self):
29			self.config_for_plugin = load_config_for_plugin(self.__name__)
30			self.db = PluginDb(self.__name__)
31
32			def after(self, file_path, destination_folder, final_file_path, media):
33			pass
34
35			def batch(self):
36			pass
37
38			def before(self, file_path, destination_folder, media):
39			pass
40
41			def log(self, msg):
42			log.info(msg)
43
44			def display(self, msg):
45			log.all(dumps(
46			{self.__name__: msg}
47			))
48
49			class PluginDb(object):
50
51			def __init__(self, plugin_name):
52			self.db_file = '{}/plugins/{}.json'.format(
53			application_directory,
54			plugin_name.lower()
55			)
56
57			# If the plugin db directory does not exist, create it
58			if(not isdir(dirname(self.db_file))):
59			mkdir(dirname(self.db_file))
60
61			# If the db file does not exist we initialize it
62			if(not isfile(self.db_file)):
63			with open(self.db_file, 'w+') as f:
64			f.write(dumps({}))
65
66
67			def get(self, key):
68			with open(self.db_file, 'r') as f:
69			db = loads(f.read())
70
71			if(key not in db):
72			return None
73
74			return db[key]
75
76			def set(self, key, value):
77			with open(self.db_file, 'r') as f:
78			db = loads(f.read())
79
80			db[key] = value
81			with open(self.db_file, 'w') as f:
82			f.write(dumps(db))
83
84			def get_all(self):
85			with open(self.db_file, 'r') as f:
86			db = loads(f.read())
87			return db
88
89			def delete(self, key):
90			with open(self.db_file, 'r') as f:
91			db = loads(f.read())
92
93			# delete key without throwing an exception
94			db.pop(key, None)
95			with open(self.db_file, 'w') as f:
96			f.write(dumps(db))
97
98
99			class Plugins(object):
100			"""A class to execute plugin actions."""
101
102			def __init__(self):
103			self.plugins = []
104			self.classes = {}
105			self.loaded = False
106
107			def load(self):
108			"""Load plugins from config file.
109			"""
110			# If plugins have been loaded then return
111			if self.loaded == True:
112			return
113
114			plugin_list = load_plugin_config()
115			for plugin in plugin_list:
116			plugin_lower = plugin.lower()
117			try:
118			# We attempt to do the following.
119			# 1. Load the module of the plugin.
120			# 2. Instantiate an object of the plugin's class.
121			# 3. Add the plugin to the list of plugins.
122			#
123			# #3 should only happen if #2 doesn't throw an error
124			this_module = import_module('elodie.plugins.{}.{}'.format(plugin_lower, plugin_lower))
125			self.classes[plugin] = getattr(this_module, plugin)()
126			# We only append to self.plugins if we're able to load the class
127			self.plugins.append(plugin)
128			except:
129			log.error('An error occurred initiating plugin {}'.format(plugin))
130			log.error(format_exc())
131
132			self.loaded = True
133
134			def run_all_before(self, file_path, destination_folder, media):
135			"""Process `before` methods of each plugin that was loaded.
136			"""
137			self.load()
138			pass_status = True
139			for cls in self.classes:
140			this_method = getattr(self.classes[cls], 'before')
141			# We try to call the plugin's `before()` method.
142			# If the method explicitly raises an ElodiePluginError we'll fail the import
143			# by setting pass_status to False.
144			# If any other error occurs we log the message and proceed as usual.
145			# By default, plugins don't change behavior.
146			try:
147			this_method(file_path, destination_folder, media)
148			except ElodiePluginError as err:
149			log.warn('Plugin {} raised an exception: {}'.format(cls, err))
150			log.error(format_exc())
151			pass_status = False
152			except:
153			log.error(format_exc())
154			return pass_status
155
156			def run_all_after(self, file_path, destination_folder, final_file_path, media):
157			"""Process `before` methods of each plugin that was loaded.
158			"""
159			self.load()
160			pass_status = True
161			for cls in self.classes:
162			this_method = getattr(self.classes[cls], 'before')
163			# We try to call the plugin's `before()` method.
164			# If the method explicitly raises an ElodiePluginError we'll fail the import
165			# by setting pass_status to False.
166			# If any other error occurs we log the message and proceed as usual.
167			# By default, plugins don't change behavior.
168			try:
169			this_method(file_path, destination_folder, media)
170			except ElodiePluginError as err:
171			log.warn('Plugin {} raised an exception: {}'.format(cls, err))
172			log.error(format_exc())
173			pass_status = False
174			except:
175			log.error(format_exc())
176			return pass_status
177
178			def run_batch(self):
179			self.load()
180			pass_status = True
181			for cls in self.classes:
182			this_method = getattr(self.classes[cls], 'batch')
183			# We try to call the plugin's `before()` method.
184			# If the method explicitly raises an ElodiePluginError we'll fail the import
185			# by setting pass_status to False.
186			# If any other error occurs we log the message and proceed as usual.
187			# By default, plugins don't change behavior.
188			try:
189			this_method()
190			except ElodiePluginError as err:
191			log.warn('Plugin {} raised an exception: {}'.format(cls, err))
192			log.error(format_exc())
193			pass_status = False
194			except:
195			log.error(format_exc())
196			return pass_status
197

jmathai / elodie

Pull Request — master (#319)

elodie.plugins.plugins.Plugins.run_batch() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like