snakeface package

Submodules

snakeface.argparser module

class snakeface.argparser.SnakefaceArgument(action, required=False)

Bases: object

A Snakeface argument takes an action from a parser, and is able to easily generate front end views (e.g., a form element) for it

boolean_field()

generate a boolean field (radio button) via a jinja2 template

choice_field()

generate a choice field for using a pre-loaded jinja2 template

field()

generate a form field for the argument

property field_name

property is_boolean

load_template(path)

Given a path to a template file, load the template with jinja2

text_field()

generate a text field for using a pre-loaded jinja2 template

update_choice_fields(updates)

class snakeface.argparser.SnakefaceParser

Bases: object

A Snakeface Parser is a wrapper to an argparse.Parser, and aims to make it easy to loop over arguments and options, and generate various representations (e.g., an input field) for the interface. The point is not to use it to parse arguments and validate, but to output all fields to a front end form.

property command

Given a loaded set of arguments, generate the command.

property errors

get(name, default=None)

A general get function to return an argument that might be nested under a group. These objects are the same as linked in _groups.

property groups

yield arguments organized by groups, with the intention to easily map into a form on the front end. The groups seem to have ALL arguments each, so we have to artificially separate them.

include_argument(name, group)

Given an argument name, and a group name, skip if settings disable it

load(argdict)

Load is a wrapper around set - we loop through a dictionary and set all arguments.

property required

set(name, value)

Set a value for an argument. This is typically what the user has selected.

property snakefile

snakefiles = []

to_dict()

the opposite of load, this function exports an argument

validate()

ensure that all required args are defined

snakeface.client module

snakeface.client.get_parser()

snakeface.client.main()

main entrypoint for snakeface

snakeface.apps.api module

class snakeface.apps.api.permissions.AllowAnyGet

Bases: rest_framework.permissions.BasePermission

Allows an anonymous user access for GET requests only.

has_permission(request, view)

Return True if permission is granted, False otherwise.

snakeface.apps.api.permissions.check_user_authentication(request)

Given a request, check that the user is authenticated via a token in the header.

snakeface.apps.api.permissions.get_token(request)

The same as validate_token, but return the token object to check the associated user.

class snakeface.apps.api.views.CreateWorkflow(**kwargs)

Bases: ratelimit.mixins.RatelimitMixin, rest_framework.views.APIView

Create a snakemake workflow. Given that we provide an API token, we expect the workflow model to already be created and simply generate a run for it.

get(request)

ratelimit_block = True

ratelimit_key = 'ip'

ratelimit_method = 'GET'

ratelimit_rate = '1000/1d'

renderer_classes = (<class 'rest_framework.renderers.JSONRenderer'>,)

class snakeface.apps.api.views.ServiceInfo(**kwargs)

Bases: ratelimit.mixins.RatelimitMixin, rest_framework.views.APIView

Return a 200 response to indicate a running service. Note that we are not currently including all required fields. See: https://ga4gh.github.io/workflow-execution-service-schemas/docs/#operation/GetServiceInfo

get(request)

ratelimit_block = True

ratelimit_key = 'ip'

ratelimit_method = 'GET'

ratelimit_rate = '1000/1d'

renderer_classes = (<class 'rest_framework.renderers.JSONRenderer'>,)

class snakeface.apps.api.views.UpdateWorkflow(**kwargs)

Bases: ratelimit.mixins.RatelimitMixin, rest_framework.views.APIView

Update an existing snakemake workflow. Authentication is required, and the workflow must exist.

post(request)

ratelimit_block = True

ratelimit_key = 'ip'

ratelimit_method = 'POST'

ratelimit_rate = '1000/1d'

renderer_classes = (<class 'rest_framework.renderers.JSONRenderer'>,)

snakeface.apps.main module

class snakeface.apps.main.consumers.WorkflowConsumer(*args, **kwargs)

Bases: channels.generic.websocket.AsyncJsonWebsocketConsumer

async connect()

async disconnect(close_code)

Called when a WebSocket connection is closed.

async receive(text_data)

Called with a decoded WebSocket frame.

async update_workflow_status()

snakeface.apps.main.consumers.async_get_statuses(workflow_id)

