quetzal.app.helpers package

Submodules

quetzal.app.helpers.celery module

Improved version of the celery object of flask_celery_helper package

The original flask_celery.Celery object has a bug where the context is pushed during unit tests. This file extends and rewrites the related code to avoid this bug.

This bug has been documented on: https://github.com/Robpol86/Flask-Celery-Helper/issues/23

class quetzal.app.helpers.celery.Celery(app=None)

Bases: flask_celery.Celery

init_app(app)

Actual method to read celery settings from app configuration and initialize the celery instance.

Positional arguments: app – Flask application instance.

quetzal.app.helpers.celery._mockable_call(base, obj, *args, **kwargs)

Helper function to replace Task.__call__ for mockable tests

quetzal.app.helpers.celery.log_task(task, level=20, limit=10, _logger=None)

Log the ids of a task or chain of tasks in celery

quetzal.app.helpers.files module

quetzal.app.helpers.files.get_readable_info(file_obj)

Extract useful information from reading a file

This function calculates the md5sum and the file size in bytes from a file-like object. It does both operations at the same time, which means that there is no need to read the object twice.

After this function reads the file content, it will set the file pointer to its original position through tell.

Parameters:file_obj (file-like) – File object. It needs the read and tell methods.
Returns:md5sum, size – MD5 sum and size of the file object contents
Return type:str, int
quetzal.app.helpers.files.split_check_path(filepath)

quetzal.app.helpers.google_api module

quetzal.app.helpers.google_api.get_bucket(url, *, client=None)

Get a GCP bucket object from an URL

Parameters:
  • url (str) – URL of the bucket
  • client (google.storage.client.Client, optional) – GCP client instance to use. If not set it uses get_client().
Returns:

bucket – A bucket instance
Return type:

google.storage.bucket.Bucket
quetzal.app.helpers.google_api.get_client()

Create a GCP client built from the app configuration

The client is saved in the currrent application context and will be reused in any future call on this context.

quetzal.app.helpers.google_api.get_data_bucket(*, client=None)

Get Quetzal’s data bucket

Parameters:client (google.storage.client.Client, optional) – GCP client instance to use. If not set it uses get_client().
Returns:bucket – A bucket instance
Return type:google.storage.bucket.Bucket
quetzal.app.helpers.google_api.get_object(url, *, client=None)

quetzal.app.helpers.pagination module

class quetzal.app.helpers.pagination.CustomPagination(*args, **kwargs)

Bases: flask_sqlalchemy.Pagination

A specialization of flask_sqlalchemy pagination object

This specialization adds a utility method to produce a dictionary that conforms to Quetzal’s pagination specification.

In addition to the original constructor parameters, this object takes a serializer method that converts whatever object type that the query from the paginate call generates and converts it to an object that must be JSON serializable. If not provided, the object is used as-is.

next(error_out=False)

Returns a Pagination object for the next page.

prev(error_out=False)

Returns a Pagination object for the previous page.

response_object()
quetzal.app.helpers.pagination.paginate(queriable, *, page=None, per_page=None, error_out=True, max_per_page=None, serializer=None)

Returns per_page items from page page.

This is a specialization of flask_sqlalchemy.BaseQuery.paginate with some custom modifications:

  • It changes the original behavior to respond throw APIException instead of calling abort. The status code has also been changed to 400 instead of 404. Normally, this errors should not be reachable since connexion handles the input validation.
  • In addition to handling regular flask_sqlalchemy.BaseQuery objects, it can also accept a cursor.
  • In addition to these changes, this function returns a custom pagination object that provides a response_object method that can build a response according to Quetzal’s paginated response specification.
  • Uses keyword arguments to avoid incorrect arguments

The original docstring is as follows:

If page or per_page are None, they will be retrieved from the request query. If max_per_page is specified, per_page will be limited to that value. If there is no request or they aren’t in the query, they default to 1 and 20 respectively.

When error_out is True (default), the following rules will cause a 404 response:

  • No items are found and page is not 1.
  • page is less than 1, or per_page is negative.
  • page or per_page are not ints.

When error_out is False, page and per_page default to 1 and 20 respectively.

Returns a CustomPagination object.

quetzal.app.helpers.sql module

class quetzal.app.helpers.sql.CreateTableAs(name, query)

Bases: sqlalchemy.sql.base.Executable, sqlalchemy.sql.elements.ClauseElement

class quetzal.app.helpers.sql.DropSchemaIfExists(name, cascade=False)

Bases: sqlalchemy.sql.base.Executable, sqlalchemy.sql.elements.ClauseElement

class quetzal.app.helpers.sql.GrantUsageOnSchema(schema, user)

Bases: sqlalchemy.sql.base.Executable, sqlalchemy.sql.elements.ClauseElement

quetzal.app.helpers.sql.print_sql(qs)