Rackio API documentation

class rackio.Rackio()

Rackio main application class.

This class is a singleton by inheritance, this makes it available in each module of an end application or code project

Usage:

>>> from rackio import Rackio
>>> app = Rackio()
set_log(self, level=20, file='')

Sets the log file and level.

Parameters:

  • level (str): logging.LEVEL.
  • file (str): log filename.

Returns: None

Usage:

>>> app.set_log(file="app.log")
set_db(self, dbtype='sqlite', drop_table=True, **kwargs)

Sets the database, it supports SQLite and Postgres, in case of SQLite, the filename must be provided.

Parameters:

  • dbfile (str): a path to database file.
  • kwargs: Same attributes to a postgres connection.

Returns: None

Usage:

>>> app.set_db(dbfile="app.db")
set_workers(self, nworkers)

Sets the maximum workers in the ThreadPoolExecutor.

Parameters:

  • nworkers (int): Number of workers.

Returns: None

set_dbtags(self, tags, period=0.5, delay=1.0)

Sets the database tags for logging.

Parameters:

  • tags (list): A list of the tags.

Returns: None

Usage:

>>> tags = ["P1", "P2", "T1"]
>>> app.set_dbtags(tags, period=1.0)
get_dbtags(self)

Returns the database tags for logging.

append_rule(self, rule)

Append a rule to the control manager.

Parameters:

  • rule (Rule): a rule object.
get_rule(self, name)

Returns a Rule defined by its name.

Parameters:

  • name (str): a rackio rule.
append_control(self, control)

Append a control to the control manager.

Parameters:

  • control (Control): a control object.
get_control(self, name)

Returns a Control defined by its name.

Parameters:

  • name (str): a rackio control.
append_alarm(self, alarm)

Append an alarm to the alarm manager.

Parameters:

  • alarm (Alarm): an alarm object.
get_alarm(self, name)

Returns a Alarm defined by its name.

Parameters:

  • name (str): an alarm name.
get_alarm_by_tag(self, tag)

Returns a Alarm defined by its tag.

Parameters:

  • tag (str): an alarm tag.
append_machine(self, machine, interval=1)

Append a state machine to the state machine manager.

Parameters:

  • machine (RackioStateMachine): a state machine object.
  • interval (int): Interval execution time in seconds.
get_machine(self, name)

Returns a Rackio State Machine defined by its name.

Parameters:

  • name (str): a rackio state machine name.
define_machine(self, name='', interval=1, **kwargs)

Append a state machine to the state machine manager by a class decoration.

Parameters:

  • interval (int): Interval execution time in seconds.
append_table(self, table)

Append a database model class definition.

Parameters:

  • table (BaseModel): A Base Model Inheritance.
define_table(self, cls)

Append a database model class definition by a class decoration.

add_route(self, route, resource)

Append a resource and route the api.

Parameters:

  • route (str): The url route for this resource.
  • resource (object): a url resouce template class instance.
define_route(self, route, **kwargs)

Append a resource and route the api by a class decoration..

Parameters:

  • route (str): The url route for this resource.
rackit(self, period)

Decorator method to register functions plugins.

This method will register into the Rackio application a new function to be executed by the Thread Pool Executor

Parameters: * period (float): Value of the default loop execution time.

Returns: None

Usage:

@app.rackit
def hello():
    print("Hello!!!")
rackit_on(self, function=None, **kwargs)

Decorator to register functions plugins with continous execution.

This method will register into the Rackio application a new function to be executed by the Thread Pool Executor

Parameters: * period (float): Value of the default loop execution time, if period is not defined 0.5 seconds is used by default.

Returns: None

Usage:

@app.rackit_on(period=0.5)
def hello():
    print("Hello!!!")

@app.rackit_on
def hello_world():
    print("Hello World!!!")
observe(self, tag)

Decorator method to register functions as tag observers.

This method will register into the Rackio application a new function as a custom observer to be executed by the Thread Pool Executor. If the tag associated changes its value, the function registered will be executed.

Parameters: * tag (str): Tag name.

Returns: None

Usage:

@app.observer
def hello("T1"):
    print("Hello, Tag T1 has changed!!!")
run(self, port=8000)

Runs the main execution for the application to start serving.

This will put all the components of the application at run

Returns: None

Usage:

>>> app.run()