API reference

WebLab object

class weblablib.WebLab(app=None, callback_url=None, base_url=None, backend=None)

WebLab is a Flask extension that manages the settings (backend, session, etc.), and the registration of certain methods (e.g., on_start, etc.)

Initializes the object. All the parameters are optional.

Parameters:
  • app – the Flask application
  • base_url – the base URL to be used. By default, the WebLab URLs will be /weblab/sessions/<something>. If you provide base_url = '/foo', then it will be listening in /foo/weblab/sessions/<something>. This is the route that will be used in the Flask application (so if your application is deployed in /bar, then it will be /bar/foo/weblab/sessions/<something>. This URLs do NOT need to be publicly available (they can be only available to WebLab-Deusto if you want, by playing with the firewall or so). You can also configure it with WEBLAB_BASE_URL in the Flask configuration.
  • callback_url – a URL that WebLab will implement that must be public. For example, /mylab/callback/, this URL must be available to the final user. The user will be redirected there with a token and this code will redirect him to the initial_url. You can also configure it with WEBLAB_CALLBACK_URL in configuration.
init_app(app, backend=None)

Initialize the app. This method MUST be called (unless ‘app’ is provided in the constructor of WebLab; then it’s called in that moment internally). Most configuration variables are taken here (so changing app.config afterwards will not affect WebLab).

initial_url(func)

This must be called, and only once. It’s a decorator for establishing where the user should be redirected (the lab itself). Example:

@weblab.initial_url
def initial_url():
    return url_for('index')
Parameters:func – The function that will be called to get the initial URL. It takes no parameter.
on_start(func)

Register a method for being called when a new user comes. The format is:

@weblab.on_start
def start(client_data, server_data):
    initialize_my_resources() # Example code
Parameters:func – The function that will be used on start.

This function has two parameters:

Parameters:
  • client_data – Data provided by the WebLab-Deusto client. It is a dictionary with different parameters.
  • server_data – Data provided by the WebLab-Deusto server (username, etc., generally wrapped in the weblab_user method)
on_dispose(func)

Register a method for being called when a new user comes:

@weblab.on_dispose
def dispose():
    pass
clean_expired_users()

Typically, users are deleted by WebLab-Deusto calling the dispose method. However, in some conditions (e.g., restarting WebLab), the dispose method might not be called, and the laboratory can end in a wrong state. So as to avoid this, weblablib provides three systems:

  1. A command flask clean_expired_users.
  2. A thread that by default is running which calls this method every few seconds.
  3. This API method, available as weblab.clean_expired_users()
run_tasks()

Run all the pending tasks, once. It does not have any loop or waits for any new task: it just runs it once. You can use it in your code for running tasks in the pace you consider, or use flask weblab loop.

task(unique=None)

A task is a function that can be called later on by the WebLab wrapper. It is a set of threads running in the background, so you don’t need to deal with it later on:

@weblab.task()
def function(a, b):
    return a + b

You can either call it directly (no thread involved):

result = function(5, 3)

Or you can call it delayed (and it will be run in a different thread):

task_result = function.delay(5, 3)
task_result.task_id # The task identifier
task_result.status # Either submitted, running, done or failed
task_result.result # If done
task_result.error # If failed

Or you can call it synchronously (but run in other thread / process):

task_result = function.run_sync(5, 3) task_result.task_id # …

You can use this WebLabTask object, or later, you can get the task in other view using WebLab.get_task():

task_result = weblab.get_task(task_id)

By using WebLab.tasks() or WebLab.running_tasks() (if still running):

for task in weblab.running_tasks:
    if task.name == 'function':
       # ...

for task in weblab.tasks:
    if task.name == 'function':
       # ...

Or even by name directly with WebLab.get_running_task() or WebLab.get_running_tasks() or similar:

# Only the running ones
weblab.get_running_tasks(function) # By the function
weblab.get_running_tasks('function') # By the name

# All (running and stopped)
weblab.get_tasks(function)
weblab.get_tasks('function')

# Only one (the first result). Useful when you run one task only once.
weblab.get_running_task(function)
weblab.get_running_task('function')

# Yes, it's the same as get_task(task_id). It supports both
weblab.get_task(function)
weblab.get_task('function')

Finally, you can join a task with WebLabTask.join() or WebLab.join_tasks():

task.join() task.join(timeout=5) # Wait 5 seconds, raise an error task.join(stop=True) # Call .stop() first

weblab.join_tasks(function) weblab.join_tasks(‘function’, stop=True)

Params unique:If you want this task to be not called if another task of the same type. It can be: None (no uniqueness), ‘global’ or ‘user’.
get_task(identifier)

Given a task of the current user, return the WebLabTask object.

The identifier can be:
  1. A task_id
  2. A function name
  3. A function

See also WebLab.task() for examples.

Parameters:identifier – either a task_id, a function name or a function.
tasks

Return all the tasks created in the current session (completed or not)

See also WebLab.task() for examples.

running_tasks

Check which tasks are still running and return them.

See also WebLab.task() for examples.

get_running_tasks(func_or_name)

Get all the running tasks with a given name or function.

See also WebLab.task() for examples.

get_running_task(func_or_name)

Get any running task with the provided name (or function). This is useful when using unique=global|user (so you know there will be only one task using it).

See also WebLab.task() for examples.

Parameters:func_or_name – a function or a function name of a task
get_tasks(func_or_name)

Get all the tasks (running or stopped) with the provided name (or function). This is useful when using unique=global|user (so you know there will be only one task using it).

See also WebLab.task() for examples.

Parameters:func_or_name – a function or a function name of a task
join_tasks(func_or_name, timeout=None, stop=False)

Stop (optionally) and join all the tasks with a given name.

Parameters:
  • func_or_name – you can either provide the task function or its name (string)
  • timeout – seconds to wait. By default wait forever.
  • stop – call stop() to each task before call join().
create_token(size=None)

Create a URL-safe random token in a safe way. You can use it for secret generation.

Parameters:size – the size of random bytes. The token will be later converted to base64, so the length of the returned string will be different (e.g., size=32 returns length=43).
Returns:a unique token.
user_loader(func)

Create a user loader. It must be a function such as:

@weblab.user_loader
def user_loader(username_unique):
    return User.query.get(weblab_username=username_unique)

Or similar. Internally, you can also work with weblab_user, for creating the object if not present or similar.

With this, you can later do:

user_db = weblab_user.user

and internally it will call the user_loader to obtain the user associated to this current user. Otherwise, weblab_user.user will return None.

Parameters:func – The function that will be called.
loop(threads, reload)

Launch threads threads that run tasks and clean expired users continuously.

Parameters:
  • threads – Number of threads.
  • reload – Reload if the source code is changed. Defaults to FLASK_DEBUG.

Functions and properties

weblablib.get_weblab_user(cached=True)

Get the current user. If cached=True it will store it and return the same each time. If you need to get always a different one get it with cached=False. In long tasks, for example, it’s normal to call it with cached=False so it gets updated with whatever information comes from other threads.

Two shortcuts exist to this function:

Given that the function always returns a CurrentUser or ExpiredUser or AnonymousUser, it’s safe to do things like:

if not weblab_user.anonymous:
    print(weblab_user.username)
    print(weblab_user.username_unique)
Parameters:cached – if this method is called twice in the same thread, it will return the same object.
weblablib.weblab_user

Shortcut to get_weblab_user() with cached=True

weblablib.socket_weblab_user

Shortcut to get_weblab_user() with cached=False

weblablib.poll()

Schedule that in the end of this call, it will update the value of the last time the user polled.

weblablib.logout()

Notify WebLab-Deusto that the user left the laboratory, so next user can enter.

This process is not real time. What it happens is that WebLab-Deusto periodically is requesting whether the user is still alive or not. If you call logout, weblablib will reply the next time that the user left. So it may take some seconds for WebLab-Deusto to realize of that. You can regulate this time with WEBLAB_POLL_INTERVAL setting (defaults to 5).

Decorators

weblablib.requires_active(func)

Decorator. Requires the user to be an active user. If the user is not logged in or his/her time expired or he called logout, the method will not be called.

weblablib.requires_login(func)

Decorator. Requires the user to have logged in. For example, the user might have finished but you still want to display his/her results or similar. With this method, the user will still be able to use this method. Don’t use it with sensors, etc.

weblablib.socket_requires_active(func)

Decorator. Requires the user to be a user (only active); otherwise it calls Flask-SocketIO disconnect.

Essentially, equivalent to requires_active(), but calling disconnect. And obtaining the information in real time (important in Flask-SocketIO events, where the same thread is used for all the events and requires_active() caches the user data)

weblablib.socket_requires_login(func)

Decorator. Requires the user to be a user (expired or active); otherwise it calls Flask-SocketIO disconnect.

Essentially, equivalent to requires_login(), but calling disconnect. And obtaining the information in real time (important in Flask-SocketIO events, where the same thread is used for all the events and requires_login() caches the user data).

Tasks

You can start tasks with WebLab.task(), as explained in detail in the Tasks documentation section.

weblablib.current_task

Running outside a task, it returns None. Running inside a task, it returns the task object. This way you can know what is the task_id or read or modify WebLabTask.data.

weblablib.current_task_stopping

Running outside a task, it returns False. Running inside a task, it calls WebLabTask.stopping and returns its value.

class weblablib.WebLabTask(weblab, task_id)

WebLab-Task. You can create it by defining a task as in:

@weblab.task()
def my_task(arg1, arg2):
    return arg1 + arg2

And then running it:

task = my_task.delay(5, 10)
print(task.task_id)

Another option is to obtain it:

task = weblab.get_task(task_id)

Or simply:

tasks = weblab.tasks

You are not supposed to create this object.

The life cycle is very simple:

See also WebLab.task().

join(timeout=None, error_on_timeout=True)

Wait for the task to finish. timeout (in seconds), if set it will raise an exception if error_on_timeout, otherwise it will just finish. You can’t call this method from the task itself (a TimeoutError will be raised).

Be aware that if task1 starts task2 and joins, and task2 joins task1 a deadlock will happen.

Parameters:
  • timeout – None (to wait forever) or number of seconds to wait. It accepts float.
  • error_on_timeout – if True, a TimeoutError will be raised.
task_id

Returns the task identifier.

session_id

The current session_id that represents this session.

name

The name of the function. For example, in:

@weblab.task()
def my_task():
    pass

The name would be my_task.

data

Dictionary that you can use to store information from outside and inside the task running. For example, you may call:

@weblab.task()
def my_task():
    # Something long, but 80%
    current_task.data['percent'] = 0.8
    current_task.sync()

# From outside
task = weblab.get_task(task_id)
print(task.data.get('percent'))

Note: every time you call data, it returns a different object. Don’t do:

current_task.data['foo'] = 'bar'
update_data(new_data)

Deprecated since version 0.5.0.

Same as:

task.data = new_data
task.store()
Parameters:new_data – new data to be stored in the task data.
store()

Stores the current data. This can only be used from the task itself (not from outside).

Returns:the same WebLabTask (already updated)
retrieve()

Retrieves a new version of the current task (updating status, data, etc.).

Returns:the same WebLabTask (already updated)
stop()

Raise a flag so WebLabTask.stopping becomes True. This method does not guarantee anything: it only provides a flag so the task implementor can use it to stop earlier or stop any loop, by reading current_task_stopping.

Example:

@weblab.task()
def read_serial():
    while not current_task_stopping:
          data = read()
          process(data)

# Outside:
task.stop() # Causing the loop to finish
status

Current status, as string:

  • 'submitted' (not yet processed by a thread)
  • 'done' (finished)
  • 'failed' (there was an error)
  • 'running' (still running in a thread)
  • None (if the task does not exist anymore)

instead of comparing strings, you’re encouraged to use:

done

Has the task finished successfully?

running

Is the task still running? Note that this is False if it was submitted and not yet started. If you want to know in general if it has finished or not, use WebLabTask.finished

submitted

Is the task submitted but not yet processed by a worker?

failed

Has the task finished by failing?

finished

Has the task finished? (either right or failing)

stopping

Did anyone call WebLabTask.stop()? If so, you should stop running the current task.

result

In case of having finished succesfully (WebLabTask.done being True), this returns the result. Otherwise, it returns None.

error

In case of finishing with an exception (WebLabTask.failed being True), this returns information about the error caused. Otherwise, it returns None.

The information is provided in a dictionary as follows:

{
   'code': 'exception',
   'class': 'ExampleError',
   'message': '<result of converting the error in string>'
}

Users

class weblablib.CurrentUser(*args, **kwargs)

This class is a WebLabUser representing a user which is still actively using a laboratory. If the session expires, it will become a ExpiredUser.

data

User data. By default an empty dictionary. You can access to it and modify it across processes. See also CurrentUser.update_data().

update_data(new_data=<object object>)

Deprecated since version 0.5.0.

Use weblab_user.data.store() or don’t use anything if inside a view or on_start.

time_without_polling

Seconds without polling

time_left

Seconds left (0 if time passed)

to_expired_user()

Create a ExpiredUser based on the data of the user

user

Load a related user from a database or similar. You might want to keep information about the current user and use this information across sessions, depending on your laboratory. Examples of this could be logs, or shared resources (e.g., storing the last actions of the user for the next time they log in). So as to load the user from the database, you can use WebLab.user_loader() to define a user loader, and use this property to access later to it.

For example, using Flask-SQLAlchemy:

from mylab.models import LaboratoryUser

@weblab.on_start
def start(client_data, server_data):
    # ...
    user = LaboratoryUser.query.filter(identifier.username_unique).first()
    if user is None:
        # Make sure the user exists in the database
        user_folder = create_user_folder()
        user = LaboratoryUser(username=weblab_user.username,
                          identifier=username_unique,
                          user_folder=user_folder)
        db.session.add(user)
        db.session.commit()


@weblab.user_loader
def loader(username_unique):
    return LaboratoryUser.query.filter(identifier=username_unique).first()

# And then later:

@app.route('/lab')
@requires_active
def home():
    user_db = weblab_user.user
    open(os.path.join(user_db.user_folder, 'myfile.txt')).read()

@app.route('/files')
@requires_active
def files():
    user_folder = weblab_user.user.user_folder
    # ...
active

Is the user active and has not called logout()?

is_anonymous

Is the user anonymous? False

class weblablib.ExpiredUser(*args, **kwargs)

This class is a WebLabUser representing a user which has been kicked out already. Typically this ExpiredUser is kept in the backend for around an hour (it depends on WEBLAB_EXPIRED_USERS_TIMEOUT setting).

Most of the fields are same as in CurrentUser.

data

User data. By default an empty dictionary. You can access to it and modify it across processes.

Do not change this data in ExpiredUser.

update_data(value=None)

Update data. Not implemented in ExpiredUser (expect an error)

time_left

Seconds left (always 0 in this case)

active

Is it an active user? (False)

is_anonymous

Is it an anonymous user? (False)

class weblablib.AnonymousUser

Implementation of WebLabUser representing anonymous users.

active

Is active? Always False

is_anonymous

Is anonymous? Always True

locale

Language requested by WebLab-Deusto? Always None

data

An object that raises error if accessed as a dict

class weblablib.WebLabUser

Abstract representation of a WebLabUser. Implementations:

active

Is the user active right now or not?

is_anonymous

Was the user a valid user recently?

Errors

class weblablib.WebLabError

Wraps all weblab exceptions

class weblablib.NoContextError

Wraps the fact that it is attempting to call an object like session outside the proper scope.

class weblablib.InvalidConfigError

Invalid configuration

class weblablib.WebLabNotInitializedError

Requesting a WebLab object when weblab.init_app has not been called.

class weblablib.TimeoutError

When joining (WebLabTask.join()) a task with a timeout, this error may arise

class weblablib.AlreadyRunningError

When creating a task (WebLab.task()) with unique='global' or unique='user', the second thread/process attempting to run the same method will obtain this error