utils.py

General helper functions.

class utils.CommitResult(log, data, ins_id)

Bases: tuple

data

Alias for field number 1

ins_id

Alias for field number 2

log

Alias for field number 0

class utils.ValidationResult(result, status)

Bases: tuple

result

Alias for field number 0

status

Alias for field number 1

utils.basic_check_indata(indata: dict, reference_data: dict, prohibited: Union[tuple, list]) → tuple[source]

Perform basic checks of indata.

  • All fields are allowed in the entity type

  • If title is a field for the entity, it may not be empty

  • All fields are of the correct type

  • All prohibited fields are unchanged (if update)

Parameters

  • indata (dict) – The incoming data.

  • reference_data (dict) – Either the old data or a reference dict.

  • prohibited (Union[tuple, list]) – Fields that may not be modified. If they are included in indata, their values must be equal to the values in reference_data. Defaults to None.

Returns

(bool: whether the check passed, code: Suggested http code)

Return type

namedtuple

utils.check_email_uuid(user_identifier: str) → Union[str, uuid.UUID][source]

Check if the provided user is found in the db as email or _id.

If the user is found, return the user.UUID of the user. If the identifier is a uuid, return a user.UUID. If the identifier is an email, return the email.

Notes

user_identifier is assumed to be either a valid email or a valid uuid.

Parameters

user_identifier (str) – The identifier to look up.

Returns

The new value for the field.

Return type

Union[str, uuid.UUID]

utils.check_permissions(permissions: list, user_permissions: list, logged_in: bool) → int[source]

Perform the standard permissions check for a request.

Will return a status code: * 200: accepted * 401: not logged in * 403: permission missing

Parameters

  • permissions (list) – The required permissions.

  • user_permissions (list) – List of permissions for the user.

  • logged_in (bool) – Whether the current user is logged in.

Returns

The suggested status code.

Return type

int

utils.commit_to_db(db, dbcollection: str, operation: str, data: dict, logger=None)[source]

Commit to one entry in the database.

_id should be included in data for delete and update operations.

Only uses <type>_one commands for the db.

Parameters

  • db – Connection to the database (client).

  • dbcollection (str) – Name of the target collection.

  • operation (str) – Operation to perform (add, edit, delete).

  • data (dict) – Data to commit to db.

  • id (dict) – The entry to perform the operation on (_id).

  • logger – The logging object to use for errors.

Raises

ValueError – Missing _id in data for delete or update, or bad operation type.

Returns

The response from the db commit.

Return type

dict

utils.convert_keys_to_camel(chunk: Any) → Any[source]

Convert keys given in snake_case to camelCase.

The capitalization of the first letter is preserved.

Parameters

chunk (Any) – Object to convert.

Returns

Chunk converted to camelCase dict, otherwise chunk.

Return type

Any

utils.gen_api_key()[source]

Generate an API key with salt.

Returns

The API key with salt.

Return type

APIkey

utils.gen_api_key_hash(api_key: str, salt: str)[source]

Generate a hash of the api_key for storing/comparing to db.

Parameters

  • api_key (str) – The cleartext API key (hex).

  • salt (str) – The salt to use (hex).

Returns

SHA512 hash as hex.

Return type

str

utils.gen_csrf_token() → str[source]

Generate a csrf token.

Returns

The csrf token.

Return type

str

utils.get_db(dbserver: pymongo.mongo_client.MongoClient, conf) → pymongo.database.Database[source]

Get the connection to the MongoDB database.

Parameters

  • dbserver (pymongo.mongo_client.MongoClient) – Connection to the database.

  • conf – A mapping with the relevant mongo keys available.

Returns

The database connection.

Return type

pymongo.database.Database

utils.get_dbclient(conf) → pymongo.mongo_client.MongoClient[source]

Get the connection to the MongoDB database server.

Parameters

conf – A mapping with the relevant mongo keys available.

Returns

The client connection.

Return type

pymongo.mongo_client.MongoClient

utils.get_entry(db, dbcollection: str, identifier: str) → dict[source]

Confirm that the identifier is valid and, if so, return the entry.

Parameters

  • db – Connection to the database (client).

  • dbcollection (str) – Name of the target collection.

  • operation (str) – Operation to perform (add, edit, delete)

  • data (dict) – Data to commit to db.

  • id (dict) – The entry to perform the operation on (_id).

  • logger – The logging object to use for errors.

Raises

ValueError – Missing _id in data for delete or update.

Returns

The response from the db commit.

Return type

dict

utils.has_permission(permission: str, user_permissions: list)[source]

Check if the current user permissions fulfills the requirement.

Parameters

  • permission (str) – The required permission

  • user_permissions (list) – List of permissions for the user. Should be flask.g.permissions for most requests.

Returns

whether the user has the required permissions or not

Return type

bool

utils.incremental_logs(logs: list)[source]

Make an incremental log.

The log starts from the first log and keeps only the changed fields in data.

logs is changed in-place.

utils.is_email(indata: str)[source]

Check whether a string seems to be an email address or not.

Will not do thorough checking, just a basic check that the basic components are there.

Parameters

indata (str) – Data to check.

Returns

Is the indata an email address or not.

Return type

bool

utils.list_to_uuid(uuids: list) → list[source]

Convert the uuids in a list to uuid.UUID.

Parameters

uuids (list) – The uuid to be converted.

Returns

All the provided uuids as uuid.UUID.

Return type

list

utils.make_log(data_type: str, action: str, comment: str, data: Optional[dict] = None, no_user: bool = False, dbsession=None)[source]

Log a change in the system.

Saves a complete copy of the new object.

Warning

