elodie.plugins.plugins.PluginDb.set() - Code Metrics - Inspection of "Store metadata in an SQLite database (gh-68)" - jmathai/elodie - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#443)

by Jaisen

created 2023-05-07 07:29 UTC

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

↳ Parent: elodie.plugins.plugins

Complexity

Conditions

Size

Total Lines	9
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	8
nop	3
dl	0
loc	9
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.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):
    """Exception which can be thrown by plugins to return failures.
    """
    pass


class PluginBase(object):
    """Base class which all plugins should inherit from.
       Defines stubs for all methods and exposes logging and database functionality
    """
    __name__ = 'PluginBase'

    def __init__(self):
        # Initializes variables for the plugin
        self.application_directory = application_directory
        # Loads the config for the plugin from config.ini
        self.config_for_plugin = load_config_for_plugin(self.__name__)
        # Initializes a database for the plugin.
        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):
        # Writes an info log not shown unless being run in --debug mode.
        log.info(dumps(
            {self.__name__: msg}
        ))

    def display(self, msg):
        # Writes a log for all modes and will be displayed.
        log.all(dumps(
            {self.__name__: msg}
        ))

    def generate_db(self, hash_db):
        pass

class PluginDb(object):
    """A database module which provides a simple key/value database.
       The database is a JSON file located at %application_directory%/plugins/%pluginname.lower()%.json
    """
    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):
    """Plugin object which manages all interaction with plugins.
       Exposes methods to load plugins and execute their methods.
    """

    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
24			class ElodiePluginError(Exception):
25			"""Exception which can be thrown by plugins to return failures.
26			"""
27			pass
28
29
30			class PluginBase(object):
31			"""Base class which all plugins should inherit from.
32			Defines stubs for all methods and exposes logging and database functionality
33			"""
34			__name__ = 'PluginBase'
35
36			def __init__(self):
37			# Initializes variables for the plugin
38			self.application_directory = application_directory
39			# Loads the config for the plugin from config.ini
40			self.config_for_plugin = load_config_for_plugin(self.__name__)
41			# Initializes a database for the plugin.
42			self.db = PluginDb(self.__name__)
43
44			def after(self, file_path, destination_folder, final_file_path, metadata):
45			pass
46
47			def batch(self):
48			pass
49
50			def before(self, file_path, destination_folder):
51			pass
52
53			def log(self, msg):
54			# Writes an info log not shown unless being run in --debug mode.
55			log.info(dumps(
56			{self.__name__: msg}
57			))
58
59			def display(self, msg):
60			# Writes a log for all modes and will be displayed.
61			log.all(dumps(
62			{self.__name__: msg}
63			))
64
65			def generate_db(self, hash_db):
66			pass
67
68			class PluginDb(object):
69			"""A database module which provides a simple key/value database.
70			The database is a JSON file located at %application_directory%/plugins/%pluginname.lower()%.json
71			"""
72			def __init__(self, plugin_name):
73			self.db_file = '{}/plugins/{}.json'.format(
74			application_directory,
75			plugin_name.lower()
76			)
77
78			# If the plugin db directory does not exist, create it
79			if(not isdir(dirname(self.db_file))):
80			mkdir(dirname(self.db_file))
81
82			# If the db file does not exist we initialize it
83			if(not isfile(self.db_file)):
84			with io.open(self.db_file, 'wb') as f:
85			f.write(_bytes(dumps({})))
86
87
88			def get(self, key):
89			with io.open(self.db_file, 'r') as f:
90			db = loads(f.read())
91
92			if(key not in db):
93			return None
94
95			return db[key]
96
97			def set(self, key, value):
98			with io.open(self.db_file, 'r') as f:
99			data = f.read()
100			db = loads(data)
101
102			db[key] = value
103			new_content = dumps(db, ensure_ascii=False).encode('utf8')
104			with io.open(self.db_file, 'wb') as f:
105			f.write(new_content)
106
107			def get_all(self):
108			with io.open(self.db_file, 'r') as f:
109			db = loads(f.read())
110			return db
111
112			def delete(self, key):
113			with io.open(self.db_file, 'r') as f:
114			db = loads(f.read())
115
116			# delete key without throwing an exception
117			db.pop(key, None)
118			new_content = dumps(db, ensure_ascii=False).encode('utf8')
119			with io.open(self.db_file, 'wb') as f:
120			f.write(new_content)
121
122
123			class Plugins(object):
124			"""Plugin object which manages all interaction with plugins.
125			Exposes methods to load plugins and execute their methods.
126			"""
127
128			def __init__(self):
129			self.plugins = []
130			self.classes = {}
131			self.loaded = False
132
133			def load(self):
134			"""Load plugins from config file.
135			"""
136			# If plugins have been loaded then return
137			if self.loaded == True:
138			return
139
140			plugin_list = load_plugin_config()
141			for plugin in plugin_list:
142			plugin_lower = plugin.lower()
143			try:
144			# We attempt to do the following.
145			# 1. Load the module of the plugin.
146			# 2. Instantiate an object of the plugin's class.
147			# 3. Add the plugin to the list of plugins.
148			#
149			# #3 should only happen if #2 doesn't throw an error
150			this_module = import_module('elodie.plugins.{}.{}'.format(plugin_lower, plugin_lower))
151			self.classes[plugin] = getattr(this_module, plugin)()
152			# We only append to self.plugins if we're able to load the class
153			self.plugins.append(plugin)
154			except:
155			log.error('An error occurred initiating plugin {}'.format(plugin))
156			log.error(format_exc())
157
158			self.loaded = True
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 in run_all_before: {}'.format(cls, err))
177			log.error(format_exc())
178			log.error('false')
179			pass_status = False
180			except:
181			log.error(format_exc())
182			return pass_status
183
184			def run_batch(self):
185			self.load()
186			pass_status = True
187			for cls in self.classes:
188			this_method = getattr(self.classes[cls], 'batch')
189			# We try to call the plugin's `before()` method.
190			# If the method explicitly raises an ElodiePluginError we'll fail the import
191			# by setting pass_status to False.
192			# If any other error occurs we log the message and proceed as usual.
193			# By default, plugins don't change behavior.
194			try:
195			this_method()
196			log.info('Called batch() for {}'.format(cls))
197			except ElodiePluginError as err:
198			log.warn('Plugin {} raised an exception in run_batch: {}'.format(cls, err))
199			log.error(format_exc())
200			pass_status = False
201			except:
202			log.error(format_exc())
203			return pass_status
204
205			def run_all_before(self, file_path, destination_folder):
206			"""Process `before` methods of each plugin that was loaded.
207			"""
208			self.load()
209			pass_status = True
210			for cls in self.classes:
211			this_method = getattr(self.classes[cls], 'before')
212			# We try to call the plugin's `before()` method.
213			# If the method explicitly raises an ElodiePluginError we'll fail the import
214			# by setting pass_status to False.
215			# If any other error occurs we log the message and proceed as usual.
216			# By default, plugins don't change behavior.
217			try:
218			this_method(file_path, destination_folder)
219			log.info('Called before() for {}'.format(cls))
220			except ElodiePluginError as err:
221			log.warn('Plugin {} raised an exception in run_all_after: {}'.format(cls, err))
222			log.error(format_exc())
223			pass_status = False
224			except:
225			log.error(format_exc())
226			return pass_status
227

jmathai / elodie

Pull Request — master (#443)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like