Library

This section contains a complete API reference of the kadi.lib Python module.

API

Bases: Blueprint

Custom Flask blueprint with support for API versioning.

route(rule, **options)[source]

Decorator to register a view function for a given URL rule.

Adds a new option v to Flask’s route decorator to support API versioning within endpoints.

Example:

@blueprint.route("/records", v=["v1", "v2"])
def get_records():
    pass

The specified API versions all have to be valid, i.e. they have to be part of the available API versions defined in kadi.lib.constants.API_VERSIONS. If no versions are given, the endpoint defaults to all available versions. The regular endpoint without any version will be created as well, pointing to the same function as the endpoint correponding to the version defined in kadi.lib.constants.API_VERSION_DEFAULT.

The code snippet above would lead to the following endpoints and URLs, assuming one of the given versions corresponds to the default API version:

api.get_records -> /api/records
api.get_records_v1 -> /api/v1/records
api.get_records_v2 -> /api/v2/records

Parameters:

rule – The URL rule as string.
endpoint – (optional) The endpoint for the registered URL rule. Defaults to the name of the function.
v – (optional) A string or list of strings specifying the supported API versions.
**options – Additional options to be forwarded to the underlying rule system of Flask.

kadi.lib.api.core.json_response(status_code, body=None)[source]

Return a JSON response to a client.

Parameters:

status_code – The HTTP status code of the response.
body – (optional) The response body, which must be JSON serializable. Defaults to an empty dictionary.

Returns:

The JSON response.

kadi.lib.api.core.json_error_response(status_code, message=None, description=None, **kwargs)[source]

Return a JSON error response to a client.

Uses json_response() with the given headers and a body in the following form, assuming no additional error information was provided:

{
    "code": 404,
    "message": "<message>",
    "description": "<description>"
}

Parameters:

status_code – See json_response().
message – (optional) The error message. Defaults to a message corresponding to the given status code.
description – (optional) The error description. Defaults to the result of kadi.lib.web.get_error_description() using the given status code.
**kwargs – Additional error information that will be included in the response body. All values need to be JSON serializable.

Returns:

The JSON response.

kadi.lib.api.core.get_access_token()[source]

Get an access token from the current request.

Currently, this will either be a personal token or an OAuth2 server token. The token value has to be included as a Bearer token within an Authorization header.

Returns:: An access token object or None if no valid token can be found or no request context currently exists.

kadi.lib.api.core.check_access_token_scopes(*scopes)[source]

Check if the current access token contains certain scope values.

The current access token will be retrieved using utils.get_access_token().

Parameters:: *scopes – One or multiple scope values to check in the form of "<object>.<action>".
Returns:: True if the access token either contains all given scope values, has full access or the current request contains no valid access token at all, False otherwise or if the given operator is invalid.

kadi.lib.api.core.scopes_required(*scopes)[source]

Decorator to add required access token scope values to an API endpoint.

Uses check_access_token_scopes(), so the scopes are only checked if the current request actually contains a valid access token. Therefore, this decorator usually only makes sense for public API endpoints that can be accessed by using an access token.

Example:

@blueprint.route("/records")
@login_required
@scopes_required("record.read")
def get_records():
    pass

The information about the scope values is also used when generating the API documentation.

Parameters:: *scopes – See check_access_token_scopes().

kadi.lib.api.core.internal(func)[source]

Decorator to mark an API endpoint as internal.

Internal endpoints can only be accessed via the session, not via access tokens. This is not to be confused with kadi.lib.api.utils.is_internal_api_request().

The information about an endpoint being internal is also used when generating the API documentation.

kadi.lib.api.core.experimental(func)[source]

Decorator to mark an API endpoint as experimental.

Experimental endpoints can only be called if the EXPERIMENTAL_FEATURES flag in the application’s configuration is set.

The information about an endpoint being experimental is also used when generating the API documentation.

class kadi.lib.api.models.PersonalToken(**kwargs)[source]

Bases: SimpleReprMixin, AccessTokenMixin, Model

Model to represent personal tokens.

These kind of access tokens always belong to and are managed by a certain user, so they may also be referred to as personal access tokens (PAT).

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'name']: See SimpleReprMixin.

check_constraints = {'name': {'length': {'max': 150}}}: See kadi.lib.db.generate_check_constraints().

id: The ID of the personal token, auto incremented.

name

The name of the personal token.

Restricted to a maximum length of 150 characters.

token_hash: The actual, hashed token value.

expires_at: The optional date and time the personal token expires in.

scope

The optional scope of the access token.

Represented as a single string defining a list of space-delimited scope values.

user

The user relationship of the access token.

A corresponding relationship should also be defined in the user table.

user_id: The ID of the user the access token belongs to.

created_at: The date and time the personal token was created at.

last_used: The date and time the personal token was last used.

property is_expired: Check if the personal token is expired.

static new_token(include_prefix=True)[source]

Create a new random token value.

Parameters:: include_prefix – (optional) Whether to include a prefix before the actual token value to distinguish it with other types of access tokens.
Returns:: The generated token value.

static hash_token(token)[source]

Create a secure hash of a token value.

Parameters:: token – The token value to hash.
Returns:: The calculated hash as a hexadecimal value.

classmethod get_by_token(token)[source]

Get a personal token using a token value.

Parameters:: token – The token value to search for.
Returns:: The personal token or None.

classmethod create(*, user, name, scope=None, expires_at=None, token=None)[source]

Create a new personal token and add it to the database session.

Parameters:

user – The user the personal token should belong to.
name – The name of the personal token.
scope – (optional) The scope of the personal token.
expires_at – (optional) The expiration date of the personal token.
token – (optional) The actual token value, which will be hashed before persisting. Defaults to a token value created by new_token().

Returns:

The new PersonalToken object.

class kadi.lib.api.schemas.PersonalTokenSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent personal tokens.

See PersonalToken.

kadi.lib.api.utils.is_api_request()[source]

Check if the current request is an API request.

A request is an API request if the path of the current request path starts with "/api".

Returns:: True if the request is an API request, False otherwise.

kadi.lib.api.utils.is_internal_api_request()[source]

Check if the current API request is an “internal” one.

An API request is marked as internal if it includes a query parameter _internal with any value (e.g. "https://...?_internal=true"). This can be useful for e.g. returning additional data that is only relevant for internal use. Note that it does not matter whether the request uses the session or an access token.

Returns:: True if the request is internal, False otherwise.

kadi.lib.api.utils.get_api_version(default='v1')[source]

Get the API version from the current request.

Parameters:: default – (optional) The API version to return if the current request is an API request but does not contain a valid version.
Returns:: The current API version or the given default value as fallback. If the current request is not an API request at all, None will be returned.

kadi.lib.api.utils.get_access_token_scopes()[source]

Get all available access token scopes.

The available scopes are combined from all resource actions and all additional scopes defined in kadi.lib.constants.ACCESS_TOKEN_SCOPES.

Returns:: A dictionary mapping a scope’s object to a list of corresponding scopes.

kadi.lib.api.utils.create_pagination_data(total, page, per_page, endpoint=None, **kwargs)[source]

Create pagination information for use in a JSON response.

Since the pagination data will include links to the current, next and previous “pages”, the necessary information to build said links needs to be given as well, i.e. the endpoint and its corresponding URL parameters.

Parameters:

total – The total amount of items.
page – The current page.
per_page – Items per page.
endpoint – The endpoint used to build links to the current, next and previous page. Defaults to the endpoint of the current request.
**kwargs – Additional keyword arguments to build the links with.

Returns:

The pagination information as dictionary in the following form:

{
    "_pagination": {
        "page": 2,
        "per_page": 10,
        "total_pages": 3,
        "total_items": 25,
        "_links": {
            "prev": "https://...?page=1&...",
            "self": "https://...?page=2&...",
            "next": "https://...?page=3&...",
        }
    }
}

The list of items is initially empty and can be filled afterwards with whatever data should be returned. Note that the links to the previous and next pages are only present if the respective page actually exists.

kadi.lib.api.utils.status(status_code, description)[source]

Decorator to add response status information to an API endpoint.

This information is currently only used when generating the API documentation.

Parameters:

status_code – The status code of the response.
description – The description corresponding to the status code, describing when the status code occurs or whether there is a response body. Supports reST syntax.

kadi.lib.api.utils.reqschema(schema, description='', bind=True)[source]

Decorator to add request body information to an API endpoint using a schema.

This information is mainly used when generating the API documentation.

Parameters:

schema – The schema class or instance to use as base for the request body information.
description – (optional) Additional description of the request body. Supports reST syntax.
bind – (optional) Flag indicating whether the schema should also be injected into the decorated function as keyword argument schema.

kadi.lib.api.utils.reqheaders(headers)[source]

Decorator to add custom request header information to an API endpoint.

This information is currently only used when generating the API documentation.

Parameters:

headers –

The custom request header information as a dictionary in the following form:

{
    "<header>": {
        "required": True,
        # Supports reST syntax.
        "description": "",
    }
}

Cache

kadi.lib.cache.memoize_request(func)[source]

Decorator to cache a function call’s result during a request.

Uses an in-memory dictionary as cache that will be deleted again after the current request. The functions fully qualified name and arguments will be used as key to store its result for following calls.

Disabled during testing.

Config

kadi.lib.config.core.get_sys_config(key, use_fallback=True)[source]

Get the value of a global config item from the database.

This function can be used as an alternative to directly accessing the application’s configuration if a certain config item can be stored in the database as well.

Parameters:

key – The key of the config item.
use_fallback – (optional) Whether the application’s configuration should be used as a fallback if no matching key could be found in the database.

Returns:

The value of the config item or a fallback value if no matching item could be found and use_fallback is True, kadi.lib.config.core.MISSING otherwise.

kadi.lib.config.core.set_sys_config(key, value)[source]

Set the value of a global config item in the database.

Note that trying to set an existing config item to its default value, as specified in the application’s current configuration class, will instead remove this config item from the database.

Parameters:

key – The key of the config item.
value – The value of the config item, which needs to be JSON serializable.

Returns:

The created or updated config item or None if either the given key does not exist in the application’s current configuration class or the given value matches the default value of the corresponding key.

kadi.lib.config.core.remove_sys_config(key)[source]

Remove a global config item from the database.

Parameters:: key – The key of the config item.
Returns:: True if the config item was deleted successfully, False if no such item exists.

kadi.lib.config.core.get_user_config(key, user=None, default=<kadi.lib.config.core.MISSING>, decrypt=False)[source]

Get the value of a user-specific config item from the database.

Parameters:

key – The key of the config item.
user – (optional) The user the config item belongs to. Defaults to the current user.
default – (optional) The value to return if no config item was found. Defaults to kadi.lib.config.core.MISSING.
decrypt – (optional) Flag indicating whether the value of the config item should be decrypted.

Returns:

The value of the config item or the default value if either no matching item could be found or if decrypt is True and the value could not be decrypted.

kadi.lib.config.core.set_user_config(key, value, user=None, encrypt=False)[source]

Set the value of a user-specific config item in the database.

Parameters:

key – The key of the config item.
value – The value of the config item, which needs to be JSON serializable.
user – (optional) The user the config item belongs to. Defaults to the current user.
encrypt – (optional) Flag indicating whether the value of the config item should be encrypted.

Returns:

The created or updated config item.

kadi.lib.config.core.remove_user_config(key, user=None)[source]

Remove a user-specific config item from the database.

Parameters:

key – The key of the config item.
user – (optional) The user the config item belongs to. Defaults to the current user.

Returns:

True if the config item was deleted successfully, False if no such item exists.

class kadi.lib.config.models.ConfigItem(**kwargs)[source]

Bases: SimpleReprMixin, SimpleTimestampMixin, Model

Model to store global or user-specific config items.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'key', 'user_id']: See SimpleReprMixin.

id: The ID of the config item, auto incremented.

key: The key of the config item.

value: The value of the config item.

created_at

The date and time an object has been created at.

Always uses the current UTC time.

last_modified

The date and time an object was last modified.

Always uses the current UTC time as initial value. After calling register_timestamp_listener() for an inheriting mixin class, this timestamp can automatically get updated by implementing _before_flush_timestamp() as a class method.

user_id

The optional ID of the User the config item belongs to.

If not set, the config item is global.

classmethod create(*, key, value, user=None)[source]

Create a new config item and add it to the database session.

Parameters:

key – The key of the config item.
value – The value of the config item, which needs to be JSON serializable.
user – (optional) The user the config item belongs to.

Returns:

The new ConfigItem object.

classmethod update_or_create(*, key, value, user=None)[source]

Update an existing config item or create one if it does not exist yet.

See create() for an explanation of the parameters.

Returns:: The new or updated ConfigItem object.

Conversion

kadi.lib.conversion.strip(value)[source]

Strip all surrounding whitespaces in a string.

Parameters:: value – The string to strip.
Returns:: The stripped string or the original input if it was not a string.

kadi.lib.conversion.normalize(value)[source]

Normalize all and strip surrounding whitespaces in a string.

Parameters:: value – The string to normalize.
Returns:: The normalized string or the original input if it was not a string.

kadi.lib.conversion.normalize_uri(value)[source]

Normalize a URI by lowercasing its domain name portion.

Parameters:: value – The URI to normalize as string.
Returns:: The normalized URI or the original input if it was not a string.

kadi.lib.conversion.lower(value)[source]

Lowercase all characters in a string.

Parameters:: value – The string to lowercase.
Returns:: The lowercased string or the original input if it was not a string.

kadi.lib.conversion.truncate(value, length)[source]

Truncate a string based on a given length.

Parameters:

value – The string to truncate.
length – The maximum length of the string.

Returns:

The truncated string or the original input if it was not a string.

kadi.lib.conversion.recode(value, from_encoding='utf-8', to_encoding='utf-8')[source]

Change the encoding of a string.

Parameters:

value – The string value.
from_encoding – (optional) The original encoding of the string.
to_encoding – (optional) The target encoding of the string.

Returns:

The newly encoded string or the original input if it was not a string or the recoding failed.

kadi.lib.conversion.clamp(value, min_value, max_value)[source]

Clamp a numeric value to the inclusive range of the given min and max values.

Parameters:

min_value – The minumum value.
max_value – The maximum value.

Returns:

The clamped value or the original input if it was not a numeric value.

kadi.lib.conversion.none(value)[source]

Return None if a given value is falsy.

Parameters:: value – A value to check for truthness.
Returns:: The unmodified value or None if it is falsy.

kadi.lib.conversion.empty_str(value)[source]

Return an empty string if a given value is None.

Parameters:: value – A value to check for being None.
Returns:: The unmodified value or an empty string if it is None.

kadi.lib.conversion.markdown_to_html(value)[source]

Render a markdown string as HTML.

Note that manually entered HTML will be left intact, as it will be escaped accordingly.

Parameters:: value – The string to render.
Returns:: The rendered string or the original input if it was not a string or could not be rendered.

kadi.lib.conversion.strip_markdown(value)[source]

Strip a string of its markdown directives and normalize its whitespaces.

Note that not all directives may be stripped, since some may not be supported by the markdown renderer used in markdown_to_html().

Parameters:: value – The string to strip.
Returns:: The stripped string or the original input if it was not a string.

kadi.lib.conversion.parse_datetime_string(value)[source]

Parse a datetime string.

Parameters:: value – The datetime string to parse in ISO 8601 format.
Returns:: A timezone aware datetime object in UTC as specified in Python’s datetime module or None if the given string was not a valid datetime string.

kadi.lib.conversion.parse_boolean_string(value)[source]

Parse a boolean string.

The given string is parsed based on typical values used for thruthness, including True, "true", "t", "yes", "y", "on" and "1" (case insensitive for all string values), instead of using Python’s bool conversion. All other values are considered false.

Parameters:: value – The boolean string to parse.
Returns:: True if the given string is considered truthy, False otherwise.

kadi.lib.conversion.parse_json_object(value)[source]

Parse a JSON object string as a dictionary.

Parameters:: value – The JSON object string to parse.
Returns:: The parsed dictionary or an empty dictionary if the given string was not a valid JSON object.

Database

class kadi.lib.db.KadiAesEngine[source]

Bases: AesEngine

Custom AES engine for encrypting and decrypting database values.

static get_secret_key()[source]

Get the secret key to use for encryption.

