Harpseal Base

Harpseal App

Harpseal app, includes web server and harpseal daemon.

class harpseal.app.Harpseal(conf='config.json')[source]

Bases: harpseal.plugin.PluginMixin

Harpseal daemon.

config = None

(harpseal.conf.Config) Harpseal configuration

loop = None

(asyncio.BaseEventLoop) Base event-loop


Store a new mongo instance with given plugin data to mongodb when got a new result.

plugins = None

(tuple) Plugins

queue = None

(:class:`asyncio.Queue’) Queue that saves plugin result to store data to mongodb


Start a web server and periodic task after register and execute plugins.

Parameters:loop – Base event-loop
tasks = None

(tuple) Tasks (by plugins)

web = None

(harpseal.web.WebServer) Harpseal API server

Harpseal Classes

Classes, include Mixin, Task and others.

class harpseal.classes.PeriodicTask(plugin, app)[source]

Bases: object

A class for creating periodic task.

plugin = None

(harpseal.plugin.Plugin) Target plugin to run periodically


Execute plugin and then put the result into app’s queue.


Start a periodic task.

class harpseal.classes.StrictDict(fields)[source]

Bases: object

A dictionary-like semi-dynamic dict class. Values are stored in strictly with the field types given.


Get a value of name key.


Return the keys of dict. The result is not a kind of view.

set(name, value)[source]

Set a value of name key; to be stored in strictly (type-sensitive).

Config parser

Harpseal config parser based on JSON

class harpseal.conf.Config(path)[source]

Bases: object

Config parser, working like dict.

Harpseal Plugin

class harpseal.plugin.Plugin[source]

Bases: object

Base plugin model


import asyncio
from harpseal.plugin import Plugin

class YourPlugin(Plugin):
    name = '(required) your-plugin-name'
    description = '(required) plugin description here'
    priority = 0  # not yet implemented
    every = 1  # every 1 minute

    def init(self):
        # graph type definitions
        self.field_types['a'] = 'line'
        self.field_types['b'] = 'stack'
        self.field_types['c'] = 'full-stack'
        # field type definitions (`int` or `float`)
        self.fields['a'] = [('normal', int, ), ('abnormal', int, ), ]
        self.fields['b'] = [('normal', float, ), ('abnormal', float, ), ]
        self.fields['c'] = [('normal', float, ), ('abnormal', float, ), ]

    def provider(self):  # data provider
        data = self.data_form()
        data['a'].set('normal', 100)
        data['a'].set('abnormal', 150)
        # ...
        return data

Execute a command on the event-loop and then return the result when finished.

description = ''

Plugin description

every = 1

Plugin execution cycle (minutes)


Execute plugin

field_types = None

(collections.defaultdict) Model field type definitions (line, stack, full-stack, bar)

fields = None

(collections.defaultdict) Model field definitions

last_executed_at = None

(datetime.datetime) Last executed time

last_executed_result = None

(bool) Last executed result

models = None

(dict) Model definitions to be stored after executed init method

name = ''

Plugin name

priority = 0

Execution priority


Return the instance properties as a dict.


Plugin provider that returns a plugin’s result, this method must be overridden.

class harpseal.plugin.PluginMixin[source]

Bases: harpseal.plugin._PluginMixin

Mixin class for Plugin management.



harpseal.models.make_model(name, args)[source]

Create a new model with name and arguments given.