It is assumed that all values are exactly like in the db, e.g. data should only contain permitted fields.

Parameters

  • action (str) – Type of action (add, edit, delete).

  • comment (str) – Note about why the change was done (e.g. “Dataset added via addDataset”).

  • data_type (str) – The collection name.

  • data (dict) – The new data for the entry.

  • no_user (bool) – Whether the entry should be accredited to “system”.

  • dbsession – The MongoDB session used.

Returns

Whether the log insertion successed.

Return type

bool

utils.make_log_new(db, data_type: str, action: str, comment: str, user_id, data: dict, logger=None) → bool[source]

Log a change in the system.

Saves a complete copy of the new object.

Warning

It is assumed that all values are exactly like in the db, e.g. data should only contain permitted fields.

Parameters

  • db – Connection to the database (client).

  • data_type (str) – The collection name.

  • action (str) – Type of action (add, edit, delete).

  • comment (str) – Note about why the change was done (e.g. “Dataset added via addDataset”).

  • user_id (uuid.UUID) – The _id for the user performing the operation.

  • data (dict) – The new data for the entry.

  • logger – The logging object to use for errors.

Raises

ValueError – No data provided.

Returns

Whether the log insertion succeeded.

Return type

bool

utils.make_timestamp()[source]

Generate a timestamp of the current time.

Returns

The current time.

Return type

datetime.datetime

utils.new_uuid() → uuid.UUID[source]

Generate a uuid for a field in a MongoDB document.

Returns

The new uuid in binary format.

Return type

uuid.UUID

utils.prepare_for_db(data: dict) → dict[source]

Prepare incoming data for the database.

  • Convert string UUIDS to uuid.UUID

  • Escape html in description

Parameters

data (dict) – The incoming data.

Returns

The prepared data.

Return type

dict

utils.prepare_permissions(in_perms: list) → list[source]

Generate a full list of permissions for the user.

Parameters

in_perms (list) – The raw list of permissions from the user entry.

Returns

The complete list of permissions for the user.

Return type

list

utils.prepare_response(data: dict, url: str = '')[source]

Prepare the fields before running jsonify.

data is modified in-place.

  • Rename _id to id

  • If available, add origin URL to the response

Parameters

data (dict) – Structure to prepare.

utils.req_check_permissions(permissions)[source]

Call check_permissions from inside a Flask request.

Convenience function to use the Flask variables.

utils.req_commit_to_db(dbcollection: str, operation: str, data: Optional[dict] = None, comment: str = '') → bool[source]

Commit to one entry in the database from a Flask request.

Data should contain {_id: uuid}} if there is a deletion.

Parameters

  • dbcollection (str) – Name of the target database collection.

  • operation (str) – Operation to perform (add, edit, delete).

  • data (dict) – Data to commit to db.

  • comment (str) – Custom comment for the log.

utils.req_get_entry(dbcollection: str, identifier: str) → dict[source]

Confirm that the identifier is valid and, if so, return the entry.

Wrapper for usage from a Flask request.

Parameters

  • dbcollection (str) – The database collection to use (e.g. collections).

  • identifier (str) – The provided identifier.

Returns

The entry from the database. None if not found.

Return type

dict

utils.req_has_permission(permission: str)[source]

Check if the current user permissions fulfills the requirement.

Parameters

  • permission (str) – The required permission

  • user_permissions (list) – List of permissions for the user. Should be flask.g.permissions for most requests.

Returns

whether the user has the required permissions or not

Return type

bool

utils.req_make_log_new(data_type: str, action: str, comment: str, data: dict) → bool[source]

Log a change in the system.

Wrapper for Flask requests.

Saves a complete copy of the new object.

Warning

It is assumed that all values are exactly like in the db, e.g. data should only contain permitted fields.

Parameters

  • data_type (str) – The collection name.

  • action (str) – Type of action (add, edit, delete).

  • comment (str) – Note about why the change was done (e.g. “Dataset added via addDataset”).

  • data (dict) – The new data for the entry.

Returns

Whether the log insertion succeeded.

Return type

bool

utils.response_json(data: dict)[source]

Prepare a json response from the provided data.

Parameters

date (dict) – Structure to make into a respone.

Returns

Prepared response containing json structure with camelBack keys.

Return type

flask.Response

utils.secure_description(data: str)[source]

Process the description to make sure it does not contain dangerous data.

Current checks: * Escape HTML

Parameters

data – The description to process.

Returns

The processed description.

Return type

str

utils.str_to_uuid(in_uuid: Union[str, uuid.UUID]) → uuid.UUID[source]

Convert str uuid to uuid.UUID.

Provided as a convenience function if the identifier must be changed in the future.

Parameters

in_uuid (str or uuid.UUID) – The uuid to be converted.

Returns

The uuid as a UUID object.

Return type

uuid.UUID

utils.user_uuid_data(user_ids: Union[str, list, uuid.UUID], mongodb: pymongo.database.Database) → list[source]

Retrieve some extra information about a user using a uuid as input.

Note that _id` will be returned as str, not uuid.UUID.

Parameters

  • user_ids (str, list, or uuid.UUID) – UUID of the user(s).

  • mongodb (pymongo.database.Database) – The Mongo database to use for the query.

Returns

The matching entries.

Return type

list

utils.verify_api_key(username: str, api_key: str)[source]

Verify an API key against the value in the database.

Aborts with status 401 if the verification fails.

Parameters

  • username (str) – The username to check.

  • api_key (str) – The received API key (hex).

utils.verify_csrf_token()[source]

Compare the csrf token from the request (header) with the one in cookie.session.

Parameters

request – The Flask request.

Aborts with status 400 if the verification fails.