Note that this secret key is the same SECRET_KEY Flask uses as well, as specified in the application’s configuration. If it ever changes, all values encrypted with this key will become unreadable.

Returns:: The secret key.

classmethod create()[source]

Create a new AES engine with default configuration.

Convenience method to use the AES engine outside of an ORM context.

Returns:: The created AES engine.

decrypt(value)[source]

Try to decrypt the given value.

Parameters:: value – The value to decrypt.
Returns:: The decrypted value.
Raises:: KadiDecryptionKeyError – If the key used for decrypting the value is invalid.

class kadi.lib.db.UTCDateTime(*args, **kwargs)[source]

Bases: TypeDecorator

Custom timezone aware DateTime type using UTC.

As dates are currently saved without timezone information (and always interpreted as UTC), the timezone information has to be removed from datetime objects before persisting, as otherwise they are converted to local time. When retrieving the value, the timezone will be added back in.

impl: alias of DateTime

process_bind_param(value, dialect)[source]: Convert to UTC and then remove the timezone.

process_result_value(value, dialect)[source]: Replace the missing timezone with UTC.

class kadi.lib.db.BaseTimestampMixin[source]

Bases: object

Base mixin class for SQLAlchemy models to add timestamp columns.

In all current implementations, changes in columns and relationship-based collections can be ignored by specifying the Meta.timestamp_exclude attribute in the inheriting class. It should be a list of strings specifying the attribute names to exclude.

Example:

class Foo:
    class Meta:
        timestamp_exclude = ["bar", "baz"]

created_at = Column(None, UTCDateTime(), table=None, nullable=False, default=ColumnDefault(<function utcnow>))

The date and time an object has been created at.

Always uses the current UTC time.

last_modified = Column(None, UTCDateTime(), table=None, nullable=False, default=ColumnDefault(<function utcnow>))

The date and time an object was last modified.

Always uses the current UTC time as initial value. After calling register_timestamp_listener() for an inheriting mixin class, this timestamp can automatically get updated by implementing _before_flush_timestamp() as a class method.

classmethod register_timestamp_listener()[source]

Register a listener to automatically update the last modification timestamp.

Uses SQLAlchemy’s before_flush event and propagates to all inheriting models.

update_timestamp()[source]: Manually trigger an update to the last modification timestamp.

class kadi.lib.db.SimpleTimestampMixin[source]

Bases: BaseTimestampMixin

Timestamp mixin class which triggers on all changes.

class kadi.lib.db.StateTimestampMixin[source]

Bases: BaseTimestampMixin

Timestamp mixin class which only triggers on changes in active objects.

An object is considered active if its marked as active via its state attribute, which needs to be present in an inheriting model. Besides the object itself, changes in relationship-based collections consisting of inactive objects only are also ignored.

update_timestamp()[source]

Manually trigger an update to the last modification timestamp.

Only actually triggers the update if the object is marked as active.

class kadi.lib.db.NestedTransaction(exc=<class 'sqlalchemy.exc.SQLAlchemyError'>)[source]

Bases: object

Context manager to start a “nested” transaction.

The nested transaction uses the SAVEPOINT feature of the database, which will be triggered when entering the context manager. Once the context manager exits, the savepoint is released, which does not actually persist the changes done in the savepoint yet. If the release produces the given exception, the savepoint will be rolled back automatically. The success attribute of the transaction can be used to check the result of the release operation afterwards.

Parameters:: exc – (optional) The exception to catch when releasing the savepoint.

property success

Get the status of the nested transaction once the savepoint is released.

Will always be False before the context manager exits.

kadi.lib.db.update_object(obj, **kwargs)[source]

Convenience function to update database objects.

Only columns (i.e. attributes) that actually exist will get updated.

Parameters:

obj – The object to update.
**kwargs – The columns to update and their respective values.

kadi.lib.db.composite_index(tablename, *cols)[source]

Generate a composite index.

Parameters:

tablename – The name of the table.
*cols – The names of the columns.

Returns:

The Index instance.

kadi.lib.db.unique_constraint(tablename, *cols)[source]

Generate a unique constraint.

Parameters:

tablename – The name of the table.
*cols – The names of the columns.

Returns:

The UniqueConstraint instance.

kadi.lib.db.check_constraint(constraint, name)[source]

Generate a check constraint.

Parameters:

constraint – The constraint expression as string.
name – The name of the constraint.

Returns:

The CheckConstraint instance.

kadi.lib.db.length_constraint(col, min_value=None, max_value=None)[source]

Generate a length check constraint for a column.

Parameters:

col – The name of the column.
min_value – (optional) Minimum length.
max_value – (optional) Maximum length.

Returns:

The CheckConstraint instance.

kadi.lib.db.range_constraint(col, min_value=None, max_value=None)[source]

Generate a range check constraint for a column.

Parameters:

col – The name of the column.
min_value – (optional) Minimum value.
max_value – (optional) Maximum value.

Returns:

The CheckConstraint instance.

kadi.lib.db.values_constraint(col, values)[source]

Generate a values check constraint for a column.

Parameters:

col – The name of the column.
values – List of values.

Returns:

The CheckConstraint instance.

kadi.lib.db.generate_check_constraints(constraints)[source]

Generate database check constraints.

Supports check constraints of type "length", "range" and "values". The constraints have to be given in the following form:

{
    "col_1": {"length": {"min": 0, "max": 10}},
    "col_2": {"range": {"min": 0, "max": 10}},
    "col_3": {"values": ["val_1", "val_2"]},
}

Parameters:: constraints – Dictionary of constraints to generate.
Returns:: A tuple of CheckConstraint instances.

kadi.lib.db.get_class_by_tablename(tablename)[source]

Get the model class mapped to a certain database table name.

Parameters:: tablename – Name of the table.
Returns:: The class reference or None if the table does not exist.

kadi.lib.db.is_column(model, attr)[source]

Check if a model’s attribute is a regular column.

Parameters:

model – The model that contains the column.
attr – Name of the column attribute.

Returns:

True if the attribute is a column, False otherwise.

kadi.lib.db.get_column_type(model, attr)[source]

Get the type of a column.

Parameters:

model – The model that contains the column.
attr – Name of the column attribute.

Returns:

The type of the column or None if the attribute is not a regular column.

kadi.lib.db.is_relationship(model, attr)[source]

Check if a model’s attribute is a relationship.

Parameters:

model – The model that contains the column.
attr – Name of the relationship attribute.

Returns:

True if the attribute is a relationship, False otherwise.

kadi.lib.db.is_many_relationship(model, attr)[source]

Check if a model’s attribute is a many-relationship.

Parameters:

model – The model that contains the column.
attr – Name of the relationship attribute.

Returns:

True if the attribute is a many-relationship, False otherwise.

kadi.lib.db.get_class_of_relationship(model, attr)[source]

Get the class of a relationship.

Parameters:

model – The model that contains the relationship.
attr – Name of the relationship attribute.

Returns:

The class reference or None if the attribute is not a relationship.

kadi.lib.db.escape_like(value, escape='\\')[source]

Escape a string for use in LIKE queries.

Will escape "%", "_" and the escape character specified by escape.

Parameters:

value – The string to escape.
escape – (optional) The escape character to use.

Returns:

The escaped string.

kadi.lib.db.acquire_lock(obj)[source]

Acquire a database lock on a given object.

The database row corresponding to the given object will be locked using FOR UPDATE until the session is committed or rolled back. Once the lock is acquired, the given object is also refreshed. Should only be used if strictly necessary, e.g. to prevent certain race conditions.

Parameters:: obj – The object to lock.
Returns:: The locked and refreshed object.

kadi.lib.db.has_extension(extension_name)[source]

Check if a given database extension is installed.

Parameters:: extension_name – The name of the extension.
Returns:: True if the extension is installed, False otherwise.

kadi.lib.db.get_disk_space()[source]

Get the used disk space of the database.

Returns:: The disk space of the database in bytes.

kadi.lib.db.has_pending_revisions()[source]

Check if the database has pending revisions.

Returns:: True if there are pending revisions, False otherwise.

Exceptions

exception kadi.lib.exceptions.KadiException[source]

Bases: Exception

Base exception class.

exception kadi.lib.exceptions.KadiConfigurationError[source]

Bases: KadiException

For errors related to invalid configuration.

exception kadi.lib.exceptions.KadiStorageError[source]

Bases: KadiException

Base file storage error class.

exception kadi.lib.exceptions.KadiFilesizeExceededError[source]

Bases: KadiStorageError

For errors related to exceeded file size.

exception kadi.lib.exceptions.KadiFilesizeMismatchError[source]

Bases: KadiStorageError

For errors related to file size validation.

exception kadi.lib.exceptions.KadiChecksumMismatchError[source]

Bases: KadiStorageError

For errors related to file checksum validation.

exception kadi.lib.exceptions.KadiValidationError[source]

Bases: KadiException

For errors related to value format validation.

exception kadi.lib.exceptions.KadiPermissionError[source]

Bases: KadiException

For errors related to permissions.

exception kadi.lib.exceptions.KadiDatabaseError[source]

Bases: KadiException

Base database error class.

exception kadi.lib.exceptions.KadiDecryptionKeyError[source]

Bases: KadiDatabaseError

For errors related to an invalid database value decryption key.

Export

class kadi.lib.export.PDF(title='')[source]

Bases: FPDF

Base PDF export class using FPDF.

Parameters:: title – (optional) The title of the PDF, which will appear in the header on each page and in the metadata of the PDF itself.

static format_date(date_time)[source]

Format a datetime object in a user-readable manner.

Parameters:: date_time – The datetime object to format as specified in Python’s datetime module.
Returns:: The formatted datetime string.

header()[source]: Automatically prints a header on each page of the generated PDF.

footer()[source]: Automatically prints a footer on each page of the generated PDF.

truncated_cell(w, text='', **kwargs)[source]

Print a cell with potentially truncated text based on the cell’s width.

Parameters:

w – The width of the cell.
text – (optional) The text content of the cell.
**kwargs – Additional keyword arguments to pass to fpdf2’s cell function.

calculate_max_height(contents)[source]

Calculate the maximum height that will be required by multiple multi-cells.

Note that this method always uses the current font family and size for its calculations.

Parameters:: contents – A list of tuples containing the width, the text content and the font style of each cell.
Returns:: The maximum height the cells will require.

class kadi.lib.export.ROCrate(*args, version=None, sized=True, **kwargs)[source]

Bases: ZipStream

Base RO-Crate export class.

This class behaves like a ZipStream, which can be used to attach file paths and streams. Note that the files and the content of the metadata are currently not synchronized automatically.

Parameters:

*args – Additional arguments to pass to the ZipStream.
version – (optional) A version string that represents the current specification that is used for the RO-Crate, which will be saved as “version” in the metadata describing the main metadata file.
sized – (optional) Whether the crate should keep track of its size.
**kwargs – Additional keyword arguments to pass to the ZipStream.

property root_graph: Get the root graph of the RO-Crate metadata.

property root_dataset: Get the root dataset of the RO-Crate metadata.

get_entity(entity_id)[source]

Get an entity of the root graph by its ID.

Parameters:: entity_id – The ID of the entity to retrieve.
Returns:: The entity or None if no suitable entity could be found.

dump_metadata()[source]

Dump the RO-Crate metadata as formatted JSON string.

Returns:: The JSON formatted string.

class kadi.lib.export.RDFGraph(*args, **kwargs)[source]

Bases: Graph

Base RDF graph export class.

author_ref(user_data)[source]

Create a URI reference of an author.

Parameters:: user_data – The serialized data of the user, via UserSchema, to use as an author.
Returns:: The created URI reference.

Favorites

kadi.lib.favorites.core.is_favorite(object_name, object_id, user=None)[source]

Check if the given object is favorited by the given user.

Parameters:

object_name – The type of object.
object_id – The ID of the object.
user – (optional) The user the favorite belongs to. Defaults to the current user.

Returns:

True if the object is favorited, False otherwise.

kadi.lib.favorites.core.toggle_favorite(object_name, object_id, user=None)[source]

Toggle the favorite state of the given object for the given user.

If a favorite already exists for the given object and user, it will be deleted from the database, otherwise it will be created.

Parameters:

object_name – The type of object.
object_id – The ID of the object.
user – (optional) The user the favorite belongs to. Defaults to the current user.

kadi.lib.favorites.core.delete_favorites(object_name, object_id)[source]

Delete all favorites of the given object.

Parameters:

object_name – The type of object.
object_id – The ID of the object.

class kadi.lib.favorites.models.FavoriteMixin[source]

Bases: object

Mixin for SQLALchemy models to check whether an object is favorited.

is_favorite(user=None)[source]

Check if the current object is favorited by the given user.

Wraps is_favorite() with the type and ID of the current object.

Parameters:: user – (optional) The user the favorite belongs to. Defaults to the current user.
Returns:: See is_favorite().

class kadi.lib.favorites.models.Favorite(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model representing favorited objects.

Each favorite is associated with a user, a specific type of object and an ID referring to a specific object instance.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'object', 'object_id']: See SimpleReprMixin.

id: The ID of the favorite, auto incremented.

user_id: The ID of the User the favorite belongs to.

object

The type of object the favorite refers to.

Currently always refers to a specific model via its table name.

object_id: The ID of the object the favorite refers to.

created_at: The date and time the favorite was created at.

classmethod create(*, user, object, object_id)[source]

Create a new favorite and add it to the database session.

Parameters:

user – The user the favorite belongs to.
object – The type of object the favorite refers to.
object_id – The ID of the object.

Returns:

The new Favorite object.

Federation

kadi.lib.federation.get_federated_instances(include_credentials=False, user=None)[source]

Get a list of federated Kadi instances.

Makes use of the FEDERATED_INSTANCES specified in the application’s configuration.

Parameters:

include_credentials – (optional) Whether to include the client ID ("client_id") and secret ("client_secret") in the returned instances.
user – (optional) The user who should be checked for whether they are connected with the OAuth2 provider of each corresponding instance, in which case the "is_connected" key will be included in the returned instances.

Returns:

A list of instance dictionaries in the following form:

[
    {
        "name": "example",
        "title": "Kadi4Mat Example",
        "url": "https://kadi4mat.example.edu",
        "client_id": "<client_id>",
        "client_secret": "<client_secret>",
        "is_connected": True,
    },
]

kadi.lib.federation.get_federated_instance(name, include_credentials=False, user=None)[source]

Get a specific federated Kadi instance.

Parameters:

name – The unique name of the instance.
include_credentials – (optional) See get_federated_instances().
user – (optional) See get_federated_instances().

Returns:

The instance in a format as described in get_federated_instances() or None if no instance with the given name could be found.

kadi.lib.federation.federated_request(name, endpoint, params=None, user=None)[source]

Perform a HTTP GET request in a federated Kadi instance.

Parameters:

name – The unique name of the instance.
endpoint – The endpoint to request as path.
params – (optional) A dictionary of additional query parameters to include in the request.
user – (optional) The user who is performing the request. Defaults to the current user.

Returns:

A JSON response depending on the success of the operation.

Format

kadi.lib.format.duration(seconds)[source]

Create a human-readable, translated duration string from an amount of seconds.

Note that locale-aware translations are only supported when having an active request context.

Parameters:: seconds – The amount of seconds.
Returns:: The formatted duration string.

kadi.lib.format.filesize(num_bytes)[source]

Create a human-readable, localized file size from a given amount of bytes.

Based on Jinja’s filesizeformat filter. Note that locale-aware localization is only supported when having an active request context.

Parameters:: num_bytes – The amount of bytes as a string or number.
Returns:: The formatted file size string.

kadi.lib.format.timestamp(date_time=None, include_micro=False)[source]

Build a UTC timestamp from a specific date and time.

The timestamp will be in the form of "YYYYMMDDHHmmss".

Parameters:

date_time – (optional) A datetime object as specified in Python’s datetime module. Defaults to the current time.
include_micro – (optional) Flag indicating whether to include microseconds in the timestamp as well or not.

Returns:

The formatted timestamp string.

kadi.lib.format.pretty_type_name(cls_or_string)[source]

Return a pretty type name based on a class or a string.

Parameters:: cls_or_string – A class reference (e.g. str) or a corresponding string (e.g. "str").
Returns:: The pretty type name string.

Forms

class kadi.lib.forms.CustomFieldMixin[source]

Bases: object

Mixin class for all custom fields and for wrapping existing fields.

