API References

Plugin

class src.Plugin(import_name: Optional[str] = None, static_folder: Optional[str] = None, static_url_path: Optional[str] = None, template_folder: Optional[str] = None, root_path: Optional[str] = None)

Create plugin by instantiating this class.

After binding with the Flask application, the instance will be automatically discovered by the manager.

The Plugin class inherits from the flask.scaffold.Scaffold class, and its working principle is roughly similar to flask.Blueprint, but the difference is that the Plugin class implements dynamic routing management.

Parameters

• id (str) – using for identify and search plugin.

• name (str, optional) – plugin name.

• domain (str) – scope of the plug-in determines the path through which the plug-in can be accessed, your plugin will be accessible in pattern: /{PluginManager.config.blueprint}/{Plugin.domain}/{endpoint}.

• import_name (str, optional) – plugin will inspect your module __name__, but unless you have other reasons, please use __name__ as the parameter value when registering the plug-in, otherwise don’t pass it in. Defaults to None.

• static_folder (str, optional) – static resource directory. Defaults to None.

• template_folder – (str, optional): template folder inside plugin directory. Defaults to None.

• static_url_path (str, optional) – static url path. Defaults to None.

• root_path (str, optional) – when you initialize the plugin with a not __name__ parameter import_name, you should pass this parameter as your plugin directory, because flask will unable to locate your plugin. Defaults to None.

Raises

• ValueError – if we could not use inspect to get valid module __name__ it will raise ValueError.

• FileNotFoundError – if plugin config not found.

• ValidationError – if plugin config not valid with schema.

Variables

• id_ – plugin id.

• domain – plugin domain.

• info – plugin info utils.attrdict.

• basedir – plugin dirname.

• status – plugin status machine.

• name – plugin name.

property endpoints: Set[str]

Return all active endpoints.

When status() is states.PluginStatus.Unloaded, means plugin not loaded endpoints refers to empty set.

Once loaded plugins, return all registered endpoints.

Returns

all endpoints.

Return type

t.Set[str]

static notfound(*args, **kwargs) → flask.wrappers.Response

A shortcut function to flask.abort.

When we stopped or removed a plugin, followed mapping from urls to endpoints should be removed also.

However, after binding url to werkzeug.routing.Map, we need to call werkzeug.routing.Map.remap for remapping, it will re-compile all Regex instances, which will also caused huge performance cost.

So here is a more elegant and easier way:

Just remap these endpoints to an ‘invalid’ function which will directly call flask.abort.

Returns

404 Not Found.

Return type

Response

add_url_rule(rule: str, endpoint: Optional[str] = None, view_func: Optional[Callable] = None, provide_automatic_options: Optional[bool] = None, **options: Any) → None

Register a rule for routing incoming requests and building URLs. The route() decorator is a shortcut to call this with the view_func argument. These are equivalent:

@app.route("/")
def index():
    ...

def index():
    ...

app.add_url_rule("/", view_func=index)

See Flask API Documentation.

The endpoint name for the route defaults to the name of the view function if the endpoint parameter isn’t passed. An error will be raised if a function has already been registered for the endpoint.

The methods parameter defaults to ["GET"]. HEAD is always added automatically, and OPTIONS is added automatically by default.

view_func does not necessarily need to be passed, but if the rule should participate in routing an endpoint name must be associated with a view function at some point with the endpoint() decorator.

app.add_url_rule("/", endpoint="index")

@app.endpoint("index")
def index():
    ...

If view_func has a required_methods attribute, those methods are added to the passed and automatic methods. If it has a provide_automatic_methods attribute, it is used as the default if the parameter is not passed.

Parameters

• rule – The URL rule string.

• endpoint – The endpoint name to associate with the rule and view function. Used when routing and building URLs. Defaults to view_func.__name__.

• view_func – The view function to associate with the endpoint name.

• provide_automatic_options – Add the OPTIONS method and respond to OPTIONS requests automatically.

• options – Extra options passed to the Rule object.

endpoint(endpoint: str) → Callable

Decorate a view function to register it for the given endpoint.

Use if a rule is added without a view_func with Plugin.add_url_rule().

Parameters

endpoint (str) – endpoint name

Returns

decorated function

Return type

t.Callable

register_error_handler(code_or_exception: Union[Type[flask.typing.GenericException], int], f: ft.ErrorHandlerCallable[ft.GenericException]) → None

Alternative error attach function to the errorhandler() decorator that is more straightforward to use for non decorator usage.

New in version 0.7.