Return a dictionary of workflow statuses on success. If the workflow doesn’t exist, then return False and we disconnect from the socket.

snakeface.apps.main.consumers.get_statuses(workflow_id)

Return a dictionary of workflow statuses on success. If the workflow doesn’t exist, then return False and we disconnect from the socket.

class snakeface.apps.main.forms.WorkflowForm(data=None, files=None, auto_id='id_%s', prefix=None, initial=None, error_class=<class 'django.forms.utils.ErrorList'>, label_suffix=None, empty_permitted=False, instance=None, use_required_attribute=None, renderer=None)

Bases: django.forms.models.ModelForm

class Meta

Bases: object

fields = ['name', 'workdirs']

model

alias of snakeface.apps.main.models.Workflow

base_fields = {'name': <django.forms.fields.CharField object>, 'workdirs': <django.forms.fields.ChoiceField object>}

declared_fields = {'workdirs': <django.forms.fields.ChoiceField object>}

property media

Return all media required to render the widgets on this form.

class snakeface.apps.main.models.JSONField(verbose_name=None, name=None, primary_key=False, max_length=None, unique=False, blank=False, null=False, db_index=False, rel=None, default=<class 'django.db.models.fields.NOT_PROVIDED'>, editable=True, serialize=True, unique_for_date=None, unique_for_month=None, unique_for_year=None, choices=None, help_text='', db_column=None, db_tablespace=None, auto_created=False, validators=(), error_messages=None)

Bases: django.db.models.fields.Field

db_type(connection)

Return the database column data type for this field, for the provided connection.

from_db_value(value, expression, connection)

get_prep_value(value)

Perform preliminary non-db specific value checks and conversions.

to_python(value)

Convert the input value into the expected Python data type, raising django.core.exceptions.ValidationError if the data can’t be converted. Return the converted value. Subclasses should override this.

value_to_string(obj)

Return a string value of this field from the passed obj. This is used by the serialization framework.

class snakeface.apps.main.models.Workflow(*args, **kwargs)

Bases: django.db.models.base.Model

A workflow is associated with a specific git repository and one or more workflow runs.

exception DoesNotExist

Bases: django.core.exceptions.ObjectDoesNotExist

exception MultipleObjectsReturned

Bases: django.core.exceptions.MultipleObjectsReturned

add_date

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

command

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

contributors

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example:

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

dag

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

data

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

error

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_absolute_url()

get_label()

get_next_by_add_date(*, field=<django.db.models.fields.DateTimeField: add_date>, is_next=True, **kwargs)

get_next_by_modify_date(*, field=<django.db.models.fields.DateTimeField: modify_date>, is_next=True, **kwargs)

get_previous_by_add_date(*, field=<django.db.models.fields.DateTimeField: add_date>, is_next=False, **kwargs)

get_previous_by_modify_date(*, field=<django.db.models.fields.DateTimeField: modify_date>, is_next=False, **kwargs)

get_private_display(*, field=<django.db.models.fields.BooleanField: private>)

get_report()

load the report file, if it exists.

get_status_display(*, field=<django.db.models.fields.TextField: status>)

has_edit_permission()

If we are running in a notebook environment, there is just one user that has edit access to anything. Otherwise, the user must be an owner

has_report()

returns True if the workflow command has a designated report, and the report file exists

has_view_permission()

id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

property members

property message_fields

modify_date

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

name

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>

output

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

owners

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example:

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

private

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

reset()

Empty all run related fields to prepare for a new run.

retval

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

snakefile

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

snakemake_id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

status

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

thread

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

update_command(command=None, do_save=False)

Given a command (or an automated save from the signal) update the command for the workflow.

update_dag(do_save=False)

given a snakefile, run the command to update the dag

workdir

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

workflowstatus_set

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

class snakeface.apps.main.models.WorkflowStatus(*args, **kwargs)

Bases: django.db.models.base.Model

A workflow status is a status message send from running a workflow

exception DoesNotExist

Bases: django.core.exceptions.ObjectDoesNotExist

exception MultipleObjectsReturned

Bases: django.core.exceptions.MultipleObjectsReturned