Adds a common dictionary conversion to all inheriting fields and also handles some common corner cases.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.BooleanField(*args, **kwargs)[source]

Bases: CustomFieldMixin, BooleanField

Regular boolean field inheriting from CustomFieldMixin.

class kadi.lib.forms.IntegerField(*args, **kwargs)[source]

Bases: CustomFieldMixin, IntegerField

Regular integer field inheriting from CustomFieldMixin.

class kadi.lib.forms.StringField(*args, **kwargs)[source]

Bases: CustomFieldMixin, StringField

Regular string field inheriting from CustomFieldMixin.

class kadi.lib.forms.SubmitField(*args, **kwargs)[source]

Bases: CustomFieldMixin, SubmitField

Regular submit field inheriting from CustomFieldMixin.

class kadi.lib.forms.PasswordField(*args, **kwargs)[source]

Bases: CustomFieldMixin, PasswordField

Regular password field inheriting from CustomFieldMixin.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.FileField(*args, **kwargs)[source]

Bases: CustomFieldMixin, FileField

Custom file field.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.LFTextAreaField(*args, **kwargs)[source]

Bases: CustomFieldMixin, TextAreaField

Custom text area field that converts CRLF to LF.

process_formdata(valuelist)[source]

Process data received over the wire from a form.

This will be called during form construction with data supplied through the formdata argument.

Parameters:: valuelist – A list of strings to process.

class kadi.lib.forms.UTCDateTimeField(*args, **kwargs)[source]

Bases: CustomFieldMixin, DateTimeField

Custom timezone aware DateTimeField using UTC.

Parameters:: date_format – (optional) The date format to use for parsing and serializing.

process_formdata(valuelist)[source]

Process data received over the wire from a form.

This will be called during form construction with data supplied through the formdata argument.

Parameters:: valuelist – A list of strings to process.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.SelectField(*args, **kwargs)[source]

Bases: CustomFieldMixin, SelectField

Custom select field.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.DynamicSelectField(*args, **kwargs)[source]

Bases: CustomFieldMixin, SelectField

Custom select field for dynamically generated selections.

Note that this field automatically replaces empty strings in the form data with its default value (None) instead of trying to coerce them.

In addition, the instance variable initial can be used to specify an initial value to prefill the selection with by setting it to a tuple containing the actual value and a corresponding text to be displayed within the selection.

process_formdata(valuelist)[source]

Process data received over the wire from a form.

This will be called during form construction with data supplied through the formdata argument.

Parameters:: valuelist – A list of strings to process.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.DynamicMultiSelectField(*args, **kwargs)[source]

Bases: CustomFieldMixin, SelectMultipleField

Custom multi select field for dynamically generated selections.

The instance variable initial can be used to set a list of initial values to prefill the selection with. See also DynamicSelectField.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.JSONField(*args, **kwargs)[source]

Bases: CustomFieldMixin, Field

Custom field that processes its data as JSON.

process_formdata(valuelist)[source]

Process data received over the wire from a form.

This will be called during form construction with data supplied through the formdata argument.

Parameters:: valuelist – A list of strings to process.

to_dict()[source]: Convert this field into a dictionary representation.

class kadi.lib.forms.BaseForm(*args, **kwargs)[source]

Bases: FlaskForm

Base class for all forms.

Parameters:: suffix – (optional) A suffix that will be appended to all field IDs in the form of "<id>_<suffix>". This is especially useful when dealing with multiple forms on the same page. Note that this differs in behavior from the prefix parameter of WTForms, since only the IDs of fields are effected and not their names.

class kadi.lib.forms.BaseConfigForm(*args, **kwargs)[source]

Bases: BaseForm

Base form class for use in setting config items.

All fields in an inheriting form will be populated automatically from suitable config items stored in the database, if applicable. As keys for the config items, the field names are taken in uppercase.

Parameters:

user – (optional) A user indicating whether global or user-specific config items are to be used for prepopulating the form and when setting the values of config items in the database via set_config_values().
key_prefix – (optional) A string value to use as a prefix for all config items retrieved from and saved in the database. The prefix is used in uppercase and combined with each uppercase field name in the form of "<key_prefix>_<field_name>".
ignored_fields – (optional) A set of field names, as specified in the class attributes, to ignore when prepopulating the form and when setting the values of config items in the database via set_config_values(). Note that the "submit" field is always ignored.
encrypted_fields – (optional) A set of field names, as specified in the class attributes, to use encryption/decryption for when prepopulating the form and when setting the values of config items in the database via set_config_values(). Note that this only works for user-specific config items.
**kwargs – Additional keyword arguments to pass to BaseForm.

set_config_values()[source]

Automatically set all config items based on the respective field data.

Useful to populate all relevant config items in the database after a form is submitted. Similar to prepopulating the form fields, the names of the fields are taken in uppercase as keys for each config item.

kadi.lib.forms.validate_identifier(form, field)[source]

Validate an identifier in a form field.

Uses kadi.lib.validation.validate_identifier().

Parameters:

form – The form object.
field – The field object.

kadi.lib.forms.validate_mimetype(form, field)[source]

Validate a MIME type in a form field.

Uses kadi.lib.validation.validate_mimetype().

Parameters:

form – The form object.
field – The field object.

kadi.lib.forms.validate_username(form, field)[source]

Validate a local username in a form field.

Uses kadi.lib.validation.validate_username().

Parameters:

form – The form object.
field – The field object.

kadi.lib.forms.validate_iri(form, field)[source]

Validate an IRI in a form field.

Uses kadi.lib.validation.validate_iri().

Parameters:

form – The form object.
field – The field object.

kadi.lib.forms.convert_schema_validation_msg(msg, **interpolations)[source]

Convert a schema validation message to a corresponding form field message.

This is mainly useful when using a schema for validation in a field in order to better match the usual form validation messages. Note that, if possible, the messages will be translated using the translations provided by WTForms.

Parameters:

msg – The validation message to convert.
**interpolations – Additional keyword arguments to provide interpolation values for the converted validation message.

Returns:

The converted validation message or the original message if it could not be converted.

Jinja

class kadi.lib.jinja.SnippetExtension(environment: Environment)[source]

Bases: Extension

Jinja extension to easily pass variables to HTML snippets.

Example:

{% snippet "my_snippet.html", foo=1, bar=2 %}

tags: t.Set[str] = {'snippet'}: if this extension parses this is the list of tags it’s listening to.

parse(parser)[source]: Parse the snippet tag and arguments.

Licenses

kadi.lib.licenses.core.get_licenses(filter_term='')[source]

Get all licenses stored in the database.

Parameters:: filter_term – (optional) A (case insensitive) term to filter the licenses by their title or name.
Returns:: The licenses as query, ordered by their title in ascending order.

kadi.lib.licenses.core.get_builtin_licenses()[source]

Get all built-in licenses from their resource file.

Returns:: The licenses as dictionary, mapping the unique name of each license to another dictionary, containing the title ("title") and url ("url") of each license.

kadi.lib.licenses.core.get_plugin_licenses()[source]

Get all licenses added via plugins.

Uses the kadi.plugins.spec.kadi_get_licenses() plugin hook to collect the licenses. Invalid licenses will be ignored.

Returns:: The licenses as dictionary, in the same format as returned via get_builtin_licenses().

class kadi.lib.licenses.models.License(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model to represent licenses.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'name']: See SimpleReprMixin.

id: The ID of the license, auto incremented.

name: The unique name of the license.

title: The title of the license.

url: The optional URL of the license.

classmethod create(*, name, title, url=None)[source]

Create a new license and add it to the database session.

Parameters:

name – The name of the license.
title – The title of the license.
url – (optional) The URL of the license.

Returns:

The new License object.

class kadi.lib.licenses.schemas.LicenseSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent licenses.

See License.

kadi.lib.licenses.utils.initialize_builtin_licenses()[source]

Initialize all built-in licenses in the database.

Will create database objects of the licenses returned by get_builtin_licenses().

Returns:: True if at least one license was created, False otherwise.

LDAP

kadi.lib.ldap.make_upn(username, dn)[source]

Create a user principal name (UPN) for use in an Active Directory.

Parameters:

username – The name of a user to use as a prefix for the UPN.
dn – A DN to parse in order to retrieve the suffix of the UPN.

Returns:

The UPN in the form of "<username>@<domain>" or None if the given DN could not be parsed or does not contain any domain components.

kadi.lib.ldap.bind(connection)[source]

Try to authenticate with a server given an LDAP connection.

By default, LDAP connections are anonymous. The BIND operation establishes an authentication state between a client and a server.

Parameters:: connection – The connection object, see also make_connection().
Returns:: True if the BIND operation was successful, False otherwise.

kadi.lib.ldap.unbind(connection)[source]

Disconnect a given LDAP connection.

Parameters:: connection – The connection object, see also make_connection().

kadi.lib.ldap.make_server(host, port=389, use_ssl=False, validate_cert='REQUIRED', ciphers=None)[source]

Create a new LDAP Server object.

Parameters:

host – The host name or IP address of the LDAP server.
port – (optional) The port the LDAP server is listening on.
use_ssl – (optional) Flag indicating whether the entire connection should be secured via SSL/TLS.
validate_cert – (optional) Whether the certificate of the server should be validated. One of "NONE", "OPTIONAL" or "REQUIRED".
ciphers – (optional) One or more SSL/TLS ciphers to use as a single string according to the OpenSSL cipher list format. May also be set to "DEFAULT", in which case the default ciphers of the installed OpenSSL version are used.

Returns:

The new Server object or None if it could not be created.

kadi.lib.ldap.make_connection(server, user=None, password=None, use_starttls=False)[source]

Create a new LDAP Connection object.

Parameters:

server – The server object to use for the connection. See make_server().
user – (optional) The user for simple BIND.
password – (optional) The password of the user for simple BIND.
use_starttls – (optional) Flag indicating whether the connection should be secured via STARTTLS.

Returns:

The new Connection object or None if it could not be created.

kadi.lib.ldap.search(connection, search_base, search_filter, attribute_map, keep_list_attrs=False)[source]

Perform a search in an LDAP database given a connection.

Parameters:

connection – The LDAP connection to use. See make_connection().
search_base – The base of the search request.
search_filter – The filter of the search request.
attribute_map – A dictionary mapping arbitrary names to LDAP attribute names. The former names specify the keys to use for each search result (e.g. "firstname"), while the latter names specify the name of the attribute that should be extracted from the resulting LDAP entry (e.g. "givenName").
keep_list_attrs – (optional) Flag to indicate if results that have multiple values should be returned as lists or not. If not, only the first value of a result will be returned.

Returns:

A dictionary similar to the given attribute_map or None if no results could be retrieved. The LDAP attribute names will be replaced by the respective result value(s) in the result or with None if the attribute could not be found.

kadi.lib.ldap.modify_password(connection, user, new_password, old_password=None, active_directory=False)[source]

Modify a user’s LDAP password using an extended password modification operation.

Parameters:

connection – The LDAP connection to use. See make_connection().
user – The user whose password should be changed.
new_password – The new password of the user.
old_password – (optional) The old password of the user, if the LDAP server requires it.
active_directory – (optional) Flag indicating whether the LDAP server is an Active Directory, which does not support the standard extended password modification operation.

Returns:

A boolean value indicating whether the change was successful.

Mails

kadi.lib.mails.core.send_mail(*, subject, message, to_addresses, from_address=None, cc=None, bcc=None, attachments=None, reply_to=None, html_message=None, headers=None)[source]

Send an email to one or multiple recipients.

Uses the configuration values SMTP_HOST, SMTP_PORT, SMTP_USERNAME, SMTP_PASSWORD, SMTP_TIMEOUT and SMTP_USE_TLS set in the application’s configuration for the connection.

Parameters:

subject – The subject of the email.
message – The plain text message of the email.
to_addresses – A list of recipient addresses.
from_address – (optional) The sender’s email address. Defaults to the address set in MAIL_NO_REPLY in the current application’s configuration.
cc – (optional) A list of recipient addresses used in the “CC” header when sending the email.
bcc – (optional) A list of recipient addresses used in the “BCC” header when sending the email.
attachments – (optional) A list of attachments to put on the message. The list has to consist of triples in the form of (filename, content, mimetype). The content can either be a string or bytes object, while the MIME type will be guessed based on the given filename if omitted (i.e. set to None).
reply_to – (optional) A list of recipient addresses used in the “Reply-To” header when sending the email.
html_message – (optional) An HTML message of the email as alternative to the plain text version.
headers – (optional) A dictionary of additional headers to put on the message, mapping header names to their respective values.

Returns:

The number of emails that were sent successfully.

Raises:

ConnectionRefusedError: If no connection with the SMTP server could be established.

kadi.lib.mails.tasks.start_send_mail_task(*, subject, message, to_addresses, from_address=None, cc=None, bcc=None, attachments=None, reply_to=None, html_message=None, headers=None)[source]

Send a mail in a background task.

See kadi.lib.mails.core.send_mail() for the possible parameters.

In case the connection to the mail server fails, the task will be retried every 60 seconds until a maximum defined in CELERY_ANNOTATIONS in the application’s configuration is reached. Other errors will cause the task to fail, however.

Returns:: True if the task was started successfully, False otherwise. Note that the task being started successfully does not necessarily mean that the email will be sent successfully as well.

kadi.lib.mails.utils.send_email_confirmation_mail(identity, email=None)[source]

Send an email confirmation mail in a background task.

Uses kadi.lib.mails.tasks.start_send_mail_task() to send the mail.

Parameters:

identity – The identity of the user whose email should be confirmed.
email – (optional) The email address to use as the recipient address and to include in the email confirmation token. Defaults to the email address of the given identity.

Returns:

See kadi.lib.mails.tasks.start_send_mail_task().

kadi.lib.mails.utils.send_password_reset_mail(identity)[source]

Send a password reset mail in a background task.

Uses kadi.lib.mails.tasks.start_send_mail_task() to send the mail.

Parameters:: identity – The local identity of the user whose password should be reset.
Returns:: See kadi.lib.mails.tasks.start_send_mail_task().

kadi.lib.mails.utils.send_test_mail(user)[source]

Send a test mail in a background task.

Uses kadi.lib.mails.tasks.start_send_mail_task() to send the mail.

Parameters:: user – The user to send the test email to.
Returns:: See kadi.lib.mails.tasks.start_send_mail_task().

Notifications

kadi.lib.notifications.core.create_notification_data(notification)[source]

Create notification data suitable for presenting it to a client.

Parameters:: notification – A Notification object to use for creating the notification data.
Returns:: A tuple containing the title and the HTML body of the notification.

kadi.lib.notifications.core.dismiss_notification(notification)[source]

Dismiss a notification.

If the notification is of type "task_status", the referenced task will be revoked as well.

Parameters:: notification – The Notification to dismiss.

class kadi.lib.notifications.models.NotificationNames[source]

Bases: StringEnum

String enum containing all possible name values for notifications.

class kadi.lib.notifications.models.Notification(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model to represent notifications.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'name']: See SimpleReprMixin.

id: The ID of the notification, auto incremented.

user_id: The ID of the User the notification belongs to.

name: The name of the notification.

data: The optional data of the notification, depending on its type.

created_at: The date and time the notification was created at.

classmethod create(*, user, name, data=None)[source]

Create a new notification and add it to the database session.

Parameters:

user – The user the notification belongs to.
name – The name of the notification.
data – (optional) The data of the notification.

Returns:

The new Notification object.

class kadi.lib.notifications.schemas.NotificationSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent notifications.

See Notification.

OAuth

class kadi.lib.oauth.core.AuthorizationCodeGrant(request: OAuth2Request, server)[source]

Bases: AuthorizationCodeGrant

OAuth2 authorization code grant.

TOKEN_ENDPOINT_AUTH_METHODS = ['client_secret_post']: Allowed authentication methods for the token endpoint.

save_authorization_code(code, request)[source]: Save an OAuth2 authorization code in the database.

query_authorization_code(code, client)[source]: Retrieve an existing OAuth2 authorization code.

delete_authorization_code(authorization_code)[source]: Delete an existing OAuth2 authorization code.

authenticate_user(authorization_code)[source]: Authenticate a user related to an OAuth2 authorization code.

class kadi.lib.oauth.core.RefreshTokenGrant(request: OAuth2Request, server)[source]