export_status_to_dict() → Dict

Export plugin info to dict.

Included keys:

• id: plugin id.

• name: plugin name.

• status: plugin status, refered to states.PluginStatus.

• domain: plugin working domain.

• info: other plugin info.

Returns

plugin info and status.

Return type

t.Dict

load(app: flask.app.Flask, config: src.utils.staticdict) → None

Load plugin.

All routes inside plugin module are prepard in deferred registering functions.

Set current plugin status to states.PluginStatus.Loaded.

register(app: flask.app.Flask, config: src.utils.staticdict) → None

Register plugin into manager.

Execute all deferred registering functions, and transfer plugin status to states.PluginStatus.Running.

unregister(app: flask.app.Flask, config: src.utils.staticdict) → None

Unregister plugin.

Redirect all plugin endpoints in app.view_function to Plugin.notfound(), which is a shortcut to flask.abort(404).

clean(app: flask.app.Flask, config: src.utils.staticdict) → None

Clean plugin resource and unload module.

Deferred clean fucntions will be executed to remove all url rule in app.url_rules which used by plugin, also pop all preprocessors and error handler registered in app.

PluginManager

class src.PluginManager(app: Optional[flask.app.Flask] = None)

PluginManager allows you to load, start, stop and unload plugin.

After being bound with the Flask application, manager can automatically discover plugins in configuration directory and record their status.

Also, PluginManager provides a set of control functions to manage plugin.

Config Items:

• blueprint: will be applied as the name of the plug-in blueprint and the corresponding url_prefix.

• directory: the plugins path relative to the application directory.

• excludes_directory: directories that are skipped when scanning.

If app not provided, you can use PluginManager.init_app() with your app to initialize and configure later.

init_app(app: flask.app.Flask) → None

Initialize manager, load configs, create and bind blueprint for plugin management.

Blueprint named config.blueprint will be created and registered in app with argument url_prefix as same as config.blueprint.

Blueprint will have two functioncs registered as before_request and after_request:

• before_request: using PluginManager.dynamic_select_jinja_loader() to set app global jinja_loader in app.jinja_env.loader to Plugin.jinja_loader().

• after_request: restore app.jinja_env.loader to raw app.jinja_loader.

app.plugin_manager will be bind to reference of current manager, so it can be used with request context using current_app.plugin_manager.

Parameters

app (Flask) – your flask application.

static dynamic_select_jinja_loader() → Optional[jinja2.loaders.FileSystemLoader]

Dynamic switch plugin jinja_loader to replace app.jinja_env.loader.

If routing to an exist plugin, request.blueprints will be a list like: ['plugins.PLUGIN_DOMAIN', 'plugins'].

So select first blueprint and using .lstrip(self._config.blueprint + '.') to get current plugin domain.

Then iter self._loaded plugins to find which domain are registered into it. And becasue Plugin inherit from Scaffold, it can handle plugin.jinja_loader correctly, just return it.

It cannot use locked_cached_property because we hope template loader switch dynamically everytime.

Returns

plugin.jinja_loader

Return type

Optional[BaseLoader]

property status: List[Dict]

Return all plugins status dict, calling Plugin.export_status_to_dict().

Returns

all plugins status.

Return type

t.List[t.Dict]

property domain: str

PluginMangaer domain bound to blueprint name and url_prefix.

property basedir: str

Return working dir for plugin manager.

property plugins: Iterable[src.plugin.Plugin]

Iter all plugins, including loaded and not loaded.

Firstly iter all unloaded plugins using scan() for scanning unloaded plugins, then give a copy list of loaded plugins references.

Returns

plugin.

Return type

t.Iterable[Plugin]

find(id_: Optional[str] = None, domain: Optional[str] = None, name: Optional[str] = None) → Optional[src.plugin.Plugin]

Find a plugin.

Parameters

• id (str, optional) – plugin id. Defaults to None.

• domain (str, optional) – plugin domain. Defaults to None.

• name (str, optional) – plugin name. Defaults to None.

Returns

found plugin or None means no plugin found.

Return type

t.Optional[Plugin]

scan() → Iterable[src.plugin.Plugin]

Scan all unloaded plugin configured in config.directory.

After scanning and importing module as plugin, it will bind Plugin.basedir to the dir name used for importing.

Plugin module will be named by rule, which gives hint to Flask for loading static files and templates:

app.import_name + '.' + config.directory + '.' + plugin.basedir.

Yields

Iterator[t.Iterable[t.Tuple[Plugin, str]]] – couple Plugin with plugin dirname.

