elodie.plugins.plugins.PluginBase.generate_db() - 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-03-11 05:52 UTC

elodie.plugins.plugins.PluginBase.generate_db() A

↳ Parent: elodie.plugins.plugins

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

jmathai / elodie

Pull Request — master (#443)

elodie.plugins.plugins.PluginBase.generate_db() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like