Bases: RefreshTokenGrant

OAuth2 refresh token grant.

TOKEN_ENDPOINT_AUTH_METHODS = ['client_secret_post']: Allowed authentication methods for the token endpoint.

INCLUDE_NEW_REFRESH_TOKEN = True: The authorization server MAY issue a new refresh token

authenticate_refresh_token(refresh_token)[source]: Retrieve an existing OAuth2 server token based on a refresh token.

revoke_old_credential(refresh_token)[source]: Revoke an old OAuth2 server token.

authenticate_user(refresh_token)[source]: Authenticate a user related to an OAuth2 server token.

class kadi.lib.oauth.core.RevocationEndpoint(server)[source]

Bases: RevocationEndpoint

OAuth2 token revocation endpoint.

CLIENT_AUTH_METHODS = ['client_secret_post']: Allowed authentication methods for the revocation endpoint.

query_token(token_string, token_type_hint)[source]: Retrieve an existing OAuth2 server token.

revoke_token(token, request)[source]: Revoke an existing OAuth2 server token.

kadi.lib.oauth.core.create_oauth2_client_token(*, name, access_token, refresh_token=None, user=None, expires_at=None, expires_in=None)[source]

Create a new OAuth2 client token.

Parameters:

name – See OAuth2ClientToken.name.
access_token – See OAuth2ClientToken.access_token.
refresh_token – (optional) See OAuth2ClientToken.refresh_token.
user – (optional) The user the client token should belong to. Defaults to the current user.
expires_at – (optional) The expiration date and time of the access token as a Unix timestamp. Will be prioritized if expires_in is also given.
expires_in – (optional) The lifetime of the access token in seconds.

Returns:

The created OAuth2 client token.

kadi.lib.oauth.core.update_oauth2_client_token(oauth2_client_token, expires_at=None, expires_in=None, **kwargs)[source]

Update an existing OAuth2 client token.

Parameters:

oauth2_client_token – The client token to update.
expires_at – (optional) See create_oauth2_client_token().
expires_in – (optional) See create_oauth2_client_token().
**kwargs – Keyword arguments that will be passed to kadi.lib.db.update_object(). See also create_oauth2_client_token().

class kadi.lib.oauth.models.OAuth2ClientToken(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model to represent OAuth2 client tokens.

Note that this model uses encrypted fields and can potentially raise a KadiDecryptionKeyError when a value cannot be decrypted.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'name']: See SimpleReprMixin.

id: The ID of the client token, auto incremented.

user_id: The ID of the User the client token belongs to.

name

The name of the client token.

Currently always refers to the name of a specific OAuth2 provider.

access_token: The actual access token value, stored encrypted.

refresh_token: The optional refresh token value, stored encrypted.

expires_at: The optional expiration date and time of the access token.

property is_expired: Check if the access token is expired.

classmethod create(*, user, name, access_token, refresh_token=None, expires_at=None)[source]

Create a new OAuth2 client token and add it to the database session.

Parameters:

user – The user the client token should belong to.
name – The name of the client token.
access_token – The actual access token value.
refresh_token – (optional) The refresh token value.
expires_at – (optional) The expiration date and time of the access token.

Returns:

The new OAuth2ClientToken object.

to_authlib_token()[source]

Convert the client token to a format usable by an Authlib client.

Returns:: A dictionary representation of the client token.

class kadi.lib.oauth.models.OAuth2ServerClient(**kwargs)[source]

Bases: SimpleReprMixin, SimpleTimestampMixin, OAuth2ClientMixin, Model

Model to represent registered OAuth2 clients/applications.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'client_id', 'client_name']: See SimpleReprMixin.

id: The ID of the client, auto incremented.

user_id: The ID of the User who created the client.

client_id: The OAuth2 client ID.

client_secret

The OAuth2 client secret.

Note that only a hash of the actual client secret value is stored.

property client_metadata: Get the additional metadata of this client.

property scope: Get the scope of this client.

static new_client_secret()[source]

Create a new random client secret.

Returns:: The generated client secret.

static hash_client_secret(client_secret)[source]

Create a secure hash of a client secret.

Parameters:: client_secret – The client secret to hash.
Returns:: The calculated hash as a hexadecimal value.

classmethod create(*, user, client_name, client_uri, redirect_uris, scope=None, client_secret=None)[source]

Create a new OAuth2 client and add it to the database session.

Parameters:

user – The user the client should belong to.
client_name – The name of the client. Will be stored as part of the client metadata.
client_uri – The website of the client. Will be stored as part of the client metadata.
redirect_uris – A list of allowed redirect URIs. Will be stored as part of the client metadata.
scope – (optional) The scope of the client as a single string defining a list of space-delimited scope values. Will be stored as part of the client metadata.
client_secret – (optional) The client secret, which will be hashed before persisting. Defaults to a client secret created by new_client_secret().

Returns:

The new OAuth2ServerClient object.

set_client_metadata(value)[source]

Set the additional metadata of this client.

Parameters:: value – The metadata as a JSON serializable dictionary.

update_client_metadata(**kwargs)[source]

Update the additional metadata of this client.

Parameters:: **kwargs – JSON serializable keyword arguments to update the metadata with.

check_client_secret(client_secret)[source]

Compare the client secret of this client with a given client secret.

Parameters:: client_secret – The client secret to compare with, which will be hashed before comparing.

client_id_issued_at

client_secret_expires_at

created_at

The date and time an object has been created at.

Always uses the current UTC time.

last_modified

The date and time an object was last modified.

Always uses the current UTC time as initial value. After calling register_timestamp_listener() for an inheriting mixin class, this timestamp can automatically get updated by implementing _before_flush_timestamp() as a class method.

class kadi.lib.oauth.models.OAuth2ServerToken(**kwargs)[source]

Bases: SimpleReprMixin, AccessTokenMixin, OAuth2TokenMixin, Model

Model to represent OAuth2 server tokens.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'client_id']: See SimpleReprMixin.

id: The ID of the server token, auto incremented.

client_id: The client ID of the OAuth2ServerClient the server token belongs to.

access_token

The actual access token value.

Note that only a hash of the actual access token value is stored.

refresh_token

The actual refresh token value.

Note that only a hash of the actual refresh token value is stored.

property is_expired: Check if the access token is expired.

access_token_revoked_at

expires_in

issued_at

static new_access_token(include_prefix=True)[source]

Create a new random access token value.

Parameters:: include_prefix – (optional) Whether to include a prefix before the actual access token value to distinguish it with other types of access tokens.
Returns:: The generated access token value.

refresh_token_revoked_at

scope

The optional scope of the access token.

Represented as a single string defining a list of space-delimited scope values.

token_type

user

The user relationship of the access token.

A corresponding relationship should also be defined in the user table.

user_id: The ID of the user the access token belongs to.

static new_refresh_token()[source]

Create a new random refresh token value.

Returns:: The generated refresh token value.

static hash_token(token)[source]

Create a secure hash of an access or refresh token value.

Parameters:: token – The token value to hash.
Returns:: The calculated hash as a hexadecimal value.

classmethod get_by_access_token(token)[source]

Get a server token using an access token value.

Parameters:: token – The access token value to search for.
Returns:: The server token or None.

classmethod get_by_refresh_token(token)[source]

Get a server token using a refresh token value.

Parameters:: token – The refresh token value to search for.
Returns:: The server token or None.

classmethod create(*, user, client, expires_in, access_token=None, refresh_token=None, scope=None)[source]

Create a new OAuth2 server token and add it to the database session.

Parameters:

user – The user the server token should belong to.
client – The client the server token should belong to.
expires_in – The expiration time of the access token in seconds.
access_token – (optional) The actual access token value, which will be hashed before persisting. Defaults to an access token value created by new_access_token().
refresh_token – (optional) The actual refresh token value, which will be hashed before persisting. Defaults to a refresh token value created by new_refresh_token().
scope – (optional) The scope of the server token.

Returns:

The new OAuth2ServerToken object.

class kadi.lib.oauth.models.OAuth2ServerAuthCode(**kwargs)[source]

Bases: SimpleReprMixin, OAuth2AuthorizationCodeMixin, Model

Model to represent OAuth2 authorization codes.

Required for the implementation of the OAuth2 authorization code grant.

auth_time

code

code_challenge

code_challenge_method

nonce

redirect_uri

response_type

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'client_id']: See SimpleReprMixin.

id: The ID of the authorization code, auto incremented.

user_id: The ID of the User the authorization code belongs to.

client_id: The client ID of the OAuth2ServerClient the auth. code belongs to.

scope: The optional scope of the authorization code.

classmethod create(*, user, client, code, redirect_uri, scope=None, code_challenge=None, code_challenge_method=None)[source]

Create a new OAuth2 authorization code and add it to the database session.

Parameters:

user – The user the authorization code should belong to.
client – The client the authorization code should belong to.
code – The actual authorization code value.
redirect_uri – The allowed redirect URI of the authorization code.
scope – (optional) The scope of the authorization code.
code_challenge – (optional) The code challenge of the authorization code used for PKCE.
code_challenge_method – (optional) The code challenge method of the authorization code used for PKCE.

Returns:

The new OAuth2ServerAuthCode object.

is_expired()[source]

Check if the authorization code is expired.

Returns:: True if the authorization code is expired, False otherwise.

class kadi.lib.oauth.schemas.OAuth2ServerClientSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent OAuth2 clients.

See OAuth2ServerClient.

class kadi.lib.oauth.schemas.OAuth2ServerTokenSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent OAuth2 server tokens.

Permissions

kadi.lib.permissions.core.has_permission(subject, action, object_name, object_id, check_groups=True, check_defaults=True)[source]

Check if a user or group has permission to perform a specific action.

Checks all permissions corresponding to the roles of the given subject.

Parameters:

subject – The User or Group.
action – The action to check for.
object_name – The type of object.
object_id – The ID of a specific object or None for a global permission.
check_groups – (optional) Flag indicating whether the groups of a user should be checked as well for their permissions.
check_defaults – (optional) Flag indicating whether the default permissions of any object should be checked as well.

Returns:

True if permission is granted, False otherwise or if the object instance to check does not exist.

kadi.lib.permissions.core.get_permitted_objects(subject, action, object_name, check_groups=True, check_defaults=True)[source]

Get all objects a user or group has a specific permission for.

Checks all permissions corresponding to the roles of the given subject.

Parameters:

subject – The User or Group.
action – The action to check for.
object_name – The type of object.
check_groups – (optional) Flag indicating whether the groups of a user should be checked as well for their permissions.
check_defaults – (optional) Flag indicating whether the default permissions of the objects should be checked as well.

Returns:

The permitted objects as query or None if the object type does not exist.

kadi.lib.permissions.core.add_role(subject, object_name, object_id, role_name, update_timestamp=True)[source]

Add an existing role to a user or group.

Parameters:

subject – The User or Group.
object_name – The type of object the role refers to.
object_id – The ID of the object.
role_name – The name of the role.
update_timestamp – (optional) Flag indicating whether the timestamp of the underlying object should be updated or not. The object needs to implement BaseTimestampMixin in that case.

Returns:

True if the role was added successfully, False if the subject already has a role related to the given object.

Raises:

ValueError – If no object or role with the given arguments exists or when trying to add a role to the object that is being referred to by that role.

kadi.lib.permissions.core.remove_role(subject, object_name, object_id, update_timestamp=True)[source]

Remove an existing role of a user or group.

Parameters:

subject – The User or Group.
object_name – The type of object the role refers to.
object_id – The ID of the object.
update_timestamp – (optional) Flag indicating whether the timestamp of the underlying object should be updated or not. The object needs to implement BaseTimestampMixin in that case.

Returns:

True if the role was removed successfully, False if there was no role to remove.

Raises:

ValueError – If no object with the given arguments exists.

kadi.lib.permissions.core.set_system_role(user, system_role)[source]

Set an existing system role for a given user.

Parameters:

user – The user to set the system role for.
system_role – The name of the system role to set as defined in kadi.lib.constants.SYSTEM_ROLES.

Returns:

True if the system role was set successfully, False otherwise or if the given system role does not exist.

kadi.lib.permissions.core.create_role_rule(object_name, object_id, role_name, rule_type, condition, update_timestamp=True)[source]

Create a new role rule.

Parameters:

object_name – The type of object the role refers to.
object_id – The ID of the object.
role_name – The name of the role.
rule_type – The type of the role rule.
condition – The condition of the role rule.
update_timestamp – (optional) Flag indicating whether the timestamp of the underlying object should be updated or not. The object needs to implement BaseTimestampMixin in that case.

Returns:

The created role rule or None if the role rule could not be created.

kadi.lib.permissions.core.remove_role_rule(role_rule, update_timestamp=True)[source]

Remove an existing role rule.

Parameters:

role_role – The role rule to remove.
update_timestamp – (optional) Flag indicating whether the timestamp of the underlying object should be updated or not. The object needs to implement BaseTimestampMixin in that case.

kadi.lib.permissions.core.apply_role_rule(role_rule, user=None)[source]

Apply a given role rule.

Parameters:

role_rule – The role rule to apply.
user – (optional) A specific user to apply the role rule to. If not given, all existing users are considered.

class kadi.lib.permissions.models.Role(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model representing roles.

A role can be used to determine permissions of a user or group. There are two kinds of roles specified by this model:

Roles belonging to a specific object instance, in which case the object and object_id are set.
System roles, possibly belonging to multiple object types and instances, in which case the object and object_id are not set.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'name', 'object', 'object_id']: See SimpleReprMixin.

id: The ID of the role, auto incremented.

name: The name of the role.

object

The type of object the role refers to.

If set, always refers to a specific model via its table name. If not set, the object_id has to be None as well.

object_id

The ID of an object the role refers to.

If not set, the object has to be None as well.

classmethod create(*, name, object=None, object_id=None)[source]

Create a new role and add it to the database session.

Parameters:

name – The name of the role.
object – (optional) The type of object the role refers to.
object_id – (optional) The ID of the object.

Returns:

The new Role object.

class kadi.lib.permissions.models.RoleRuleType[source]

Bases: StringEnum

String enum containing all possible type values for role rules.

class kadi.lib.permissions.models.RoleRule(**kwargs)[source]

Bases: SimpleReprMixin, SimpleTimestampMixin, Model

Model to represent role rules.

Role rules can be used to automate permission management by automatically granting users or groups roles for different resources based on different conditions.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'role_id', 'type', 'condition']: See SimpleReprMixin.

created_at

The date and time an object has been created at.

Always uses the current UTC time.

last_modified

The date and time an object was last modified.

Always uses the current UTC time as initial value. After calling register_timestamp_listener() for an inheriting mixin class, this timestamp can automatically get updated by implementing _before_flush_timestamp() as a class method.

id: The ID of the role rule, auto incremented.

role_id: The ID of the Role the role rule refers to.

type

The type of the role rule.

Currently only "username".

condition

The condition of the role rule, depending on its type.

For each of the role rule types, the data consists of:

"username": An object containing the type of a user’s identity ("identity_type") and a pattern ("pattern") to check the corresponding username with.

classmethod create(*, role, type, condition)[source]

Create a new role rule and add it to the database session.

Parameters:

role – The role the role rule refers to.
type – The type of the role rule.
condition – The condition of the role rule.

Returns:

The new RoleRule object.

class kadi.lib.permissions.schemas.RoleSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent roles.

See Role.

class kadi.lib.permissions.schemas.RoleRuleSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent role rules.

See RoleRule.

kadi.lib.permissions.tasks.start_apply_role_rules_task(role_rule=None, user=None)[source]

Apply a specific or all existing role rules in a background task.

Parameters:

role_rule – (optional) A specific role rule to apply.
user – (optional) A specific user to apply the role rule(s) to. If not given, all existing users are considered.

Returns:

True if the task was started successfully, False otherwise.

kadi.lib.permissions.utils.permission_required(action, object_name, object_id_identifier, status_code=403)[source]

Decorator to add access restrictions based on permissions to an endpoint.

If the current user is not authenticated, the decorator will behave the same as Flask-Login’s login_required decorator. If the object or object instance to check do not exist, the request will automatically get aborted with a 404 status code.

Uses kadi.lib.permissions.core.has_permission() to check for access permissions.

Parameters:

action – See kadi.lib.permissions.core.has_permission().
object_name – See kadi.lib.permissions.core.has_permission().
object_id_identifier – The name of the variable to use as object_id, which needs to be part of the keyword arguments of the decorated function. May also be None.
status_code – (optional) The status code to use if no permission was granted.

kadi.lib.permissions.utils.initialize_roles(object_name, object_id)[source]

Initialize all roles of a specific object.

Will create database objects of the roles specified in the Meta.permissions attribute in the model corresponding to the given object type.

Example:

class Foo:
    class Meta:
        permissions = {
            "roles": {
                "admin": ["read", "update"],
            },
        }

Parameters:

object_name – The type of object the roles refer to.
object_id – The ID of the object.

Raises:

ValueError – If no object with the given arguments exists.

kadi.lib.permissions.utils.delete_roles(object_name, object_id)[source]

Delete the roles of a specific object.

Parameters:

object_name – The type of object the roles refer to.
object_id – The ID of the object.

kadi.lib.permissions.utils.get_user_roles(object_name, object_id=None)[source]

Get all users and roles of a specific object or object type.

Note that inactive users will be filtered out.

Parameters:

object_name – The type of the object.
object_id – (optional) The ID of a specific object.

Returns:

The users and corresponding roles of the object(s) as query.

kadi.lib.permissions.utils.get_group_roles(object_name, object_id=None)[source]

Get all groups and roles of a specific object or object type.

Note that inactive groups will be filtered out.

Parameters:

object_name – The type of the object.
object_id – (optional) The ID of a specific object.

Returns:

The groups and corresponding roles of the object(s) as query.

kadi.lib.permissions.utils.get_object_roles(object_name)[source]

Get all available roles and corresponding permissions of an object type.

Parameters:

object_name – The type of the object.

Returns:

A list of dictionaries in the following form:

[
    {
        "name": "admin",
        "permissions": [
            {
                "action": "read,
                "description": "Read this resource.",
            }
        ]
    }
]

kadi.lib.permissions.utils.initialize_system_roles()[source]

Initialize all system roles.

Will create database objects of the system roles defined in kadi.lib.constants.SYSTEM_ROLES that do not exist yet.

Returns:: True if at least one system role was created, False otherwise.

kadi.lib.permissions.utils.create_username_role_rule(object_name, object_id, role_name, identity_type, pattern)[source]

Create a role rule with conditions to check the values of usernames.

The conditions of username rules consist of an identity type (identity_type) and a pattern (pattern). The former specifies the type of identities to check the usernames of, while the letter specifies the possible values of the usernames. The pattern may include one or more wildcards using "*", which match a sequence of zero or more characters.

Parameters:

object_name – See kadi.lib.permissions.core.create_role_rule().
object_id – See kadi.lib.permissions.core.create_role_rule().
role_name – See kadi.lib.permissions.core.create_role_rule().
identity_type – The identity type of the condition.
pattern – The pattern expression of the condition.

Returns:

See kadi.lib.permissions.core.create_role_rule().

kadi.lib.permissions.utils.get_role_rules(object_name, object_id, rule_type=None)[source]

Get all existing role rules corresponding to roles of a specific object.

Parameters:

object_name – The type of object the role rules refer to through their corresponding roles.
object_id – The ID of the object the role rules refer to through their corresponding roles.
rule_type – (optional) A type to limit the role rules with.

Returns:

The filtered role rules as query.

Plugins

class kadi.lib.plugins.core.PluginConfigForm(*args, **kwargs)[source]

Bases: BaseConfigForm

Form class for use in setting plugin preferences as config items.

Parameters:

plugin_name – The unique name of the plugin, which is also passed to BaseConfigForm as key_prefix and suffix.
user – (optional) The user to pass to BaseConfigForm. Defaults to the current user.

kadi.lib.plugins.core.run_hook(name, **kwargs)[source]

Run the plugin hook with the given name for all registered plugins.

Parameters:

name – The name of the hook.
**kwargs – Additional keyword arguments that will be passed to the hook.

Returns:

A single result, if firstresult is set to True in the hook spec, or a list of results.

Raises:

ValueError – If no valid hook with the given name was found.

kadi.lib.plugins.core.template_hook(name, **kwargs)[source]

Run the plugin hook with the given name inside a template.

Uses run_hook() and joins multiple results together as a string ready to be inserted into a template.

Parameters:

name – See run_hook().
**kwargs – See run_hook().

Returns:

The template string, which may be empty if the given hook was not found or raised an exception.

kadi.lib.plugins.core.get_plugin_config(name)[source]

Get the configuration of a plugin.

For each plugin, configuration can be specified by mapping the name of the plugin to the configuration that the plugin expects in the PLUGIN_CONFIG value as configured in the application’s configuration. Each configuration has to be specified as a dictionary.

Parameters:: name – The name of the plugin.
Returns:: The plugin configuration or an empty dictionary if no valid configuration could be found.

kadi.lib.plugins.utils.get_plugin_scripts()[source]

Convenience function to retrieve all script URLs provided by plugins.

Uses the kadi.plugins.spec.kadi_get_scripts() plugin hook to collect the script URLs.

Returns:: A flattened list of all script URLs or an empty list if something went wrong while collecting the scripts.

kadi.lib.plugins.utils.get_plugin_frontend_translations()[source]

Convenience function to collect all frontend translations provided by plugins.

Uses the kadi_get_translations_bundles() plugin hook to collect and merge the translation bundles.

Returns:: A dictionary mapping each possible locale of the application to the merged translation bundles.

kadi.lib.plugins.utils.get_plugin_terms(query, page, per_page)[source]

Convenience function to collect terms provided by a plugin.

Uses the kadi_get_terms() plugin hook to collect the terms.

Parameters:

query – The search query.
page – The current result page used for pagination.
per_page – The number of results per page used for pagination.

Returns:

A tuple containing the total number of terms and a list of the terms themselves.

Publication

kadi.lib.publication.get_publication_providers(resource, user=None)[source]

Get a list of registered publication providers.

Uses the kadi.plugins.spec.kadi_get_publication_providers() plugin hook to collect potential publication providers combined with the information from kadi.lib.oauth.utils.get_oauth2_providers().

Note that this function may issue one or more database commits.

Parameters:

resource – The resource to eventually publish, an instance of Record or Collection.
user – (optional) The user who should be checked for whether they are connected with the OAuth2 provider the publication provider uses, in which case "is_connected" will be set to True for the respective provider. Defaults to the current user.

Returns:

A list of provider dictionaries in the following form, sorted by name:

[
    {
        "name": "example",
        "description": "An example publication provider.",
        "title": "Example provider",
        "website": "https://example.com",
        "is_connected": True,
    },
]

kadi.lib.publication.get_publication_provider(provider, resource, user=None)[source]

Get a specific, registered publication provider.

Note that this function may issue one or more database commits.

Parameters:

provider – The unique name of the publication provider.
resource – The resource to eventually publish, an instance of Record or Collection.
user – (optional) See get_publication_providers().

Returns:

The publication provider in a format as described in get_publication_providers() or None if no provider with the given name could be found.

kadi.lib.publication.publish_resource(provider, resource, form_data=None, user=None, task=None)[source]

Publish a resource using a given provider.

Uses the kadi.plugins.spec.kadi_publish_resource() plugin hook.

Note that this function issues one or more database commits.

Parameters:

provider – The unique name of the publication provider.
resource – The resource to publish, an instance of Record or Collection.
form_data – (optional) Form data as dictionary to customize the publication process.
user – (optional) The user who started the publication process. Defaults to the current user.
task – (optional) A Task object that that may be provided if this function is executed in a background task.

Returns:

A tuple consisting of a flag indicating whether the operation succeeded and a (HTML) template further describing the result in a user-readable manner, depending on the provider.

Resources

kadi.lib.resources.api.add_link(relationship, resource, user=None)[source]

Convenience function to link two resources together.

For ease of use in API endpoints. Uses kadi.lib.resources.utils.add_link().

Note that this function may issue a database commit.

Parameters:

relationship – See kadi.lib.resources.utils.add_link().
resource – See kadi.lib.resources.utils.add_link().
user – (optional) See kadi.lib.resources.utils.add_link().

Returns:

A JSON response depending on the success of the operation.

kadi.lib.resources.api.remove_link(relationship, resource, user=None)[source]

Convenience function to remove the link between two resources.

For ease of use in API endpoints. Uses kadi.lib.resources.utils.remove_link().

Note that this function may issue a database commit.

Parameters:

relationship – See kadi.lib.resources.utils.remove_link().
resource – See kadi.lib.resources.utils.remove_link().
user – (optional) See kadi.lib.resources.utils.remove_link().

Returns:

A JSON response depending on the success of the operation.

kadi.lib.resources.api.add_role(subject, resource, role_name, user=None)[source]

Convenience function to add an existing role to a user or group.

For ease of use in API endpoints. Uses kadi.lib.permissions.core.add_role().

Note that this function may issue a database commit.

Parameters:

subject – See kadi.lib.permissions.core.add_role().
resource – The resource the role refers to. An instance of Record, Collection, Template or Group.
role_name – See kadi.lib.permissions.core.add_role().
user – (optional) The user performing the operation. Defaults to the current user.

Returns:

A JSON response depending on the success of the operation.

kadi.lib.resources.api.remove_role(subject, resource)[source]

Convenience function to remove an existing role of a user or group.

For ease of use in API endpoints. Uses kadi.lib.permissions.core.remove_role().

Note that this function may issue a database commit.

Parameters:

subject – See kadi.lib.permissions.core.remove_role().
resource – The resource the role refers to. An instance of Record, Collection, Template or Group.

Returns:

A JSON response depending on the success of the operation.

kadi.lib.resources.api.change_role(subject, resource, role_name)[source]

Convenience function to change an existing role of a user or group.

For ease of use in API endpoints. If the given subject is the creator of the given resource, the role will not be changed. Uses kadi.lib.permissions.core.remove_role() and kadi.lib.permissions.core.add_role().

Note that this function may issue a database commit or rollback.

Parameters:

subject – The User or Group.
resource – The resource the role refers to. An instance of Record, Collection, Template or Group.
role_name – The name of the role to change.

Returns:

A JSON response depending on the success of the operation.

kadi.lib.resources.api.toggle_favorite_resource(resource, user=None)[source]

Toggle the favorite state of a resource.

Uses toggle_favorite() with the type and ID of the given resource.

Note that this function issues a database commit.

Parameters:

resource – The resource whose favorite state should be toggled. An instance of Record, Collection, Template or Group.
user – (optional) The user the favorite resource belongs to. Defaults to the current user.

Returns:

A JSON response indicating the success of the operation.

kadi.lib.resources.api.get_selected_resources(model, page=1, filter_term='', exclude=None, actions=None, filters=None, user=None)[source]

Convenience function to search resources for use in dynamic selections.

For ease of use in API endpoints. Used in conjunction with Select2 to search resources via dynamic select fields. Only the resources the given user has read permission for are returned.

Parameters:

model – The resource model to search, one of Record, Collection, Template or Group.
page – (optional) The current page.
filter_term – (optional) A (case insensitive) term to filter the resources by their title or identifier.
exclude – (optional) A list of resource IDs to exclude in the results. Defaults to an empty list.
actions – (optional) A list of further actions to check as part of the access permissions.
filters – (optional) A list of additional filter expressions to apply to the respective resource query.
user – (optional) The user performing the search. Defaults to the current user.

Returns:

A JSON response containing the results in the following form:

{
    "results": [
        {
            "id": 1,
            "text": "@identifier",
            "endpoint": "<resource API endpoint",
            "body": "<HTML body>"
        }
    ],
    "pagination": {
        "more": true
    }
}

kadi.lib.resources.api.get_resource_user_roles(resource, page=1, per_page=10, filter_term='', exclude=None)[source]

Get the paginated user roles of a resource.

Parameters:

resource – The resource to get the user roles of. An instance of Record, Collection, Template or Group.
page – (optional) The current page.
per_page – (optional) Items per page.
filter_term – (optional) A (case insensitive) term to filter the users by their username or display name.
exclude – (optional) A list of user IDs to exclude in the results.

Returns:

A tuple containing a list of deserialized user roles and the total amount of user roles.

kadi.lib.resources.api.get_resource_group_roles(resource, page=1, per_page=10, filter_term='', user=None)[source]

Get the paginated group roles of a resource.

This includes the special case of the given user not having read access to a group (any more) but wanting to update permissions of the group, if the permissions of the user allow them to. Since such groups should still be listed, they are included using a limited subset of group attributes.

Parameters:

resource – The resource to get the group roles of. An instance of Record, Collection or Template.
page – (optional) The current page.
per_page – (optional) Items per page.
filter_term – (optional) A query to filter the groups by their title or identifier.
user – (optional) The user to check for any permissions regarding the resulting groups. Defaults to the current user.

Returns:

A tuple containing a list of deserialized group roles and the total amount of user roles.

kadi.lib.resources.api.get_internal_resource_export(resource, export_type, export_func, export_filter=None, preview=False, download=False)[source]

Get the exported data of a resource for direct preview or download.

Only used internally, as the preview functionality of exported data is only useful in a browser context.

Parameters:

resource – The resource to export. An instance of Record or Collection.
export_type – A valid export type for the resource as defined in kadi.lib.constants.EXPORT_TYPES.
export_func – The export function corresponding to the resource to export.
export_filter – (optional) A dictionary specifying various export filters, which is passed to the given export function.
preview – (optional) Whether the exported data should be sent directly to the browser for preview instead of returning a URL. Only relevant for certain export types.
download – (optional) Whether the exported data should be downloaded as an attachment instead of just previewed.

Returns:

The exported data as a corresponding response object, depending on the given arguments.

kadi.lib.resources.api.filter_qparam(resource_type)[source]

Decorator to add a common filter query parameter to a resource API endpoint.

Parameters:: resource_type – The resource type as plural. Only used to generate a suitable parameter description.

kadi.lib.resources.api.sort_qparam(resource_type)[source]

Decorator to add a common sort query parameter to a resource API endpoint.

Parameters:: resource_type – The resource type as plural. Only used to generate a suitable parameter description.

kadi.lib.resources.core.signal_resource_change(revision)[source]

Convenience function to signal the creation or change of a resource.

Uses the kadi.plugins.spec.kadi_post_resource_change() plugin hook.

Note that this function may issue a database rollback.

Parameters:: revision – The revision that was created after a resource was created or changed. May also be None, in which case the plugin hook is skipped.

kadi.lib.resources.core.create_resource(model, tags=None, creator=None, **kwargs)[source]

Convenience function to create a new resource.

Note that this function issues a database commit or rollback.

Parameters:

model – The model that represents the type of the new resource. One of Record, Collection, Template or Group.
tags – (optional) A list of tags to tag the resource with if it inherits from TaggingMixin.
creator – (optional) The creator of the resource. Defaults to the current user.
**kwargs – Keyword arguments that will be used to intialize the data of the new resource.

Returns:

The created resource or None if the resource could not be created.

kadi.lib.resources.core.update_resource(resource, tags=None, user=None, **kwargs)[source]

Convenience function to update an existing resource.

Note that this function may issue a database commit or rollback.

Parameters:

resource – The resource to update. An instance of Record, Collection, Template or Group.
tags – (optional) A list of tags to tag the resource with if it inherits from TaggingMixin.
user – (optional) The user who triggered the update. Defaults to the current user.
**kwargs – Keyword arguments that will be passed to kadi.lib.db.update_object().

Returns:

True if the resource was updated successfully, False otherwise.

kadi.lib.resources.core.delete_resource(resource, user=None)[source]

Convenience function to delete an existing resource.

This will perform a soft deletion, i.e. only the resource’s state will be changed.

Note that this function issues a database commit.

Parameters:

resource – The resource to delete. An instance of Record, Collection, Template or Group.
user – (optional) The user who triggered the deletion. Defaults to the current user.

kadi.lib.resources.core.restore_resource(resource, user=None)[source]

Convenience function to restore a deleted resource.

Note that this function issues a database commit.

Parameters:

resource – The resource to restore. An instance of Record, Collection, Template or Group.
user – (optional) The user who triggered the restoration. Defaults to the current user.

kadi.lib.resources.core.purge_resource(resource)[source]

Convenience function to purge an existing resource.

This will completely delete the resource from the database.

Note that this function issues a database commit.

Parameters:: resource – The resource to purge. An instance of Record, Collection, Template or Group.