load_config(app: flask.app.Flask) → src.utils.staticdict

Load config from Flask app config.

For configuration values not specified in app.config, default settings in config.DefaultConfig will be used.

All configs will also been update in app.config.

Parameters

app (Flask) – Flask instance.

Returns

loaded config.

Return type

utils.staticdict

load(plugin: src.plugin.Plugin) → None

Load plugin.

Raises

• RuntimeError – when plugin status not allowed to load.

• RuntimeError – when found deplicated plugin id.

• RuntimeError – when plugin not scanned by PluginManager, which means have invalid attribute Plugin.basedir.

start(plugin: src.plugin.Plugin) → None

Start plugin.

Raises

RuntimeError – when plugin status not allowed to start.

stop(plugin: src.plugin.Plugin) → None

Stop plugin.

Raises

RuntimeError – when plugin status not allowed to stop.

unload(plugin: src.plugin.Plugin) → None

Unload plugin.

Raises

RuntimeError – when plugin status not allowed to unload.

states module

class src.states.PluginStatus(value)

Plugin Status Enumerating.

Plugin status could be 4 enum.Enum values:

0. Loaded: When we called __import__ for importing plugin moudule and all view function has been added to Plugin.endpoints(). But in app.url_map there’s no record added.

1. Running: After called Plugin.register() all mapping from endpoint to function will be added to app.url_map so plugin will run functionally.

2. Stopped: After we called Plugin.unregister(), all record inside app.url_map will still exist, but mapping from endpoints to view functions in app.view_functions will be point to Plugin.notfound() which will directly return HTTP 404.

3. Unloaded: After calling Plugin.clean(), records in app.url_map will be remapped, and app.view_functions will also be removed, all data inner Plugin instance will be cleaned also.

Enumerations:

Loaded = 0

Running = 1

Stopped = 2

Unloaded = 3

class src.states.StateMachine(table: Dict[Tuple[src.states.PluginStatus, str], src.states.PluginStatus], current: src.states.PluginStatus = PluginStatus.Unloaded)

We dont want check Plugin.status everytime to ensure if an operation is suitable for execution, so it’s better to write an simple finite-state-machine to manage:

>>> machine = StateMachine(transfer_table, current_state)
>>> if machine.allow('start'):
    ... # Operations
>>> machine.assert_allow('start')

property value: src.states.PluginStatus

Return current state.

allow(operation: str) → bool

Check if operation allow in current state.

Parameters

operation (str) – operation going to be execute.

Returns

if allowed this operation.

Return type

bool

assert_allow(operation)

Assert current state acceptable with this operation.

Parameters

operation ([type]) – operation going to be execute.

Raises

RuntimeError – raise if transfer not allowed by table.

utils module

Contains some helper functions and classes.

class src.utils.attrdict

Sub-class like python dict with support for I/O like attr.

>>> profile = attrdict({
    'languages': ['python', 'cpp', 'javascript', 'c'],
    'nickname': 'doge gui',
    'age': 23
})
>>> profile.languages.append('Russian')  # Add language to profile
>>> profile.languages
['python', 'cpp', 'javascript', 'c', 'Russian']
>>> profile.age == 23
True

Attribute-like key should not be methods with dict, and obey python syntax:

>>> profile.1 = 0
Traceback (most recent call last):
    ...
SyntaxError: invalid syntax
>>> profile.popitem = None  # Rewrite

class src.utils.staticdict

staticdict inherit all behaviors from attrdict but banned all writing operations on it.

>>> final = staticdict({
    'loaded': False,
    'config': './carental/config.py'
})
>>> not final.loaded is True
True
>>> final.brand = 'new'
Traceback (most recent call last):
    ...
RuntimeError: cannot set value on staticdict

class src.utils.property_(name: str, type_: Type[src.utils._T] = typing.Any, prefix: str = '_', writable: bool = False, delectable: bool = False)

A one line property decorator to support reading class attributes with prefix.

Here is a sub-class inherit from dict which support it, by using __getattr__, __setattr__, and __delattr__. When we want to declare a property inside class, we always doing this:

>>> class Bar:
    def __init__(self, size: int, count: int) -> None:
        self._size = size
        self._count = count
    @property
    def size(self) -> int:
        return self._size
    @property
    def count(self) -> int:
        return self._count

Obviously its sth like redundancy, by using this property_ function we could:

>>> class AnotherBar(Bar):
        size = property_('size', type_=int)
        count = property_('count', type_=int, writeable=True)

