elodie.plugins.plugins.Plugins.run_all_after() - 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-12 05:43 UTC

elodie.plugins.plugins.Plugins.run_all_after() A

↳ Parent: elodie.plugins.plugins

Complexity

Conditions

Size

Total Lines	23
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
eloc	16
nop	5
dl	0
loc	23
rs	9.6
c	0
b	0
f	0

"""
Plugin object.

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

import io

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.compatability import _bytes
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, metadata):
        pass

    def batch(self):
        pass

    def before(self, file_path, destination_folder):
        pass

    def log(self, msg):
        log.info(dumps(
            {self.__name__: 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 io.open(self.db_file, 'wb') as f:
                f.write(_bytes(dumps({})))


    def get(self, key):
        with io.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 io.open(self.db_file, 'r') as f:
            data = f.read()
            db = loads(data)

        db[key] = value
        new_content = dumps(db, ensure_ascii=False).encode('utf8')
        with io.open(self.db_file, 'wb') as f:
            f.write(new_content)

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

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

        # delete key without throwing an exception
        db.pop(key, None)
        new_content = dumps(db, ensure_ascii=False).encode('utf8')
        with io.open(self.db_file, 'wb') as f:
            f.write(new_content)


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_after(self, file_path, destination_folder, final_file_path, metadata):
        """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], 'after')
            # 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, final_file_path, metadata)
                log.info('Called after() for {}'.format(cls))
            except ElodiePluginError as err:
                log.warn('Plugin {} raised an exception in run_all_before: {}'.format(cls, err))
                log.error(format_exc())
                log.error('false')
                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()
                log.info('Called batch() for {}'.format(cls))
            except ElodiePluginError as err:
                log.warn('Plugin {} raised an exception in run_batch: {}'.format(cls, err))
                log.error(format_exc())
                pass_status = False
            except:
                log.error(format_exc())
        return pass_status

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

jmathai / elodie

Pull Request — master (#319)

elodie.plugins.plugins.Plugins.run_all_after() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like