class kadi.lib.resources.forms.TagsField(*args, **kwargs)[source]

Bases: DynamicMultiSelectField

Custom dynamic multi select field for tagging resources.

Tags must not be empty, are automatically converted to lowercase and whitespaces are stripped and normalized. Additionally, the result will be sorted and duplicate tags will be filtered out.

Parameters:: max_len – (optional) The maximum length of each tag.

process_formdata(valuelist)[source]

Process data received over the wire from a form.

This will be called during form construction with data supplied through the formdata argument.

Parameters:: valuelist – A list of strings to process.

class kadi.lib.resources.forms.RolesField(*args, **kwargs)[source]

Bases: JSONField

Custom field to process and validate user and group roles of resources.

Uses ResourceRoleDataSchema for its validation.

Parameters:

roles – A list of roles, each item consisting of a tuple containing the actual role value and title to be displayed, similar to the choices in select fields.
allow_none – (optional) Whether to allow None as a valid role value.

process_formdata(valuelist)[source]

Process data received over the wire from a form.

This will be called during form construction with data supplied through the formdata argument.

Parameters:: valuelist – A list of strings to process.

to_dict()[source]: Convert this field into a dictionary representation.

set_initial_data(data=None, resource=None, user=None, keep_user_roles=False)[source]

Set the initial data of this field.

Parameters:

data – (optional) The form data to use for prefilling. Defaults to the submitted data of the current field instance.
resource – (optional) An existing resource, which can be used to set the initial data instead of the given form data. One of Record, Collection or Template.
user – (optional) A user that will be used for checking various access permissions when setting the data. Defaults to the current user.
keep_user_roles – (optional) Flag indicating whether to keep any roles of the given user.

class kadi.lib.resources.forms.BaseResourceForm(*args, **kwargs)[source]

Bases: BaseForm

Base form class for use in creating or updating different kinds of resources.

These resources may refer to instances of Record, Collection, Template or Group.

static identifier_field(**kwargs)[source]: Factory method to create a basic resource identifier field.

static visibility_field(**kwargs)[source]: Factory method to create a basic resource visibility field.

kadi.lib.resources.forms.check_duplicate_identifier(model, identifier, exclude=None)[source]

Check for a duplicate identifier in a form.

Parameters:

model – The model class to check the identifier of. One of Record, Collection, Template or Group.
identifier – The identifier to check.
exclude – (optional) An instance of the model that should be excluded in the check.

class kadi.lib.resources.mappings.ResourceMapping(meta: Dict[str, Any] | None = None, **kwargs: Any)[source]

Bases: BaseMapping

Base search mapping class to represent different kinds of resources.

These resources may refer to instances of Record, Collection, Template or Group.

class kadi.lib.resources.schemas.BaseResourceSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Base schema class to represent different kinds of resources.

These resources may refer to instances of Record, Collection, Template or Group.

class kadi.lib.resources.schemas.BaseResourceRoleSchema(obj=None, **kwargs)[source]

Bases: BaseSchema

Base schema class to represent different kinds of resource roles.

Parameters:: obj – (optional) An object that the current resource role refers to, which may be used when generating corresponding actions.

class kadi.lib.resources.schemas.UserResourceRoleSchema(obj=None, **kwargs)[source]

Bases: BaseResourceRoleSchema

Schema to represent user roles.

Parameters:: obj – (optional) An object that the current user role refers to. An instance of Record, Collection, Template or Group. See also BaseResourceRoleSchema.

dump_from_iterable(iterable)[source]

Serialize an iterable containing user roles.

Parameters:: iterable – An iterable yielding tuples each containing a user and a corresponding role object.
Returns:: The serialized output.

class kadi.lib.resources.schemas.GroupResourceRoleSchema(obj=None, **kwargs)[source]

Bases: BaseResourceRoleSchema

Schema to represent group roles.

Parameters:: obj – (optional) An object that the current group role refers to. An instance of Record, Collection or Template. See also BaseResourceRoleSchema.

dump_from_iterable(iterable)[source]

Serialize an iterable containing group roles.

Parameters:: iterable – An iterable yielding tuples each containing a group and a corresponding role object.
Returns:: The serialized output.

class kadi.lib.resources.schemas.ResourceRoleDataSchema(roles, allow_none=False, **kwargs)[source]

Bases: BaseSchema

Schema to represent the data of user or group resource roles.

Mainly useful in combination with kadi.lib.resources.views.update_roles() and within templates.

Parameters:

roles – A list of valid role values.
allow_none – (optional) Whether to allow None as a valid role value.

kadi.lib.resources.schemas.check_duplicate_identifier(model, identifier, exclude=None)[source]

Check for a duplicate identifier in a schema.

Parameters:

model – The model class to check the identifier of. One of Record, Collection, Template or Group.
identifier – The identifier to check.
exclude – (optional) An instance of the model that should be excluded in the check.

kadi.lib.resources.tasks.start_publish_resource_task(provider, resource, form_data=None, user=None, force_locale=True)[source]

Publish a resource using a given provider in a background task.

The created task will be kept in the database and the user who started the task will get notified about its current status as well.

Note that this function issues one or more database commits.

Parameters:

provider – The unique name of the publication provider.
resource – The resource to publish. An instance of Record or Collection.
form_data – (optional) Form data as dictionary to customize the publication process.
user – (optional) The user who started the task. Defaults to the current user.
force_locale – (optional) Flag indicating whether the current locale as returned by kadi.lib.web.get_locale() should be used inside the task. If False, the default locale will be used instead.

Returns:

A tuple containing a flag whether a task started by the given user is already pending/running, in which case no new task will be started, and either the new task object or None, depending on whether the task was started successfully.

kadi.lib.resources.tasks.start_purge_resources_task(user=None)[source]

Purge all deleted resources created by a given user in a background task.

Parameters:: user – (optional) The user who started the task and whose resources should be purged. Defaults to the current user.
Returns:: A tuple containing a flag whether a task started by the given user is already pending/running, in which case no new task will be started, and either the new task object or None, depending on whether the task was started successfully.

kadi.lib.resources.utils.get_filtered_resources(model, visibility=None, explicit_permissions=False, user_ids=None, user=None)[source]

Convenience function to get filtered resources of a specific model.

Parameters:

model – The model to filter. One of Record, Collection, Template or Group.
visibility – (optional) A visibility value to filter the resources with.
explicit_permissions – (optional) Flag indicating whether only resources with explicit access permissions for the given user should be included, independent of the given visibility value.
user_ids – (optional) A list of user IDs to filter the creators of the resources with.
user – (optional) The user to check for any permissions regarding the filtered resources. Defaults to the current user.

Returns:

The filtered resources as query.

kadi.lib.resources.utils.search_resources(model, search_query=None, page=1, per_page=10, sort='_score', filter_ids=None, extra_es_query=None)[source]

Convenience function to query the search index of a specific model.

Uses SearchableMixin.search() for the given model.

Parameters:

model – The model to search. One of Record, Collection, Template or Group.
search_query – (optional) The search query string.
page – (optional) The current page.
per_page – (optional) The amount of search results per page.
sort – (optional) The name of a field to sort on. One of "_score", "last_modified", "-last_modified", "created_at", "-created_at", "title", "-title", "identifier" or "-identifier". Falls back to "-last_modified" if no search query is given.
filter_ids – (optional) A list of resource IDs to restrict the search results to.
extra_es_query – (optional) An additional Elasticsearch DSL query object to combine with the given search query, if applicable.

Returns:

A tuple containing a list of the search results and the total amount of hits.

kadi.lib.resources.utils.get_order_column(model, sort)[source]

Convenience function to retrieve a column of a resource’s model for ordering.

Parameters:

model – The model to retrieve the column of. One of Record, Collection, Template or Group.
sort – A string representing the order column and direction. One of “last_modified”, “-last_modified”, “created_at”, “-created_at”, “title”, “-title”, “identifier” or “-identifier”.

Returns:

The order column.

kadi.lib.resources.utils.add_link(relationship, resource, user=None)[source]

Convenience function to link two resources together.

Note that only the link-permission of the given resource is checked.

Parameters:

relationship – The many-relationship to append the resource to.
resource – The resource to link. An instance of Record or Collection.
user – (optional) The user performing the link operation. Defaults to the current user.

Returns:

True if the link was established successfully, False if the link already exists.

Raises:

KadiPermissionError – If the user performing the operation does not have the necessary permissions.

kadi.lib.resources.utils.remove_link(relationship, resource, user=None)[source]

Convenience function to remove the link between two resources.

Note that only the link-permission of the given resource is checked.

Parameters:

relationship – The many-relationship to remove the resource from.
resource – The resource to remove. An instance of Record or Collection.
user – (optional) The user performing the link operation. Defaults to the current user.

Returns:

True if the link was removed successfully, False if the link does not exist.

Raises:

KadiPermissionError – If the user performing the operation does not have the necessary permissions.

kadi.lib.resources.utils.get_linked_resources(model, relationship, actions=None, user=None)[source]

Convenience function to get all linked resources that a user can access.

In this context having access to a resource means having read permission for that resource.

Parameters:

model – The model class corresopnding to the linked resources. One of Record or Collection.
relationship – The many-relationship that represents the linked resources to get.
actions – (optional) A list of further actions to check as part of the access permissions.
user – (optional) The user who will be checked for access permissions. Defaults to the current user.

Returns:

The resulting query of the linked resources.

kadi.lib.resources.utils.clean_resources(inside_task=False)[source]

Clean all expired, deleted resources.

Note that this function may issue one or more database commits.

Parameters:: inside_task – (optional) A flag indicating whether the function is executed in a task. In that case, additional information will be logged.

kadi.lib.resources.utils.purge_resources(user=None, timestamp=None, inside_task=False)[source]

Purge all deleted resources created by a given user.

Note that this function may issue one or more database commits.

Parameters:

user – (optional) The user the resources to purge were created by. Defaults to the current user.
timestamp – (optional) A timestamp as datetime object to limit the resources to be purged to the ones older than this timestamp regarding their last modification date.
inside_task – (optional) A flag indicating whether the function is executed in a task. In that case, additional information will be logged.

kadi.lib.resources.views.add_links(model, relationship, resource_ids, user=None)[source]

Convenience function to link multiple resources together.

For ease of use in view functions. Uses kadi.lib.resources.utils.add_link() but silently ignores any errors.

Parameters:

model – The model of which the resources to append are instances of. One of Record or Collection.
relationship – See kadi.lib.resources.utils.add_link().
resource_ids – A list of resource IDs that should be linked referring to instances of the given model.
user – (optional) See kadi.lib.resources.utils.add_link().

kadi.lib.resources.views.update_roles(resource, role_data, user=None)[source]

Convenience function to update roles of users and groups.

For ease of use in view functions. Uses kadi.lib.permissions.core.remove_role() and kadi.lib.permissions.core.add_role(), but silently ignores any errors.

Parameters:

resource – The resource the roles refer to, an instance of Record, Collection, Template or Group.
role_data – A list of dictionaries containing role data corresponding to the structure of ResourceRoleDataSchema.
user – (optional) The user performing the operation. Defaults to the current user.

Revisions

kadi.lib.revisions.core.create_revision(obj, user=None)[source]

Create a new revision of an object.

If none of the revisioned values changed, no new revision will be created. See also kadi.lib.revisions.core.setup_revisions().

Note that this function acquires a lock on the given object.

Parameters:

obj – The object to create a new revision for.
user – (optional) The user who triggered the revision.

Returns:

The created object revision or None if no new revision was created.

kadi.lib.revisions.core.delete_revisions(obj)[source]

Delete all revisions of an object.

Parameters:: obj – The object to delete the revisions of.

kadi.lib.revisions.core.setup_revisions()[source]

Setup revisioning for all models that support it.

The columns to store revisions of have to be specified in a Meta.revision attribute in each model. It should be a list of strings specifying the attribute names.

Example:

class Foo:
    class Meta:
        revision = ["bar", "baz[foo, bar]"]

The columns can either be regular columns, like the first value in the list, or relationships like the second value. For the latter, all regular columns of the relationship that should be included in the revision need to be specified in square brackets, separated by commas.

For each model, a new model class for the revisions will be created automatically, linked to the original model class and to Revision. The class of the revisioned model will also be stored on each new revision model as model_class. Additionally, the revision model class will be stored on the original model as revision_class as well as a convenience property to retrieve the revisions in descending order of their timestamp as ordered_revisions.

class kadi.lib.revisions.models.Revision(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model to represent general revision metadata.

The actual object revision models are created dynamically instead and linked to this model. See kadi.lib.revisions.core.setup_revisions().

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'timestamp']: See SimpleReprMixin.

id: The ID of the revision, auto incremented.

user_id

The optional ID of the User who triggered the revision.

May be omitted if there is no user that can clearly be associated with a certain revision, e.g. when automatically triggering a revision after deleting a resource that was referenced in the revisioned resource.

timestamp: The timestamp of the revision.

classmethod create(*, user=None)[source]

Create a new revision and add it to the database session.

Parameters:: user – (optional) The user who triggered the revision.
Returns:: The new Revision object.