Also you could define which selector using before attribute. The default one is ‘_’.

src.utils.listdir(path: str, excludes: Optional[Container[str]] = None) → Iterator[str]

List all dir inside specific path.

Parameters

• path (str) – path to be explore.

• excludes (Container[str], optional) – dirname to exclude. Defaults to None.

Yields

Iterator[str] – absolute path of subdirectories.

Raises

FileNotFoundError – when given invalid path.

src.utils.rmdir(path: str) → None

Remove dir and file inside it.

Parameters

path (str) – absolute dir gonna remove.

Raises

FileNotFoundError – when path not exists.

src.utils.startstrip(string: str, part: str) → str

Remove part from beginning of string if string startswith part.

Parameters

• string (str) – source string.

• part (str) – removing part.

Returns

removed part.

Return type

str

signals module

Contains all custom signals based on flask.signals.Namespace.

All these signals send with caller as instance of PluginManager, and the only argument named plugin is plugin instance operated.

If you want to receive these signals, please install the blinker library, see: https://flask.palletsprojects.com/en/2.0.x/signals/

src.signals.loaded = <flask.signals._FakeSignal object>

Plugin loaded signal.

src.signals.started = <flask.signals._FakeSignal object>

Plugin started signal.

src.signals.stopped = <flask.signals._FakeSignal object>

Plugin stopped signal.

src.signals.unloaded = <flask.signals._FakeSignal object>

Plugin unloaded signal.

config module

src.config.RequirementsFile = 'requirements.txt'

Plugin requirements.txt filename.

src.config.ConfigFile = 'plugin.json'

Plugin description filename.

src.config.ConfigSchema = {'$schema': 'http://json-schema.org/schema', 'description': 'Flask-Plugin uses Json Schema to verify the configuration. The following configurations are currently supported. Some of these options are required, and these configurations will be used by the manager to identify plugins; others are optional, and they are used to describe plugins in more detail.', 'properties': {'domain': {'description': 'Plugin working domain. ', 'type': 'string'}, 'id': {'description': 'Plugin unique ID. Using for identify plugin.', 'type': 'string'}, 'plugin': {'description': 'Plugin description info.', 'properties': {'author': {'description': 'Plugin author.', 'type': 'string'}, 'description': {'description': 'A long description.', 'type': 'string'}, 'maintainer': {'description': 'Plugin maintainers.', 'items': {'type': 'string'}, 'type': 'array'}, 'name': {'description': 'Plugin name.', 'type': 'string'}, 'repo': {'description': 'Git repository adderss.', 'type': 'string'}, 'summary': {'description': 'A short description.', 'type': 'string'}, 'url': {'description': 'Plugin official site URL.', 'type': 'string'}}, 'required': ['name', 'author', 'summary'], 'type': 'object'}, 'releases': {'description': 'Plugin releases.', 'items': {'description': 'Released versions.', 'properties': {'download': {'description': 'Download zip package address.', 'type': 'string'}, 'note': {'description': 'Release note.', 'type': 'string'}, 'version': {'description': 'Released version number, will be parsed with python `packaging.version`.', 'type': 'string'}}, 'required': ['version', 'download'], 'type': 'object'}, 'type': 'array'}}, 'required': ['id', 'domain', 'plugin', 'releases'], 'title': 'Plugin Config', 'type': 'object'}

Plugin config schema using formatted with JSON Schema.

Required config file like:

{
    "id": "e9d78b6e91644381823c1aa6bdef5606",
    "domain": "flaskex",
    "plugin": {
        "name": "flaskex",
        "author": "anfederico",
        "summary": "Ported version of flaskex example."
    },
    "releases": [
        {
            "version": "0.0.1",
            "download": "https://github.com/anfederico/flaskex/releases/0.0.1.zip"
        }
    ]
}

src.config.ConfigPrefix = 'plugins_'

All configs in app.config should startswith it.

src.config.DefaultConfig: Dict[str, Any]

It will be using when config item not found in app.config.

DefaultConfig: t.Dict[str, t.Any] = staticdict({
    'blueprint': 'plugins',
    'directory': 'plugins',
    'excludes_directory': ['__pycache__']
})

src.config.validate(config: src.utils.attrdict) → None

Validate a plugin config in plugin.json.

Parameters

config (attrdict) – config dict like.

Raises

jsonschema.ValidationError – if not valid config.

Flask API Documentation

See here for more info: https://flask.palletsprojects.com/en/2.0.x/api