flaschengeist/flaschengeist/plugins/__init__.py

186 lines
5.9 KiB
Python

import sqlalchemy
import pkg_resources
from werkzeug.exceptions import MethodNotAllowed, NotFound
from flaschengeist.database import db
from flaschengeist.models.user import _Avatar
from flaschengeist.models.setting import _PluginSetting
from flaschengeist.utils.hook import HookBefore, HookAfter
before_role_updated = HookBefore("update_role")
"""Hook decorator for when roles are modified
Args:
role: Role object to modify
new_name: New name if the name was changed (None if delete)
"""
after_role_updated = HookAfter("update_role")
"""Hook decorator for when roles are modified
Args:
role: Role object containing the modified role
new_name: New name if the name was changed (None if deleted)
"""
before_update_user = HookBefore("update_user")
"""Hook decorator, when ever an user update is done, this is called before.
Args:
user: User object
"""
class Plugin:
"""Base class for all Plugins
If your class uses custom models add a static property called ``models``"""
def __init__(self, config=None, blueprint=None, permissions=[]):
"""Constructor called by create_app
Args:
config: Dict configuration containing the plugin section
blueprint: A flask blueprint containing all plugin routes
permissions: List of permissions of this Plugin
"""
self.blueprint = blueprint
self.permissions = permissions
self.version = pkg_resources.get_distribution(self.__module__.split(".")[0]).version
def install(self):
"""Installation routine
Is always called with Flask application context
"""
pass
def post_install(self):
"""Fill database or do other stuff
Called after all plugins are installed
"""
pass
def get_setting(self, name: str, **kwargs):
"""Get plugin setting from database
Args:
name: string identifying the setting
default: Default value
Returns:
Value stored in database (native python)
"""
try:
setting = (
_PluginSetting.query.filter(_PluginSetting.plugin == self._plugin_name)
.filter(_PluginSetting.name == name)
.one()
)
return setting.value
except sqlalchemy.orm.exc.NoResultFound:
if "default" in kwargs:
return kwargs["default"]
else:
raise KeyError
def set_setting(self, name: str, value):
"""Save setting in database
Args:
name: String identifying the setting
value: Value to be stored
"""
setting = (
_PluginSetting.query.filter(_PluginSetting.plugin == self._plugin_name)
.filter(_PluginSetting.name == name)
.one_or_none()
)
if setting is not None:
setting.value = value
else:
db.session.add(_PluginSetting(plugin=self._plugin_name, name=name, value=value))
db.session.commit()
def serialize(self):
"""Serialize a plugin into a dict
Returns:
Dict containing version and permissions of the plugin
"""
return {"version": self.version, "permissions": self.permissions}
class AuthPlugin(Plugin):
def login(self, user, pw):
"""Login routine, MUST BE IMPLEMENTED!
Args:
user: User class containing at least the uid
pw: given password
Returns:
Must return False if not found or invalid credentials, True if success
"""
raise NotImplemented
def update_user(self, user):
"""If backend is using external data, then update this user instance with external data
Args:
user: User object
"""
pass
def find_user(self, userid, mail=None):
"""Find an user by userid or mail
Args:
userid: Userid to search
mail: If set, mail to search
Returns:
None or User
"""
return None
def modify_user(self, user, password, new_password=None):
"""If backend is using (writeable) external data, then update the external database with the user provided.
User might have roles not existing on the external database, so you might have to create those.
Args:
user: User object
password: Password (some backends need the current password for changes) if None force edit (admin)
new_password: If set a password change is requested
Raises:
NotImplemented: If backend does not support this feature (or no password change)
BadRequest: Logic error, e.g. password is wrong.
Error: Other errors if backend went mad (are not handled and will result in a 500 error)
"""
raise NotImplemented
def create_user(self, user, password):
"""If backend is using (writeable) external data, then create a new user on the external database.
Args:
user: User object
password: string
"""
raise MethodNotAllowed
def delete_user(self, user):
"""If backend is using (writeable) external data, then delete the user from external database.
Args:
user: User object
"""
raise MethodNotAllowed
def get_avatar(self, user) -> _Avatar:
"""Retrieve avatar for given user (if supported by auth backend)
Args:
user: User to retrieve the avatar for
Raises:
NotFound: If no avatar found or not implemented
"""
raise NotFound
def set_avatar(self, user, avatar: _Avatar):
"""Set the avatar for given user (if supported by auth backend)
Args:
user: User to set the avatar for
avatar: Avatar to set
Raises:
MethodNotAllowed: If not supported by Backend
"""
raise MethodNotAllowed