class kadi.lib.revisions.schemas.ObjectRevisionSchema(schema, compared_revision=None, api_endpoint=None, view_endpoint=None, endpoint_args=None, view_object_url=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent object revisions.

Parameters:

schema – The schema to represent the object revisions with.
compared_revision – (optional) Another revision object to compare the object revisions with. By default, the comparison always uses the previous object revision, if applicable.
api_endpoint – (optional) An API endpoint to retrieve the current object revision.
view_endpoint – (optional) An endpoint to view the current object revision. Only relevant for internal use.
endpoint_args – (optional) Additional keyword arguments to append to the API and/or view endpoints when building the corresponding URL.
view_object_url – (optional) A URL to view the actual object the current revision refers to. Only relevant for internal use.

kadi.lib.revisions.utils.get_revision_columns(model)[source]

Parse the list of revisioned columns for a specific model.

Parameters:: model – The model to get the columns from.
Returns:: A tuple containing a list of the simple columns (i.e. no relationships) as string and a list of relationships. The second list itself is made up of tuples consisting of the name of the relationship and a list of colums of the relationship to revision as strings, similar to the first list.

Schemas

class kadi.lib.schemas.BaseSchema(*args, _internal=None, **kwargs)[source]

Bases: Schema

Base class for all schemas.

Note that all schemas that inherit from this class will silently exclude unknown fields by default.

Parameters:: _internal – (optional) Flag indicating whether additional data that’s only relevant for internal use should be included when serializing objects. If not set, the value returned by kadi.lib.api.utils.is_internal_api_request() will be taken instead. Note that this flag will generally not be passed automatically to any nested schemas.

class Meta[source]

Bases: object

Container to store meta class attributes.

load_or_400(data=None)[source]

Try to deserialize the given input.

Will try to deserialize/load the given input data using the schemas load method. If the validation fails or if the input is no valid JSON data in the first place, automatically abort the current request with status code 400 and a corresponding error response as JSON.

Parameters:: data – (optional) The input to deserialize. Defaults to the JSON body of the current request.
Returns:: The deserialized input.

class kadi.lib.schemas.CustomString(*args, allow_ws_only=False, filter=None, **kwargs)[source]

Bases: String

Custom string field that implements additional filtering and validation.

Parameters:

allow_ws_only – (optional) Flag indicating whether strings that are empty or only contain whitespace will be considered valid.
filter – (optional) A single or a list of filter and/or conversion functions that will be applied when deserializing the field data.

class kadi.lib.schemas.CustomPluck(*args, sort=True, flatten=True, unique=True, **kwargs)[source]

Bases: Pluck

Custom pluck field that implements additional (de)serialization options.

Parameters:

sort – (optional) Flag indicating whether serialized and deserialized output should be sorted if many is set.
flatten – (optional) Flag indicating whether deserialized output should be flattened, i.e. extracted from its outer dictionary. The same logic will also be applied to potential validation errors.
unique – (optional) Flag indicating whether deserialized items should be unique if many is set.

class kadi.lib.schemas.ValidateUUID(version=4)[source]

Bases: object

Validate a UUID of a specific version in a schema field.

Parameters:: version – (optional) The UUID version.

kadi.lib.schemas.validate_identifier(value)[source]

Validate an identifier in a schema field.

Uses kadi.lib.validation.validate_identifier().

Parameters:: value – The field value.

kadi.lib.schemas.validate_mimetype(value)[source]

Validate a MIME type in a schema field.

Uses kadi.lib.validation.validate_mimetype().

Parameters:: value – The field value.

kadi.lib.schemas.validate_iri(value)[source]

Validate an IRI in a schema field.

Uses kadi.lib.validation.validate_iri().

Parameters:: value – The field value.

Search

class kadi.lib.search.core.BaseMapping(meta: Dict[str, Any] | None = None, **kwargs: Any)[source]

Bases: Document

Base class for all search mappings.

classmethod get_attributes()[source]: Get a list of all attributes from a mapping class.

classmethod create_document(obj)[source]

Create a new document to be indexed in ElasticSearch

Parameters:: obj – The object to be indexed.
Returns:: The created document.

kadi.lib.search.core.create_index(model, force=False)[source]

Create a new search index if it does not exist yet.

The name of the index will be in the form of "<tablename>_<timestamp>", where <tablename> depends on the given model. An alias <tablename> pointing to the actual index will also be created automatically if it does not exist yet.

Parameters:

model – The model to create the index of. See also SearchableMixin.
force – (optional) Flag indicating whether a new index should be created even if one already exists. Note that if an alias already exists it will not be updated automatically to point to the new index.

Returns:

The name of the newly created or existing index or None if no new index could be created.

kadi.lib.search.core.add_to_index(obj, index=None)[source]

Add an object to its corresponding search index.

Parameters:

obj – The object to index. See also SearchableMixin.
index – (optional) The name of an index that should be used instead of the alias corresponding to the given object. See also create_index().

Returns:

True if the object was indexed successfully, False otherwise.

kadi.lib.search.core.remove_from_index(obj, index=None)[source]

Remove an object from its corresponding search index.

Parameters:

obj – The object to remove from the index. See also SearchableMixin.
index – (optional) The name of an index that should be used instead of the alias corresponding to the given object. See also create_index().

Returns:

True if the object was removed successfully, False otherwise.

kadi.lib.search.core.search_index(index, query=None, sort='_score', filter_ids=None, start=0, end=10)[source]

Query a specific search index.

Parameters:

index – The name of the search index.
query – (optional) The search query in form of an Elasticsearch DSL query object.
sort – (optional) The name of a field or a list of multiple fields to sort on.
filter_ids – (optional) A list of IDs to restrict the search results to.
start – (optional) The start index of the results to return.
end – (optional) The end index of the results to return.

Returns:

A response object as returned by the Elasticsearch DSL or None if the request returned an error. Note that only the metadata will be returned as part of the search results, without any field data.

Raises:

elasticsearch.exceptions.ConnectionError – If no connection could be established to Elasticsearch.

class kadi.lib.search.models.SearchableMixin[source]

Bases: object

Mixin for SQLALchemy models to add support for searching.

The columns to index have to be specified in a mapping class, which has to be configured with its fully qualified name using Meta.search_mapping.

Example:

class Foo:
    class Meta:
        search_mapping = "kadi.modules.record.mapping.RecordMapping"

After calling register_search_listeners(), the search index will automatically get updated whenever an object is created or deleted or if any of the indexed columns (or the state column, if present) are updated using add_to_index() and remove_from_index().

classmethod get_mapping_class()[source]: Convenience method to get the mapping class of a model.

classmethod search(query=None, sort='_score', filter_ids=None, start=0, end=10)[source]

Query the search index corresponding to this model.

Uses search_index(), but returns the actual results instead of the raw search response.

Parameters:

query – (optional) See search_index().
sort – (optional) See search_index().
filter_ids – (optional) See search_index().
start – (optional) See search_index().
end – (optional) See search_index().

Returns:

A tuple containing a list of the search results and the total amount of hits.

Raises:

elasticsearch.exceptions.ConnectionError – If no connection could be established to Elasticsearch.

classmethod register_search_listeners()[source]

Register listeners to automatically update the search index.

Uses SQLAlchemy’s before_flush, after_commit and after_rollback events and propagates to all inheriting models.

class kadi.lib.search.models.SavedSearch(**kwargs)[source]

Bases: SimpleReprMixin, SimpleTimestampMixin, Model

Model representing saved searches.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'name', 'object']: See SimpleReprMixin.

check_constraints = {'name': {'length': {'max': 150}}, 'query_string': {'length': {'max': 4096}}}: See kadi.lib.db.generate_check_constraints().

id: The ID of the saved search, auto incremented.

user_id: The ID of the User the saved search belongs to.

name

The name of the saved search.

Restricted to a maximum length of 150 characters.

created_at

The date and time an object has been created at.

Always uses the current UTC time.

last_modified

The date and time an object was last modified.

Always uses the current UTC time as initial value. After calling register_timestamp_listener() for an inheriting mixin class, this timestamp can automatically get updated by implementing _before_flush_timestamp() as a class method.

object

The type of object the saved search refers to.

Currently always refers to a specific searchable model via its table name.

query_string

The query string representing the saved search.

This simply corresponds to the raw URL query parameter string used when searching the corresponding object. May be stored with or without a leading question mark.

Restricted to a maximum length of 4096 characters.

property qparams

Get a dictionary representation of the query string of this saved search.

Corresponds to the results of Python’s urllib.parse.parse_qs.

classmethod create(*, user, name, object, query_string)[source]

Create a new saved search and add it to the database session.

Parameters:

user – The user the saved search belongs to.
name – The name of the saved search.
object – The object the saved search refers to.
query_string – The query string of the saved search.

Returns:

The new SavedSearch object.

class kadi.lib.search.schemas.SavedSearchSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent saved searches.

See SavedSearch.

Security

kadi.lib.security.random_alnum(num_chars=16)[source]

Generate a secure random alphanumeric string.

Parameters:: num_chars – (optional) The number of chars to generate.
Returns:: The generated string.

kadi.lib.security.random_bytes(num_bytes=24, hexadecimal=True)[source]

Generate a secure random byte sequence.

Parameters:

length – (optional) The number of bytes to generate.
hexadecimal – (optional) Whether to return the bytes in a hexadecimal string representation.

Returns:

The generated byte sequence or hexadecimal string.

kadi.lib.security.hash_value(value, alg='sha256', hexadecimal=True)[source]

Create a secure hash of a value.

Parameters:

value – The value to hash, either a bytes object or a string.
alg – (optional) The hash algorithm to use, according to the algorithms available in Python’s “hashlib” module.
hexadecimal – (optional) Whether to return the hash in a hexadecimal string representation.

Returns:

The calculated hash as a byte sequence or hexadecimal string or None if the given algorithm is not available.

kadi.lib.security.encode_jwt(payload, expires_in=None)[source]

Encode a given payload inside a JSON web token.

Parameters:

payload – The payload to encode as dictionary, which needs to be JSON serializable.
expires_in – (optional) The time in seconds that the token should expire in.

Returns:

The encoded token as string.

kadi.lib.security.decode_jwt(token)[source]

Decode a given JSON web token.

Parameters:: token – The token to decode as string.
Returns:: The decoded payload dictionary or None if the token is invalid or expired.

Storage

class kadi.lib.storage.core.BaseStorage(storage_type, storage_name=None)[source]

Bases: object

Base class for all storage providers.

Note that storage providers operate on file identifiers, which are used to uniquely identify files within the underlying storage system via corresponding, generated file paths. All storage methods may raise a ValueError if a given file identifier is not suitable for generating a file path that the corresponding storage can work with. A suitable example of an identifier is a randomly generated UUID or variations thereof.

Parameters:

storage_type – The unique type of the storage.
storage_name – (optional) A user-readable name of the storage. Defaults to the given storage type.

property storage_type: Get the type of this storage.

property storage_name: Get the name of this storage.

static validate_size(actual_size, expected_size)[source]

Validate the size of a file using a simple comparison.

Parameters:

actual_size – The actual size of the file.
expected_size – The expected size of the file.

Raises:

KadiFilesizeMismatchError – If the sizes don’t match.

static validate_checksum(actual_checksum, expected_checksum)[source]

Validate the checksum of a file using a simple comparison.

Parameters:

actual_checksum – The actual checksum of the file.
expected_checksum – The expected checksum of the file.

Raises:

KadiChecksumMismatchError – If the checksums don’t match.

exists(identifier)[source]

Check if a file exists.

Parameters:: identifier – The identifier of the file to check.
Returns:: True if the file exists, False otherwise.

get_size(identifier)[source]

Get the size of a file.

Parameters:: identifier – The identifier of the file.
Returns:: The size of the file in bytes.

get_mimetype(identifier)[source]

Get the MIME type of a file based on its content.

Parameters:: identifier – The identifier of the file.
Returns:: The MIME type of the file.

open(identifier, mode='rb', encoding=None)[source]

Open a file for reading or writing.

Note that the file object returned by this method must provide a file-like API with the usual IO methods such as read, write or seek.

Parameters:

identifier – The identifier of the file to open.
mode – (optional) The mode in which the file is opened.
encoding – (optional) The encoding to use when opening the file in text mode.

Returns:

The opened file object.

save(identifier, stream, max_size=None)[source]

Save the contents of a binary stream in a file.

Parameters:

identifier – The identifier of the file.
stream – The readable binary stream to save.
max_size – (optional) The maximum size that the storage should allow for the destination file.

Returns:

An MD5 hash of the stream contents.

Raises:

KadiFilesizeExceededError – If the given max_size was exceeded. Note that any contents written to the file until this point may stay intact.

move(src_identifier, dst_identifier)[source]

Move a file to another location.

Parameters:

src_identifier – The identifier of the source file.
dst_identifier – The identifier of the destination file.

delete(identifier)[source]

Delete a file if it exists.

Parameters:: identifier – The identifier of the file to delete.

merge(identifier, identifier_list)[source]

Create a single file from a list of files.

Parameters:

identifier – The identifier of the destination file.
identifier_list – A list of identifiers of files to merge in sequential order. Note that the underlying files will stay intact.

download(identifier, *, filename, mimetype, as_attachment=True)[source]

Send a file to a client.

Parameters:

identifier – The identifier of the file to send.
filename – The name of the file to send.
mimetype – The MIME type of the file to send.
as_attachment – (optional) Whether to send the file as an attachment. Note that setting this parameter to False may pose a security risk, depending on the file contents, client and context.

Returns:

The response object.

class kadi.lib.storage.core.NullStorage(storage_type)[source]

Bases: BaseStorage

Fallback storage provider used as placeholder for invalid storage types.

Raises a KadiConfigurationError for all storage provider methods.

kadi.lib.storage.core.get_storage_provider(storage_type, use_fallback=True)[source]

Get a configured storage provider for a given storage type.

Parameters:

storage_type – The storage type.
use_fallback – (optional) Whether to return a fallback storage provider if no suitable storage provider could be found.

Returns:

The storage provider or an instance of NullStorage if no suitable storage provider could be found and use_fallback is True, None otherwise.

class kadi.lib.storage.local.LocalStorage(root_directory, num_dirs=3, dir_len=2)[source]

Bases: BaseStorage

Storage provider that uses the local file system.

Parameters:

root_directory – The directory the storage provider operates in. Must be an absolute path.
num_dirs – (optional) Number of directories for local file paths generated by this storage provider. Must be a minimum of 0.
dir_len – (optional) Length of each directory for local file paths generated by this storage provider. Must be a minimum of 1.

Raises:

ValueError – If the given root directory is not suitable.

property root_directory: Get the root directory of this storage.

exists(identifier)[source]

Check if a file exists.

Parameters:: identifier – The identifier of the file to check.
Returns:: True if the file exists, False otherwise.

get_size(identifier)[source]

Get the size of a file.

Parameters:: identifier – The identifier of the file.
Returns:: The size of the file in bytes.

get_mimetype(identifier)[source]

Get the MIME type of a file based on its content.

Parameters:: identifier – The identifier of the file.
Returns:: The MIME type of the file.

open(identifier, mode='rb', encoding=None)[source]

Open a file for reading or writing.

Note that the file object returned by this method must provide a file-like API with the usual IO methods such as read, write or seek.

Parameters:

identifier – The identifier of the file to open.
mode – (optional) The mode in which the file is opened.
encoding – (optional) The encoding to use when opening the file in text mode.

Returns:

The opened file object.

save(identifier, stream, max_size=None)[source]

Save the contents of a binary stream in a file.

Parameters:

identifier – The identifier of the file.
stream – The readable binary stream to save.
max_size – (optional) The maximum size that the storage should allow for the destination file.

Returns:

An MD5 hash of the stream contents.

Raises:

KadiFilesizeExceededError – If the given max_size was exceeded. Note that any contents written to the file until this point may stay intact.

move(src_identifier, dst_identifier)[source]

Move a file to another location.

Parameters:

src_identifier – The identifier of the source file.
dst_identifier – The identifier of the destination file.

delete(identifier)[source]

Delete a file if it exists.

Parameters:: identifier – The identifier of the file to delete.

merge(identifier, identifier_list)[source]

Create a single file from a list of files.

Parameters:

identifier – The identifier of the destination file.
identifier_list – A list of identifiers of files to merge in sequential order. Note that the underlying files will stay intact.

download(identifier, *, filename, mimetype, as_attachment=True)[source]

Send a file to a client.

Parameters:

identifier – The identifier of the file to send.
filename – The name of the file to send.
mimetype – The MIME type of the file to send.
as_attachment – (optional) Whether to send the file as an attachment. Note that setting this parameter to False may pose a security risk, depending on the file contents, client and context.

Returns:

The response object.

class kadi.lib.storage.misc.MiscStorage[source]

Bases: LocalStorage

Storage provider used for miscellaneous uploads.

Uses LocalStorage with a fixed root directory as specified in MISC_UPLOADS_PATH in the application’s configuration and a number of only two directories for all generated file paths. Note that this provider is not used for general record file storage.

kadi.lib.storage.misc.save_as_thumbnail(identifier, stream, max_resolution=(512, 512))[source]

Save image data as a JPEG thumbnail.

Uses the MiscStorage to store the thumbnails.

Parameters:

identifier – A file identifier that will be passed to the storage in order to save the image data.
stream – The image data as a readable binary stream. The actual data must be of one of the image types defined in kadi.lib.constants.IMAGE_MIMETYPES and must have a maximum size as defined in kadi.lib.constants.IMAGE_MAX_SIZE.
max_resolution – (optional) The maximum resolution of the thumbnail in pixels.

Returns:

True if the thumbnail was saved successfully, False otherwise. Note that the original image file may be (partially) saved regardless of whether the thumbnail could be generated from it.

kadi.lib.storage.misc.delete_thumbnail(identifier)[source]

Delete a thumbnail.

This is the inverse operation of save_as_thumbnail().

Parameters:: identifier – See save_as_thumbnail().

kadi.lib.storage.misc.preview_thumbnail(identifier, filename)[source]

Send a thumbnail to a client for previewing.

Uses the MiscStorage to send the thumbnails.

Parameters:

identifier – A file identifier that will be passed to the storage in order to send the thumbnail.
filename – The name of the thumbnail to send.

Returns:

The response object.

Tags

kadi.lib.tags.core.get_tags(filter_term='', resource_type=None, user=None)[source]

Get all distinct tags of resources a user can access.

Parameters:

filter_term – (optional) A (case insensitive) term to filter the tags by their name.
resource_type – (optional) A resource type to limit the tags to. One of "record" or "collection".
user – (optional) The user who will be checked for access permissions. Defaults to the current user.

Returns:

The tags as query, ordered by their name in ascending order.

class kadi.lib.tags.models.TaggingMixin[source]

Bases: object

Mixin for SQLALchemy models to add support for tagging.

The model needs a many-to-many tags relationship connecting itself with the Tag model.

set_tags(names)[source]

Set one or more tags.

Will create a new tag object for each tag name that does not yet exist in the database and add it to the relationship. Existing tags from the relationship that are not in the given list are removed.

Parameters:: names – A list of tag names.
Returns:: True if the tags were set successfully, False otherwise.

class kadi.lib.tags.models.Tag(**kwargs)[source]

Bases: SimpleReprMixin, Model

Model to represent tags.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'name']: See SimpleReprMixin.

check_constraints = {'name': {'length': {'max': 50}}}: See kadi.lib.db.generate_check_constraints().

id: The ID of the tag, auto incremented.