add_date

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_next_by_add_date(*, field=<django.db.models.fields.DateTimeField: add_date>, is_next=True, **kwargs)

get_next_by_modify_date(*, field=<django.db.models.fields.DateTimeField: modify_date>, is_next=True, **kwargs)

get_previous_by_add_date(*, field=<django.db.models.fields.DateTimeField: add_date>, is_next=False, **kwargs)

get_previous_by_modify_date(*, field=<django.db.models.fields.DateTimeField: modify_date>, is_next=False, **kwargs)

id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

modify_date

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

msg

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>

workflow

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

workflow_id

snakeface.apps.main.models.update_workflow(sender, instance, **kwargs)

snakeface.apps.main.tasks.doRun(wid, uid)

The task to run a workflow

snakeface.apps.main.tasks.run_is_allowed(request)

Given a request, check that the run is allowed meaning: 1. If running a notebook, we aren’t over quota for jobs 2. If not running a notebook, we aren’t over user or global limits

snakeface.apps.main.tasks.run_workflow(request, wid, uid)

Top level function to ensure that the user has permission to do the run, and we direct to the correct function (notebook or not written, another backend)

snakeface.apps.main.tasks.serialize_workflow_statuses(workflow)

A shared helper function to serialize a list of workflow statuses into json.

class snakeface.apps.main.utils.CommandRunner

Bases: object

Wrapper to use subprocess to run a command. This is based off of pypi vendor distlib SubprocesMixin.

reader(stream, context)

Get output and error lines and save to command runner.

reset()

run_command(cmd, env=None, cancel_func=None, cancel_func_kwargs=None, **kwargs)

class snakeface.apps.main.utils.ThreadRunner(group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None)

Bases: threading.Thread

We need to be able to run a Snakemake job as a thread, and kill it if an exception is raised based on it’s id

set_workflow(workflow)

property thread_id

Return the id of the thread, either attributed to the class or by matching the Thread instance

snakeface.apps.main.utils.get_snakefile_choices(path=None)

Given the working directory set on init, return all discovered snakefiles.

snakeface.apps.main.utils.get_tmpfile(prefix='', suffix='')

get a temporary file with an optional prefix. By default, the file is closed (and just a name returned).

Parameters:prefix (-) – prefix with this string

snakeface.apps.main.utils.get_workdir_choices(path=None)

Given the working directory set on init, return potential subdirectories.

snakeface.apps.main.utils.read_file(filename)

Write some text content to a file

snakeface.apps.main.utils.write_file(filename, content)

Write some text content to a file

snakeface.apps.main.views.edit_or_update_workflow(request, parser, workflow=None)

A shared function to edit or update an existing workflow.

snakeface.apps.main.views.view_workflow_report(request, wid)

If a workflow generated a report and the report exists, render it to a page

snakeface.apps.base module

snakeface.apps.base.views.warmup()

snakeface.apps.users module

snakeface.apps.users.decorators.login_is_required(function=None, login_url=None, redirect_field_name='next')

Decorator to extend login required to also check if a notebook auth is desired first.

class snakeface.apps.users.forms.TokenForm(data=None, files=None, auto_id='id_%s', prefix=None, initial=None, error_class=<class 'django.forms.utils.ErrorList'>, label_suffix=None, empty_permitted=False, field_order=None, use_required_attribute=None, renderer=None)

Bases: django.forms.forms.Form

base_fields = {'token': <django.forms.fields.CharField object>}

declared_fields = {'token': <django.forms.fields.CharField object>}

property media

Return all media required to render the widgets on this form.

class snakeface.apps.users.models.CustomUserManager(*args, **kwargs)

Bases: django.contrib.auth.base_user.BaseUserManager

add_staff(user)

Intended for existing user

add_superuser(user)

Intended for existing user

create_superuser(username, email, password, **extra_fields)

create_user(username, email=None, password=None, **extra_fields)

class snakeface.apps.users.models.User(id, password, last_login, is_superuser, username, first_name, last_name, email, is_staff, is_active, date_joined, active, agree_terms, agree_terms_date, notebook_token)

Bases: django.contrib.auth.models.AbstractUser

