elodie.plugins.plugins.PluginDb.__init__() - 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-11 07:47 UTC

elodie.plugins.plugins.PluginDb.init() A

↳ Parent: elodie.plugins.plugins

Complexity

Conditions

Size

Total Lines	14
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

jmathai / elodie

Pull Request — master (#319)

elodie.plugins.plugins.PluginDb.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

elodie.plugins.plugins.PluginDb.init() A