name

The unique name of the tag.

Restricted to a maximum length of 50 characters.

classmethod create(*, name)[source]

Create a new tag and add it to the database session.

Parameters:: name – The name of the tag.
Returns:: The new Tag object.

classmethod get_or_create(name)[source]

Return an existing tag or create one if it does not exist yet.

See create() for an explanation of the parameters.

Returns:: The new or existing Tag object.

class kadi.lib.tags.schemas.TagSchema(*args, _internal=None, **kwargs)[source]

Bases: BaseSchema

Schema to represent tags.

See Tag.

Tasks

kadi.lib.tasks.core.launch_task(name, args=None, kwargs=None, user=None, keep=False, notify=False, notification_data=None)[source]

Launch a new Celery task by name.

Note that this function issues one or more database commits if keep is True.

Parameters:

name – The name of the task.
args – (optional) Positional arguments to pass to the task as list. All arguments need to be JSON serializable.
kwargs – (optional) Keyword arguments to pass to the task as dictionary. All arguments need to be JSON serializable.
user – (optional) The user who started the task.
keep – (optional) Flag indicating whether the task should be kept in the database until its expiration date (via the periodic cleanup task). See also Task.
notify – (optional) Flag indicating whether the user who started the task should be notified about its status, in which case a new instance of Notification of type "task_status" will be created for that user. The data of this notification consists of a dictionary, containing the ID of the task as "task_id". This requires that a valid user is passed and keep must be True as well.
notification_data – (optional) Additional, JSON serializable data which is passed to the data dictionary of the Notification as "task_meta". Requires notify to be True. See also kadi.lib.notifications.core.create_notification_data().

Returns:

A boolean indicating whether the task was launched successfully or not if keep is False. A new Task object or None if keep is True.

class kadi.lib.tasks.models.TaskState[source]

Bases: StringEnum

String enum containing all possible state values for tasks.

class kadi.lib.tasks.models.Task(**kwargs)[source]

Bases: SimpleReprMixin, SimpleTimestampMixin, Model

Model to represent tasks.

class Meta[source]

Bases: object

Container to store meta class attributes.

representation = ['id', 'user_id', 'name', 'state']: See SimpleReprMixin.

check_constraints = {'progress': {'range': {'max': 100, 'min': 0}}, 'state': {'values': ['pending', 'running', 'revoked', 'success', 'failure']}}: See kadi.lib.db.generate_check_constraints().

id: The ID of the task, auto incremented.

user_id: The optional ID of the User who started the task.

name: The name of the task.

arguments

The arguments of the task.

Stored in the following form as JSON:

{
    "args": ["foo"],
    "kwargs": {
        "bar": "baz"
    }
}

progress

The progress of the task.

Must be a value between 0 and 100.

created_at

The date and time an object has been created at.

Always uses the current UTC time.

last_modified

The date and time an object was last modified.

Always uses the current UTC time as initial value. After calling register_timestamp_listener() for an inheriting mixin class, this timestamp can automatically get updated by implementing _before_flush_timestamp() as a class method.

result: The optional result of the task, depending on the type of task.

state

The state of the task.

One of "pending", "running", "revoked", "success" or "failure".

property is_revoked

Check if a task is revoked.

Will always refresh the task object to get up to date values, as revoking usually happens outside the current database session context (e.g. in another process).

property pretty_state: Return the state of a task in a human-readable and translated format.

classmethod create(*, creator, name, args=None, kwargs=None, state='pending')[source]

Create a new task and add it to the database session.

Parameters:

creator – The user who is starting the task.
name – The name of the task.
args – (optional) The positional arguments of the task as list.
kwargs – (optional) The keyword arguments of the task as dictionary.
state – (optional) The state of the task.

Returns:

The new Task object.

revoke()[source]: Revoke a task if it is still pending or running.

update_progress(percent)[source]

Update a tasks progress.

Parameters:: percent – The progress in percent, which needs to be an integer or float value between 0 and 100.

kadi.lib.tasks.utils.clean_tasks(inside_task=False)[source]

Clean all expired, finished tasks.

Note that this function issues a database commit.

Parameters:: inside_task – (optional) A flag indicating whether the function is executed in a task. In that case, additional information will be logged.

Utils

class kadi.lib.utils.SimpleReprMixin[source]

Bases: object

Mixin to add a simple implementation of __repr__ to a class.

The provided implementation uses all instance or class attributes specified in the Meta.representation attribute of the inheriting class. It should be a list of strings specifying the attributes to use in the representation.

Example:

class Foo:
    class Meta:
        representation = ["bar", "baz"]

    bar = 1

    baz = 2

class kadi.lib.utils.StringEnum[source]

Bases: object

Custom enum-like class that uses regular strings as values.

An inheriting class needs to specify a __values__ attribute for all possible enum string values. Each value will be added as a class attribute using its respective value as key in uppercase.

Example:

class Foo:
    __values__ = ["bar"]

kadi.lib.utils.suppress_stderr()[source]: Context manager to suppress output written to stderr.

kadi.lib.utils.named_tuple(tuple_name, **kwargs)[source]

Convenience function to build a namedtuple from keyword arguments.

Parameters:

tuple_name – The name of the tuple.
**kwargs – The keys and values of the tuple.

Returns:

The namedtuple instance.

kadi.lib.utils.compare(left, op, right)[source]

Compare two values with a given operator.

Parameters:

left – The left value.
op – One of "==", "!=", ">", "<", ">=" or "<=".
right – The right value.

Returns:

The boolean result of the comparison.

kadi.lib.utils.rgetattr(obj, name, default=None)[source]

Get a nested attribute of an object.

Parameters:

obj – The object to get the attribute from.
name – The name of the attribute in the form of "foo.bar.baz".
default – (optional) The default value to return if the attribute could not be found.

Returns:

The attribute or the default value if it could not be found.

kadi.lib.utils.get_class_by_name(name)[source]

Get a class given its name.

Parameters:: name – The complete name of the class in the form of "foo.bar.Baz".
Returns:: The class or None if it could not be found.

kadi.lib.utils.utcnow()[source]

Create a timezone aware datetime object of the current time in UTC.

Returns:: A datetime object as specified in Python’s datetime module.

kadi.lib.utils.compact_json(data, ensure_ascii=False, sort_keys=True)[source]

Serialize data to a compact JSON formatted string.

Uses the JSON encoder provided by Flask, which can deal with some additional types.

Parameters:

data – The data to serialize.
ensure_ascii – (optional) Whether to escape non-ASCII characters.
sort_keys – (optional) Whether to sort the output of dictionaries by key.

Returns:

The JSON formatted string.

kadi.lib.utils.formatted_json(data, ensure_ascii=False, sort_keys=True)[source]

Serialize data to a user-readable JSON formatted string.

Uses the JSON encoder provided by Flask, which can deal with some additional types.

Parameters:

data – The data to serialize.
ensure_ascii – (optional) Whether to escape non-ASCII characters.
sort_keys – (optional) Whether to sort the output of dictionaries by key.

Returns:

The JSON formatted string.

kadi.lib.utils.is_special_float(value)[source]

Check if a float value is a special value, i.e. nan or inf.

Parameters:: value – The float value to check.
Returns:: True if the value is a special float value, False otherwise.

kadi.lib.utils.is_iterable(value, include_string=False)[source]

Check if a value is an iterable.

Parameters:

value – The value to check.
include_string – (optional) Flag indicating whether a string value should be treated as a valid iterable or not.

Returns:

True if the value is iterable, False otherwise.

kadi.lib.utils.is_quoted(value)[source]

Check if a string value is quoted, i.e. surrounded by double quotes.

Parameters:: value – The string value to check.
Returns:: True if the value is quoted, False otherwise.

kadi.lib.utils.is_http_url(value)[source]

Check if a string represent a valid HTTP URL.

Parameters:: value – The string value to check.
Returns:: True if the value represents a HTTP URL, False otherwise.

kadi.lib.utils.find_dict_in_list(dict_list, key, value)[source]

Find a dictionary with a specific key and value in a list.

Parameters:

dict_list – A list of dictionaries to search.
key – The key to search for.
value – The value to search for.

Returns:

The dictionary or None if it was not found.

kadi.lib.utils.flatten_list(values)[source]

Flatten a list of lists.

Flattens a list containing single values or nested lists. Multiple layers of nested lists are not supported.

Parameters:: values – List containing single values or nested lists.
Returns:: The flattened list.

kadi.lib.utils.as_list(value)[source]

Wrap a value inside a list if the given value is not a list already.

Parameters:: value – The value to potentially wrap.
Returns:: The original or wrapped value. If the given value is None, the value will always be returned as is.

kadi.lib.utils.has_capabilities(*capabilities)[source]

Check if the current Kadi instance has all given capabilities.

Parameters:: capabilities – One or more capabilities to check.
Returns:: True if all given capabilities are available, False otherwise.

Validation

kadi.lib.validation.validator(exception_class)[source]

Decorator to wrap a validation function.

Handles errors of type KadiValidationError and reraises another customizable exception using the same message.

Parameters:: exception_class – The exception to raise in case of validation failure.

kadi.lib.validation.validate_uuid(uuid_string, version=4)[source]

Validate a string against a specific UUID version.

Parameters:

uuid_string – The UUID string.
version – (optional) The UUID version.

Raises:

KadiValidationError – If the validation fails.

kadi.lib.validation.validate_identifier(value)[source]

Validate the format of an identifier.

Identifiers can be used to give resources a human-readable, unique identification. An identifier is restricted to lowercase alphanumeric characters, hyphens and underscores.

Parameters:: value – The identifier string.
Raises:: KadiValidationError – If the validation fails.

kadi.lib.validation.validate_mimetype(value)[source]

Validate the format of a MIME type.

A MIME type has to start with at least one alphabetical character, followed by a forward slash, followed by lowercase alphanumeric characters or the special characters "-", "+" or ".".

Parameters:: value – The MIME type string.
Raises:: KadiValidationError – If the validation fails.

kadi.lib.validation.validate_username(value)[source]

Validate the format of a local username.

Local usernames are restricted to lowercase alphanumeric characters with single hyphens or underscores in between.

Parameters:: value – The username string.
Raises:: KadiValidationError – If the validation fails.

kadi.lib.validation.validate_iri(value)[source]

Validate an IRI reference.

Note that this function simply checks against a set of invalid characters.

Parameters:: value – The IRI reference string.
Raises:: KadiValidationError – If the validation fails.

Web

class kadi.lib.web.IdentifierConverter(map: Map, *args: t.Any, **kwargs: t.Any)[source]

Bases: BaseConverter

Custom URL converter for identifiers.

Validates arguments according to kadi.lib.validation.validate_identifier(), but also allows and converts/strips uppercase characters and surrounding whitespaces.

kadi.lib.web.flash_danger(message)[source]

Flash a danger message to the next request.

Uses Flask’s flash() function with a fixed "danger" category.

Parameters:: message – The message to be flashed.

kadi.lib.web.flash_info(message)[source]

Flash an info message to the next request.

Uses Flask’s flash() function with a fixed "info" category.

Parameters:: message – The message to be flashed.

kadi.lib.web.flash_success(message)[source]

Flash a success message to the next request.

Uses Flask’s flash() function with a fixed "success" category.

Parameters:: message – The message to be flashed.

kadi.lib.web.flash_warning(message)[source]

Flash a warning message to the next request.

Uses Flask’s flash() function with a fixed "warning" category.

Parameters:: message – The message to be flashed.

kadi.lib.web.get_locale()[source]

Get the current locale.

The locale will be retrieved from a cookie as defined in kadi.lib.constants.LOCALE_COOKIE_NAME.

Returns:: The current locale. If no valid locale could be found, the default locale will be returned.

kadi.lib.web.get_preferred_locale()[source]

Get the preferred locale of the current user’s client.

Returns:: The preferred locale. If no matching locale could be found, the default locale will be returned.

kadi.lib.web.encode_filename(filename)[source]

Encode a file name for use in a Content-Disposition header.

Parameters:: filename – The name of the file to encode.
Returns:: A dictionary containing an ASCII version of the file name as "filename" and optionally an UTF-8 version as "filename*".

kadi.lib.web.download_bytes(data, *, filename, mimetype=None, as_attachment=True, content_length=None)[source]

Send bytes or an iterable of bytes to a client as a file.

Parameters:

data – The binary data to send.
filename – The name of the file to send.
mimetype – (optional) The MIME type of the file to send. Defaults to a MIME type based on the given filename or the default MIME type as defined in kadi.lib.constants.MIMETYPE_BINARY if it cannot be guessed.
as_attachment – (optional) Whether to send the file as an attachment. Note that setting this parameter to False may pose a security risk, depending on the stream contents, client and context.
content_length – (optional) The content length of the data in bytes. If not provided, it will be calculated automatically when using bytes or io.BytesIO and is otherwise omitted in the response.

Returns:

The response object.

kadi.lib.web.url_for(endpoint, _ignore_version=False, **values)[source]

Generate an URL based on a given endpoint.

Wraps Flask’s url_for function with additional support for generating the correct URLs when using API versioning. Additionally, generated URLs are always external, i.e. absolute.

Parameters:

endpoint – The endpoint (name of the function) of the URL.
_ignore_version – (optional) Flag indicating whether the API version should be ignored when building the URL in API requests.
**values – The variable arguments of the URL rule.

Returns:

The generated URL string.

kadi.lib.web.static_url(filename)[source]

Generate a static URL for a given filename.

Will make use of the MANIFEST_MAPPING if it is defined in the application’s configuration and if an entry exists for the given filename.

Parameters:: filename – The name of the file to include in the URL.
Returns:: The generated URL string.

kadi.lib.web.get_next_url(fallback=None)[source]

Get the validated target URL to redirect a user to after login.

The target URL will be retrieved from the session via kadi.lib.constants.SESSION_KEY_NEXT_URL.

Parameters:: fallback – (optional) The fallback URL to use in case the target URL is invalid or could not be found. Defaults to the index page.
Returns:: The validated target URL.

kadi.lib.web.get_error_description(status_code)[source]

Get an error description corresponding to an HTTP status code.

Parameters:: status_code – The HTTP status code.
Returns:: The error description.

kadi.lib.web.html_error_response(status_code, description=None)[source]

Return an HTML error response to a client.

Parameters:

status_code – The HTTP status code of the response.
description – (optional) The error description. Defaults to the result of get_error_description() using the given status code.

Returns:

The HTML response.

kadi.lib.web.get_apidoc_meta(func)[source]

Get the API documentation meta dictionary of a view function.

If not present yet, a corresponding dictionary will be created first as an attribute of the given view function.

Parameters:: func – The view function.
Returns:: The newly created or existing meta dictionary.

kadi.lib.web.qparam(name, location=None, multiple=False, default='', parse=None, description='')[source]

Decorator to parse a query parameter in a view function.

Convenience decorator to retrieve and parse a specified query parameter from the current request. The decorator can be applied multiple times. Each parameter will be injected into the decorated function as part a dictionary inside the keyword argument qparams. The dictionary maps each given parameter name to its respective value.

The information about the query parameter is also used when generating the API documentation.

Parameters:

name – The name of both the query parameter and the dictionary key that is injected into the decorated function.
location – (optional) The name of the query parameter to use instead of name.
multiple – (optional) Flag indicating whether the query parameter can be specified multiple times and should be retrieved as list value.
default – (optional) The default value or a callable returning a default value to use in case the query parameter is missing and multiple is False, otherwise the default value will always be an empty list.
parse – (optional) A callable or list of callables to parse the parameter value if it is not missing. Each callable must take and return a single parameter value. If parsing fails with a ValueError, the default value is taken instead if multiple is False, otherwise each invalid value is removed from the resulting list.
description – (optional) A description of the query parameter, which is only used when generating the API documentation. Supports reST syntax.

kadi.lib.web.paginated(page_max=None, per_page_max=100)[source]

Decorator to parse paginated query parameters.

Convenience decorator to get and parse the query parameters "page" and "per_page" from the current request. The former defaults to 1 while the latter defaults to 10 if no valid integer values were found. Both parameters will be injected into the decorated function as keyword arguments page and per_page.

The information about the query parameters is also used when generating the API documentation.

Parameters:

page_max – (optional) The maximum possible value of the "page" parameter.
per_page_max – (optional) The maximum possible value of the "per_page" parameter.