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

elodie.plugins.plugins.PluginDb.set() A

↳ Parent: elodie.plugins.plugins

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	3
dl	0
loc	7
rs	10
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, 'w+') 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:
            db = loads(f.read())

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


    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)
        with io.open(self.db_file, 'rb+') as f:
            f.write(unicode(dumps(db, ensure_ascii=False).encode('utf8')))



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, 'w+') 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			db = loads(f.read())
83
84			db[key] = value
85			with io.open(self.db_file, 'rb+') as f:
86			f.write(unicode(dumps(db, ensure_ascii=False).encode('utf8')))
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-10 09:07 UTC by Report Bug Copy Issue Report The variable `unicode` does not seem to be defined. Loading history...
87
88			def get_all(self):
89			with io.open(self.db_file, 'r') as f:
90			db = loads(f.read())
91			return db
92
93			def delete(self, key):
94			with io.open(self.db_file, 'r') as f:
95			db = loads(f.read())
96
97			# delete key without throwing an exception
98			db.pop(key, None)
99			with io.open(self.db_file, 'rb+') as f:
100			f.write(unicode(dumps(db, ensure_ascii=False).encode('utf8')))
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-07-10 09:07 UTC by Report Bug Copy Issue Report The variable `unicode` does not seem to be defined. Loading history...
101
102
103			class Plugins(object):
104			"""A class to execute plugin actions."""
105
106			def __init__(self):
107			self.plugins = []
108			self.classes = {}
109			self.loaded = False
110
111			def load(self):
112			"""Load plugins from config file.
113			"""
114			# If plugins have been loaded then return
115			if self.loaded == True:
116			return
117
118			plugin_list = load_plugin_config()
119			for plugin in plugin_list:
120			plugin_lower = plugin.lower()
121			try:
122			# We attempt to do the following.
123			# 1. Load the module of the plugin.
124			# 2. Instantiate an object of the plugin's class.
125			# 3. Add the plugin to the list of plugins.
126			#
127			# #3 should only happen if #2 doesn't throw an error
128			this_module = import_module('elodie.plugins.{}.{}'.format(plugin_lower, plugin_lower))
129			self.classes[plugin] = getattr(this_module, plugin)()
130			# We only append to self.plugins if we're able to load the class
131			self.plugins.append(plugin)
132			except:
133			log.error('An error occurred initiating plugin {}'.format(plugin))
134			log.error(format_exc())
135
136			self.loaded = True
137
138			def run_all_before(self, file_path, destination_folder):
139			"""Process `before` methods of each plugin that was loaded.
140			"""
141			self.load()
142			pass_status = True
143			for cls in self.classes:
144			this_method = getattr(self.classes[cls], 'before')
145			# We try to call the plugin's `before()` method.
146			# If the method explicitly raises an ElodiePluginError we'll fail the import
147			# by setting pass_status to False.
148			# If any other error occurs we log the message and proceed as usual.
149			# By default, plugins don't change behavior.
150			try:
151			this_method(file_path, destination_folder)
152			except ElodiePluginError as err:
153			log.warn('Plugin {} raised an exception: {}'.format(cls, err))
154			log.error(format_exc())
155			pass_status = False
156			except:
157			log.error(format_exc())
158			return pass_status
159
160			def run_all_after(self, file_path, destination_folder, final_file_path, metadata):
161			"""Process `before` methods of each plugin that was loaded.
162			"""
163			self.load()
164			pass_status = True
165			for cls in self.classes:
166			this_method = getattr(self.classes[cls], 'after')
167			# We try to call the plugin's `before()` method.
168			# If the method explicitly raises an ElodiePluginError we'll fail the import
169			# by setting pass_status to False.
170			# If any other error occurs we log the message and proceed as usual.
171			# By default, plugins don't change behavior.
172			try:
173			this_method(file_path, destination_folder, final_file_path, metadata)
174			log.info('Called after() for {}'.format(cls))
175			except ElodiePluginError as err:
176			log.warn('Plugin {} raised an exception: {}'.format(cls, err))
177			log.error(format_exc())
178			pass_status = False
179			except:
180			log.error(format_exc())
181			return pass_status
182
183			def run_batch(self):
184			self.load()
185			pass_status = True
186			for cls in self.classes:
187			this_method = getattr(self.classes[cls], 'batch')
188			# We try to call the plugin's `before()` method.
189			# If the method explicitly raises an ElodiePluginError we'll fail the import
190			# by setting pass_status to False.
191			# If any other error occurs we log the message and proceed as usual.
192			# By default, plugins don't change behavior.
193			try:
194			this_method()
195			except ElodiePluginError as err:
196			log.warn('Plugin {} raised an exception: {}'.format(cls, err))
197			log.error(format_exc())
198			pass_status = False
199			except:
200			log.error(format_exc())
201			return pass_status
202

jmathai / elodie

Pull Request — master (#319)

elodie.plugins.plugins.PluginDb.set() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like