exception DoesNotExist

Bases: django.core.exceptions.ObjectDoesNotExist

exception MultipleObjectsReturned

Bases: django.core.exceptions.MultipleObjectsReturned

active

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

agree_terms

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

agree_terms_date

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

auth_token

Accessor to the related object on the reverse side of a one-to-one relation.

In the example:

class Restaurant(Model):
    place = OneToOneField(Place, related_name='restaurant')

Place.restaurant is a ReverseOneToOneDescriptor instance.

get_credentials(provider)

return one or more credentials, or None

get_label()

get_next_by_date_joined(*, field=<django.db.models.fields.DateTimeField: date_joined>, is_next=True, **kwargs)

get_previous_by_date_joined(*, field=<django.db.models.fields.DateTimeField: date_joined>, is_next=False, **kwargs)

get_providers()

return a list of providers that the user has credentials for.

groups

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example:

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

has_create_permission()

has create permission determines if the user (globally) can create new collections. By default, superusers and admin can, along with regular users if USER_COLLECTIONS is True. Otherwise, not.

id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

logentry_set

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

notebook_token

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <snakeface.apps.users.models.CustomUserManager object>

social_auth

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

property token

The user token is for interaction with creating and updating workflows

user_permissions

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example:

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

workflow_contributors

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example:

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

workflow_owners

Accessor to the related objects manager on the forward and reverse sides of a many-to-many relation.

In the example:

class Pizza(Model):
    toppings = ManyToManyField(Topping, related_name='pizzas')

Pizza.toppings and Topping.pizzas are ManyToManyDescriptor instances.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

snakeface.apps.users.models.create_auth_token(sender, instance=None, created=False, **kwargs)

Create a token for the user when the user is created (with oAuth2)

1. Assign user a token
2. Assign user to default group

Create a Profile instance for all newly created User instances. We only run on user creation to avoid having to check for existence on each call to User.save.

snakeface.apps.users.utils.get_notebook_token(request, verbose=True)

If a notebook token isn’t defined, generate it (and print to the console) The token is used to generate a user to log the user in.

snakeface.apps.users.utils.get_notebook_user()

Get the notebook user, if they have logged in before.

snakeface.apps.users.utils.get_or_create_notebook_user(token)

Get or create the notebook user. Imports are done in the function because Django startup (settings.py) uses these functions.

snakeface.apps.users.utils.get_username()

get the username based on the effective uid. This is for a notebook execution, and doesn’t add any additional security, but rather is used for personalization and being able to create an associated django user.

snakeface.apps.users.views.notebook_login(request)

Given the user doesn’t have a token in the request session, ask for it.

snakeface.settings module

class snakeface.settings.Settings(dictionary)

Bases: object

convert a dictionary of settings (from yaml) into a class

snakeface.settings.generate_secret_key(filename)

A helper function to write a randomly generated secret key to file

snakeface.logger module

class snakeface.logger.ColorizingStreamHandler(nocolor=False, stream=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, use_threads=False)

Bases: logging.StreamHandler

BLACK = 0

BLUE = 4

BOLD_SEQ = '\x1b[1m'

COLOR_SEQ = '\x1b[%dm'

CYAN = 6

GREEN = 2

MAGENTA = 5

RED = 1

RESET_SEQ = '\x1b[0m'

WHITE = 7

YELLOW = 3

can_color_tty()

colors = {'CRITICAL': 1, 'DEBUG': 4, 'ERROR': 1, 'INFO': 2, 'WARNING': 3}

decorate(record)

emit(record)

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

property is_tty

class snakeface.logger.Logger

Bases: object

cleanup()

debug(msg)

error(msg)

exit(msg, return_code=1)

handler(msg)

info(msg)

location(msg)

progress(done=None, total=None)

set_level(level)

set_stream_handler(stream_handler)

shellcmd(msg)

text_handler(msg)

The default snakemake log handler. Prints the output to the console. :param msg: the log message dictionary :type msg: dict

warning(msg)

snakeface.logger.setup_logger(quiet=False, printshellcmds=False, nocolor=False, stdout=False, debug=False, use_threads=False, wms_monitor=None)