API

zsl.application.containers

Module containing various IoC containers. IoC container holdsa collection of modules.

zsl.application.containers.celery_container

class zsl.application.containers.celery_container.CeleryContainer

Configuration for celery application.

celery_cli

alias of CeleryCliModule

worker

alias of CeleryTaskQueueMainWorkerModule

class zsl.application.containers.celery_container.CeleryStandAloneContainer

Configuration for application run with help of celery cli tools.

celery_cli

alias of CeleryCliModule

worker

alias of CeleryTaskQueueOutsideWorkerModule

zsl.application.containers.container

class zsl.application.containers.container.IoCContainer

Collection of DI modules set as class attributes.

This is used for declarative DI configuration, which can be easily extended upon.

>>> class MyContainer(IoCContainer):
...   cache = MyCacheModule
...   redis = MyRedisModule

>>> class MyTestContainer(MyContainer):
...   redis = MockRedisModule

classmethod modules()

Collect all the public class attributes.

All class attributes should be a DI modules, this method collects them and returns as a list.

Returns:

list of DI modules

Return type:

list[Union[Module, Callable]]

zsl.application.containers.core_container

class zsl.application.containers.core_container.CoreContainer

Modules for basic Zsl application.

cache

alias of RedisCacheInjectionModule

cli

alias of CliModule

context

alias of DefaultContextModule

database

alias of AlchemyModule

error_handler

alias of ErrorHandlerModule

logger

alias of LoggerModule

task_router

alias of TaskRouterModule

zsl.application.containers.web_container

class zsl.application.containers.web_container.WebContainer

Configuration for web application.

web_context

alias of WebContextModule

zsl.application.error_handler

This module does the error handling. It allows users to register an error handler for a given exception type. It also provides default error handlers.

class zsl.application.error_handler.DefaultErrorHandler

ERROR_CODE = 'UNKNOWN_ERROR'

ERROR_MESSAGE = 'An error occurred!'

can_handle(e)

Indicator if the handler is able to handle the given exception e.

Parameters:

e – The exception that shall be determined if can be handled by the handler.

Returns:

True or False depending on whether the handler can/should handle the method.

handle(ex)

Handle the exception.

Parameters:

e – The handled exception.

Returns:

The error response for the exception.

class zsl.application.error_handler.ErrorResponse(code, message)

class zsl.application.error_handler.ForbiddenErrorHandler

ERROR_CODE = 'FORBIDDEN'

can_handle(e)

Indicator if the handler is able to handle the given exception e.

Parameters:

e – The exception that shall be determined if can be handled by the handler.

Returns:

True or False depending on whether the handler can/should handle the method.

handle(e)

Handle the exception.

Parameters:

e – The handled exception.

Returns:

The error response for the exception.

class zsl.application.error_handler.ModelConversionErrorHandler

ERROR_CODE = 'INVALID_REQUEST'

can_handle(e)

Indicator if the handler is able to handle the given exception e.

Parameters:

e – The exception that shall be determined if can be handled by the handler.

Returns:

True or False depending on whether the handler can/should handle the method.

handle(e)

Handle the exception.

Parameters:

e – The handled exception.

Returns:

The error response for the exception.

class zsl.application.error_handler.RoutingErrorHandler

ERROR_CODE = 'NOT_FOUND'

can_handle(e)

Indicator if the handler is able to handle the given exception e.

Parameters:

e – The exception that shall be determined if can be handled by the handler.

Returns:

True or False depending on whether the handler can/should handle the method.

handle(ie)

Handle the exception.

Parameters:

e – The handled exception.

Returns:

The error response for the exception.

zsl.application.error_handler.error_handler(f)

Default error handler.

• On server side error shows a message ‘An error occurred!’ and returns 500 status code.

• Also serves well in the case when the resource/task/method is not found - returns 404 status code.

zsl.application.error_handler.register(e: ErrorHandler | ErrorProcessor) → None

zsl.application.initialization_context

The modules deals with the initialization of the basics - python path and then loads and initializes the application if necessary.

class zsl.application.initialization_context.InitializationContext(unit_test=False, initializers=None)

get_unit_testing()

initialize()

property is_initialized

property unit_testing

zsl.application.initializers

Module for application parts needed to initialize on start.

zsl.application.initializers.library_initializer

class zsl.application.initializers.library_initializer.LibraryInitializer

Add vendor modules to current path.

static initialize(config)

zsl.application.initializers.library_initializer.append_paths(path, vendor_modules)

zsl.application.initializers.service_initializer

class zsl.application.initializers.service_initializer.ServiceInitializer

Add outside services to application injector.

static initialize(config)

Initialize method.

Parameters:

config (Config) – current application config, injected

zsl.application.initializers.unittest_initializer

class zsl.application.initializers.unittest_initializer.UnitTestInitializer

Initializer handling the unit test settings.

static initialize(config)

static is_unit_testing(ctx)

zsl.application.modules

Module for DI modules.

zsl.application.modules.alchemy_module

class zsl.application.modules.alchemy_module.AlchemyModule

Adds SQLAlchemy to current configuration.

provide_engine(config: Config) → Engine

provide_session_holder(engine: Engine) → SessionHolder

provide_transaction_holder_factory() → TransactionHolderFactory

class zsl.application.modules.alchemy_module.DefaultTransactionHolderFactory

create_transaction_holder() → TransactionHolder

class zsl.application.modules.alchemy_module.EmptyTransactionalHolder

property session

class zsl.application.modules.alchemy_module.SessionFactory(session_holder: SessionHolder)

Creates a db session with an open transaction.

create_session() → Session

class zsl.application.modules.alchemy_module.SessionHolder(sess_cls)

class zsl.application.modules.alchemy_module.TransactionHolder

Holds the transaction and session.

Works like a SQAlchemy session proxy.

append_transaction_callback(callback)

begin()

close()

commit()

has_session() → bool

property in_transaction

inject_session(session: Session) → None

Used to set the session in the @transactional decorator.

rollback()

run_transaction_callbacks()

property session: Session

class zsl.application.modules.alchemy_module.TransactionHolderFactory

abstract create_transaction_holder() → TransactionHolder

zsl.application.modules.cache_module

class zsl.application.modules.cache_module.RedisCacheInjectionModule

Adds cache modules into to current configuration using reddis as backend.

configure(binder: Binder) → None

Initializer of the cache - creates the Redis cache module as the default cache infrastructure. The module is bound to RedisCacheModule and CacheModule keys. The initializer also creates RedisIdHelper and bounds it to RedisIdHelper and IdHelper keys.

Parameters:

binder (Binder) – The binder object holding the binding context, we add cache to the binder.

zsl.application.modules.celery_module

class zsl.application.modules.celery_module.CeleryCli(zsl_cli: ZslCli = None)

celery()

class zsl.application.modules.celery_module.CeleryCliModule

configure(binder)

Override to configure bindings.

class zsl.application.modules.celery_module.CeleryTaskQueueMainWorkerModule

Adds celery worker to current configuration.

configure(binder)

Override to configure bindings.

class zsl.application.modules.celery_module.CeleryTaskQueueOutsideWorkerModule

Adds celery outside worker to current configuration.

Outside worker is meant to run with help of celery cli tools.

configure(binder)

Override to configure bindings.

class zsl.application.modules.cli_module.CliModule

Adds Alembic support for migrations.

configure(binder: Binder) → None

Override to configure bindings.

class zsl.application.modules.cli_module.ZslCli

property cli: ZslCli

class zsl.application.modules.cli_module.ZslGenerateCli(zsl_cli: ZslCli)

class zsl.application.modules.cli_module.ZslTaskCli(zsl_cli: ZslCli)

class zsl.application.modules.cli_module.ZslTestCli(zsl_cli: ZslCli)

zsl.application.modules.context_module

Application context describes the environment in which the application is runand is responsible for specific initializations on it.

class zsl.application.modules.context_module.DefaultContextModule

Adds default application context to current configuration.

configure(binder)

Override to configure bindings.

class zsl.application.modules.context_module.TestContextModule

Test application context to current configuration

class zsl.application.modules.context_module.WorkerContextModule

Worker application context to current configuration.

zsl.application.modules.context_module.default_initializers = (<class 'zsl.application.initializers.library_initializer.LibraryInitializer'>, <class 'zsl.application.initializers.service_initializer.ServiceInitializer'>)

Initializers used in all applications

zsl.application.modules.context_module.unittest_initializers = (<class 'zsl.application.initializers.library_initializer.LibraryInitializer'>, <class 'zsl.application.initializers.service_initializer.ServiceInitializer'>, <class 'zsl.application.initializers.unittest_initializer.UnitTestInitializer'>)

Initializers used in unit test applications

class zsl.application.modules.error_handler_module.ErrorHandlerModule

configure(binder: Binder) → None

Override to configure bindings.

provide_error_config(config: Config) → ErrorConfiguration

zsl.application.modules.logger_module

class zsl.application.modules.logger_module.LoggerModule

Configure the application logger.

LOGGING_CONFIG_NAME = 'LOGGING'

configure(binder: Binder) → None

Override to configure bindings.

configure_logging()

zsl.application.modules.task_router_module

class zsl.application.modules.task_router.TaskRouterModule

Adds task router to current configuration.

configure(binder)

Override to configure bindings.

provide_task_configuration(config)

task_provider = <injector.ClassProvider object>

class zsl.application.modules.web.configuration.MethodConfiguration(package=None, packages=None, url_prefix='method')

property packages

property url_prefix

class zsl.application.modules.web.cors.CORSConfiguration(origin: str = '', allow_headers: List[str] | str = None, expose_headers: List[str] | str = None, max_age: int = 21600)

DEFAULT_ALLOW_HEADERS = ['Accept', 'Origin', 'Content-Type', 'Authorization']

DEFAULT_EXPOSE_HEADERS = ['Location', 'X-Total-Count', 'Link']

DEFAULT_MAX_AGE = 21600

property allow_headers: List[str] | str

property expose_headers: List[str] | str

property max_age: int

property origin: str

class zsl.application.modules.web.web_context_module.WebCli(zsl_cli: ZslCli)

property web

class zsl.application.modules.web.web_context_module.WebContextModule

Adds web application context to current configuration.

configure(binder: Binder) → None

Override to configure bindings.

provide_cors_configuration(config: Config) → CORSConfiguration

provide_web_cli()

provide_web_handler()

class zsl.application.modules.web.web_context_module.WebHandler

run_web(flask: ServiceApplication, host: str = '127.0.0.1', port: int = 5000, **options: Any) → None

Alias for Flask.run

class zsl.application.modules.web.web_context_module.WebInitializer

Initialize the web application.

static initialize()

Import in this form is necessary so that we avoid the unwanted behavior and immediate initialization of the application objects. This makes the initialization procedure run in the time when it is necessary and has every required resources.

zsl.application.modules.web.web_context_module.web_initializers = (<class 'zsl.application.initializers.library_initializer.LibraryInitializer'>, <class 'zsl.application.initializers.service_initializer.ServiceInitializer'>, <class 'zsl.application.modules.web.web_context_module.WebInitializer'>)

Initializers used in unit web applications

zsl.application.service_application

The module contains the main Zsl class - ServiceApplication. It is responsible for gluing everything together.

There is a simple profile support. Profile is usually tied to a configuration file for an environment. Usually environments slightly differ in settings - especially connection strings. These twists can be managed via additional configuration files with the cfg extension usually placed in the settings package. The name of the environment configuration file is controlled via ZSL_SETTINGS environment variable.

zsl.application.service_application.SETTINGS_ENV_VAR_NAME

Name of the environment variable to be read for the profile configuration file.

class zsl.application.service_application.ServiceApplication(import_name, static_url_path=None, static_folder='static', static_host=None, host_matching=False, subdomain_matching=False, template_folder='templates', instance_path=None, instance_relative_config=False, root_path=None, modules=None, version=None, config_object=None, default_settings_module='settings.default_settings')

Atteq Service Flask application.

VERSION = '1.0.0'

property app_version

get_initialization_context()

get_version()

property injector: Injector

is_initialized()

property version

property zsl_version

zsl.application.service_application.get_settings_from_profile(profile: str, profile_dir: Any = None) → str

“Returns the configuration file path for the given profile.

Parameters:

• profile – Profile name to be used.

• profile_dir – The directory where the profile configuration file should reside. It may be also a module, and then the directory of the module is used.

Returns:

Configuration file path.

zsl.application.service_application.set_profile(profile)

Sets the current profile.

Parameters:

profile – The profile to be used.

zsl.cache.cache_module

class zsl.cache.cache_module.CacheModule

Cache abstraction layer - module for caching.

abstract contains_key(key)

Check if the key is contained in the cache.

abstract contains_list(key)

Check if the key is in the cache.

abstract get_key(key)

Returns the value associated with the key.

TODO this should be named get_by_key or just simply get(key), get_key sounds like it would return a key

abstract get_list(key)

Returns the list associated with the key.

abstract invalidate_by_glob(glob)

Invalidates the keys by given glob.

Parameters:

glob (string) – All the keys matching the given glob will be invalidated.

abstract invalidate_key(key)

Deletes the key.

abstract set_key(key, value, timeout)

Saves the key-value pair for timeout seconds.

zsl.cache.id_helper

class zsl.cache.id_helper.IdHelper

abstract check_key(key)

abstract check_page(page_key)

abstract fill_page(page_key, data, timeout, encoder=<function encoder_identity>, model_key_generator=<function model_key_generator>)

abstract gather_page(page_key, decoder=<function decoder_identity>)

abstract get_key(key)

abstract invalidate_key(key)

abstract invalidate_keys_by_prefix(key_prefix)

abstract set_key(key, value, timeout)

zsl.cache.id_helper.app_model_decoder_fn(model_key, x)

zsl.cache.id_helper.app_model_encoder_fn(x)

zsl.cache.id_helper.create_key_class_prefix(cls)

zsl.cache.id_helper.create_key_object_prefix(obj)

zsl.cache.id_helper.decoder_identity(module_name, x)

zsl.cache.id_helper.encoder_identity(x)

zsl.cache.id_helper.model_key_generator(model)

zsl.cache.id_helper.model_key_hint_extractor(key)

zsl.cache.redis_cache_module

class zsl.cache.redis_cache_module.RedisCacheModule(app, config)

Abstraction layer for caching.

append_to_list(key, value)

contains_key(key)

Check if the key is contained in the cache.

contains_list(key)

Check if the key is in the cache.

get_key(key)

Returns the value associated with the key.

TODO this should be named get_by_key or just simply get(key), get_key sounds like it would return a key

get_list(key)

Returns the list associated with the key.

invalidate_by_glob(glob)

Invalidates the keys by given glob.

Parameters:

glob (string) – All the keys matching the given glob will be invalidated.

invalidate_key(key)

Deletes the key.

set_key(key, value, timeout)

Saves the key-value pair for timeout seconds.

set_key_expiration(key, timeout)

zsl.cache.redis_id_helper

class zsl.cache.redis_id_helper.RedisIdHelper(config, redis_cache_module)

CACHE_DEFAULT_TIMEOUT = 300

check_key(key)

check_page(page_key)

fill_page(page_key, data, timeout, encoder=<function encoder_identity>, model_key_generator=<function model_key_generator>)

gather_page(page_key, decoder=<function decoder_identity>)

get_key(key)

get_timeout(key, value, timeout)

invalidate_key(key)

invalidate_keys_by_prefix(key_prefix)

set_key(key, value, timeout)

exception zsl.configuration.InvalidConfigurationException

class zsl.constants.HttpHeaders(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

CONTENT_TYPE = 'Content-Type'

class zsl.constants.MimeType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

APPLICATION_JSON = 'application/json'

class zsl.contrib.alembic.alembic_config.AlembicConfiguration(alembic_directory)

alembic_directory

Alias for field number 0

zsl.contrib.alembic.alembic_module

Alembic module is responsible for handling database migrations for SqlAlchemy database backend. The complete documentation of the Alembic project, all the options and command may be found Alembic website.

If you need database migrations just install alembic extra of zsl package. Then you may use alembic commands from ZSL cli interface via alembic command.

Just use a standard way of calling cli with a container having AlembicModule.

class MyApplicationContainer(WebContainer):
    alembic = AlembicModule


@inject(zsl_cli=ZslCli)
def main(zsl_cli: ZslCli) -> None:
    zsl_cli()


if __name__ == "__main__":
    os.environ[SETTINGS_ENV_VAR_NAME] = ... or set_profile('my-profile')
    app = Zsl(__name__, modules=MyApplicationContainer.modules())
    main()

Configuration

To configure one needs to provide a valid AlembicConfiguration usually in default_settings.py

ALEMBIC = AlembicConfiguration(alembic_directory="...")

Then all the required files will be stored in the directory. Also the alembic.ini file will be stored there although this is not a standard way.

When using AlembicCli interface the alembic commands will be functional though.

A little setup is required so that the connection is opened correctly. In alembic.ini in the given directory remove the following line.

sqlalchemy.url = driver://user:pass@localhost/dbname

Also set script_location to . so that it does not rely on absolute paths

script_location = .

Then in env.py, since we removed the sqlalchemy.url we need to use a ZSL engine created using the correct urls from settings. To do so define the routines run_migrations_offline and run_migrations_online so that they use a correct Engine.

@inject(zsl_config=Config)
def run_migrations_offline(zsl_config):
    url = zsl_config['DATABASE_URI']
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)

    with context.begin_transaction():
        context.run_migrations()


@inject(engine=Engine)
def run_migrations_online(engine):
    with engine.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

class zsl.contrib.alembic.alembic_module.AlembicCli(zsl_cli: ZslCli)

Alembic Cli interface support.

property alembic

call_alembic(args: List[str], alembic_cfg: AlembicConfiguration) → None

class zsl.contrib.alembic.alembic_module.AlembicModule

Adds Alembic support for migrations.

ALEMBIC_CONFIG_NAME = 'ALEMBIC'

configure(binder: Binder) → None

Override to configure bindings.

provide_alembic_configuration(config: Config) → AlembicConfiguration

class zsl.contrib.sentry.sentry_config.SentryConfiguration(dsn, environment=None, tags=None, register_logging_handler=True, sentry_logging_handler_level=40)

zsl.contrib.modules.sentry_module

Sentry module is responsible for handling errors and sending them to a sentry backend. Just provide a DSN (url) of the backend and you are all set up.

To add the Sentry support just add SentryModule to your container.

class MyApplicationContainer(WebContainer):
    sentry = SentryModule

Configuration

To configure one needs to provide a valid SentryConfiguration usually in default_settings.py or in your profiles.

SENTRY = SentryConfiguration(dsn="...")

The following options are available:

1. dsn: DSN/URl of your Sentry project,

2. environment: the environment (release defaults to the version of your application),

3. tags: additional tags str to str dictionary,

4. register_sentry_logging_handler: bool value indicating if logging handler should be added to logging,

5. sentry_handler_logging_level: minimal log level of log messages sent to sentry logging handler.

To test the functionality use CLI

python app.py sentry test

class zsl.contrib.sentry.sentry_module.SentryCli(zsl_cli: ZslCli)

property sentry

class zsl.contrib.sentry.sentry_module.SentryErrorProcessor(config: SentryConfiguration, zsl: ServiceApplication)

handle(e)

class zsl.contrib.sentry.sentry_module.SentryModule

Adds Sentry support.

SENTRY_CONFIG_NAME = 'SENTRY'

configure(binder: Binder) → None

Override to configure bindings.

provide_sentry_configuration(config: Config) → SentryConfiguration

zsl.db.helpers.app_model(model)

Just a convenience shortcut for model.get_app_model()

zsl.db.helpers.app_model_or_none(raw_model)

Transforms raw_model to its application model. Function can handle None value.

zsl.db.helpers.app_models(raw_models)

For a list of raw models return the list of the corresponding app models.

zsl.db.helpers.visit_app_model(model, visitor)

zsl.db.helpers.nested

zsl.db.helpers.nested.get_nested_field_name(field)

zsl.db.helpers.nested.nested_model(model: ModelBase, nested_fields: Any) → AppModel | None

Return zsl.db.model.app_model import AppModel with the nested models attached. nested_fields can be a simple list as model fields, or it can be a tree definition in dict with leafs as keys with None value

zsl.db.helpers.nested.nested_models(models, nested_fields)

For a list of models apply get_nested_model with given nested_fields.

zsl.db.helpers.pagination

class zsl.db.helpers.pagination.Pagination(pagination: PaginationRequest | Dict[str, str | int] | None = None)

Pagination support. Allows to paginate a query. There are two choices.

1. paginate() - paginates a query and obtains the count automatically.

2. apply_pagination() - paginates a query and assumes that record count is set in advance.

apply_pagination(q: Query) → Query

Filters the query so that a given page is returned. The record count must be set in advance. :param q: Query to be paged. :return: Paged query.

property offset

property page_size

paginate(q: Query) → Query

Filters the query so that a given page is returned. The record count is computed automatically from query. :param q: Query to be paged. :return: Paged query.

property record_count

class zsl.db.helpers.pagination.PaginationRequest(raw=None, id_name='id', hints=None)

static create(page_no: int = 1, page_size: int = 25) → PaginationRequest

class zsl.db.helpers.pagination.PaginationResponse(record_count: int, page_size: int, pagination: PaginationRequest)

zsl.db.helpers.query_filter

class zsl.db.helpers.query_filter.OperatorBetween

static apply(q, attr, v)

class zsl.db.helpers.query_filter.OperatorCompareDates

Compares only dates, year is not taken into account. Compared date value must be string in format ‘%m-%d’

static apply(q, attr, v)

class zsl.db.helpers.query_filter.OperatorEq

static apply(q, attr, v)

class zsl.db.helpers.query_filter.OperatorLeftLike

Left side of string is like …

static apply(q, attr, v)

class zsl.db.helpers.query_filter.OperatorLike

static apply(q, attr, v)

class zsl.db.helpers.query_filter.OperatorNeq

static apply(q, attr, v)

class zsl.db.helpers.query_filter.OperatorRightLike

Right side of string is like …

static apply(q, attr, v)

class zsl.db.helpers.query_filter.QueryFilter(query_filter, mappings=None, allow_null=False)

Helper class for applying filter criteria to query.

apply_query_filter(q, cls)

class zsl.db.helpers.query_filter.RelationshipOperatorContains

static apply(q, attr, v)

zsl.db.helpers.query_helper

class zsl.db.helpers.query_helper.QueryHelper(cls, query_filter, pagination, sorter)

execute(q)

get_pagination()

get_sorter()

zsl.db.helpers.sorter

class zsl.db.helpers.sorter.Sorter(sorter, mappings=None)

Helper class for applying ordering criteria to query.

apply_sorter(q, cls)

get_fields()

get_orders()

is_enabled()

zsl.db.model.app_model

class zsl.db.model.app_model.AppModel(raw, id_name='id', hints=None)

AppModel’s are used as a thin and simple communication objects. Also they can be saved into cache. Basically they are known as Data Transfer Objects or DTOs.

__init__(raw, id_name='id', hints=None)

The application model model constructor.

Parameters:

• raw – Dictionary of properties of the raw data.

• id_name – Name of the identifier property.

• hints –

  Tells which of the raw attributes are date or datetime string and what is theirs format. Example: ``` {

  DATE_DATA: { ‘birthday’: ‘%d.%m.%Y’ }, DATETIME_DATA: { ‘created’: ‘%Y-%m-%d %H:%M:%S’ }

  this attributes are then saved in the standard zsl service date/datetime format (consult zsl.utils.date_helper for more.)

static convert(v)

get_attributes()

get_id()

zsl.db.model.app_model_json_decoder

zsl.db.model.app_model_json_decoder.get_json_decoder(full_class_name, hints=None)

zsl.db.model.app_model_json_encoder

class zsl.db.model.app_model_json_encoder.AppModelJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

default(o)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

zsl.db.model.raw_model

class zsl.db.model.raw_model.ModelBase

get_app_model(id_name='id', hints=None)

update(app_model, forbidden_keys=None, inverse=False)

Updates the raw model. Consult zsl.utils.model_helper.update_model.

zsl.db.model.sql_alchemy

one base to rule them all

class zsl.errors.ErrorConfiguration(handlers: List[ErrorHandler] = None, processors: List[ErrorProcessor] = None, use_flask_handler: bool = False)

The configuration object for error handling.

property handlers: List[ErrorHandler]

property processors: List[ErrorProcessor]

property use_flask_handler: bool

In case of web requests, flasks provides a convenient way of exception handling. This handler shows the stack trace, etc. On the other hand this setting will turn of ZSL’s exception handling for web request.

Returns:

Status

class zsl.errors.ErrorHandler

Custom error handler providing a response on a particular error.

abstract can_handle(e: ErrorHandler) → bool

Indicator if the handler is able to handle the given exception e.

Parameters:

e – The exception that shall be determined if can be handled by the handler.

Returns:

True or False depending on whether the handler can/should handle the method.

abstract handle(e: ErrorHandler) → bool

Handle the exception.

Parameters:

e – The handled exception.

Returns:

The error response for the exception.

class zsl.errors.ErrorProcessor

Custom error processor handling an error state caused by an error. For example an error processor may only log an error, send an email, etc. The main difference between an ErrorProcessor and an ErrorHandler is that the latter one returns a response. There may be only a single ErrorHandler returning whereas there may be more ErrorProcessors handling the same error.

abstract handle(e: ErrorHandler) → None

exception zsl.errors.ZslError

A base class intended for all exceptions thrown by ZSL or ZSL application.

This script is to specify an appropriate module working with Python 3 (python3_gearman) or Python 2 (gearman).

zsl.interface.celery.worker

Implementation of celery workers.

class zsl.interface.celery.worker.CeleryTaskQueueMainWorker

Worker implementation for Celery task queue.

run(argv: list[str])

Run the celery worker cmd with given arguments from the list.

Note: the first argument should be “worker”.

stop_worker()

Stop the worker.

class zsl.interface.celery.worker.CeleryTaskQueueOutsideWorker(app: ServiceApplication = None, config: Config = None)

Celery worker used only for task execution.

This can be used when the worker is managed with some other application, like celery worker or celery multi.

run()

Run the worker.

stop_worker()

Stop the worker.

class zsl.interface.celery.worker.CeleryTaskQueueWorkerBase(app: ServiceApplication = None, config: Config = None)

Base class for celery task queue worker.

It contains only the task execution part of worker.

execute_celery_task(job_data: dict) → JobResult

Creates job from task and executes the job.

Parameters:

job_data – job’s data

Returns:

job results

Return type:

dict

zsl.interface.celery.worker.create_celery_app(config: Config = None) → Celery

zsl.interface.cli

The module is responsible for creating the CLI interface of ZSL and provides the main click group for all the CLI commands and groups. If any command is to be added add it to this group cli defined in this module.

zsl.interface.resource

Helper types describing resource results for resource router.

class zsl.interface.resource.ResourceResult(body, count, links, status, location, expose_headers)

body

Alias for field number 0

count

Alias for field number 1

expose_headers

Alias for field number 5

links

Alias for field number 2

location

Alias for field number 4

status

Alias for field number 3

zsl.interface.task

exception zsl.interface.task.ModelConversionError(obj, attribute)

property attribute

property obj

class zsl.interface.task.RequestJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

default(o)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)

zsl.interface.task.create_simple_model(name, items, defaults=None, parent=<class 'object'>, model_module=None)

zsl.interface.task.create_task(task_path, task_router)

zsl.interface.task.exec_task(task_path, data)

Execute task.

Parameters:

• task_path (str|Callable) – task path

• data (Any) – task’s data

Returns:

zsl.interface.task.fill_model_with_payload(data: Dict[str, Any], obj: object) → None

zsl.interface.task.payload_into_model(model_type: Callable, argument_name: str = 'request', remove_data: bool = True) → Callable

zsl.interface.task_queue

This module contains interfaces and functions to handle task queues in ZSL. A task queue handles asynchronous and distributed code executions.

class zsl.interface.task_queue.JobResult

data: Any

task_name: Any

exception zsl.interface.task_queue.KillWorkerException

If any task raises this exception a standalone worker will be killed.

class zsl.interface.task_queue.TaskQueueWorker(app: ServiceApplication = None, config: Config = None)

Task queue worker abstraction.

A task queue worker is responsible for communicating with a task queue and executing any task given by it. It should be able to run as a stand alone application.

execute_job(job: Job) → JobResult

Execute job given by the task queue.

Parameters:

job (Job) – job

Returns:

task result

Return type:

dict

handle_exception(e: Exception, task_path: str) → JobResult

Handle exception raised during task execution.

Parameters:

• e (Exception) – exception

• task_path (str) – task path

Returns:

exception as task result

Return type:

dict

abstract run(*args, **kwargs)

Run the worker.

abstract stop_worker()

Stop the worker.

zsl.interface.task_queue.execute_job(job: Job, app: ServiceApplication = None, task_router: TaskRouter = None) → JobResult

Execute a job.

Parameters:

• job (Job) – job to execute

• app (ServiceApplication) – service application instance, injected

• task_router (TaskRouter) – task router instance, injected

Returns:

task result

Return type:

dict

zsl.interface.task_queue.run_worker(*args: Any, **kwargs: Any) → None

Run the app as a task queue worker.

The worker instance is given as a DI module.

zsl.interface.webservice.importer

zsl.interface.webservice.performers.default

zsl.interface.web.performers.default.create_not_found_mapping(app)

zsl.interface.webservice.performers.method

zsl.interface.web.performers.method.call_exposers_in_method_packages()

zsl.interface.webservice.performers.resource

zsl.interface.web.performers.resource.create_resource_mapping(app)

zsl.interface.webservice.performers.task

zsl.interface.web.performers.task.create_task_mapping(app: ServiceApplication, task_configuration: TaskConfiguration) → None

zsl.interface.web.performers.task.perform_web_task(namespace, path, task_router=None)

zsl.interface.web.utils.execution.convert_to_web_response(f)

zsl.interface.web.utils.execution.execute_web_task(job_context: WebJobContext, callable: Callable, app: ServiceApplication) → Response

zsl.interface.web.utils.execution.notify_responders(f)

zsl.interface.web.utils.request_data.extract_data(request)

zsl.interface.webservice.utils.response_headers

zsl.interface.web.utils.response_headers.append_all(response)

zsl.interface.web.utils.response_headers.append_asl(response, app)

zsl.interface.web.utils.response_headers.append_cache(response)

zsl.interface.web.utils.response_headers.append_crossdomain(response: Response, config: Config) → None

Adds the default crossdomain headers. Uses the zsl.task.task_decorator.CrossdomainWebTaskResponder to generate the headers.

Parameters:

• response – Current web response.

• config – Current configuration.

zsl.interface.web.utils.response_headers.append_headers(f)

Appends all the web headers:

• ZSL version and information,

• default CORS if not already set up,

• cache.

Parameters:

f – The decorated function.

Returns:

The function which appends the web headers.

zsl.resource.guard

Guard module defines tools to inject security checks into a resource. With help of the guard class decorator and ResourcePolicy declarative policy class a complex security resource behaviour can be achieved.

class zsl.resource.guard.Access(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

ALLOW = 1

CONTINUE = 3

DENY = 2

class zsl.resource.guard.GuardedMixin

Add guarded CRUD methods to resource.

The guard replaces the CRUD guarded methods with a wrapper with security checks around these methods. It adds this mixin into the resource automatically, but it can be declared on the resource manually for IDEs to accept calls to the guarded methods.

guarded_create(params: str, args: Dict[str, str], data: Dict[str, Any]) → Dict[str, Any]

guarded_delete(params: str, args: Dict[str, str], data: Dict[str, Any]) → Dict[str, Any]

guarded_read(params: str, args: Dict[str, str], data: Dict[str, Any]) → Dict[str, Any]

guarded_update(params: str, args: Dict[str, str], data: Dict[str, Any]) → Dict[str, Any]

exception zsl.resource.guard.PolicyViolationError(message, code=403)

Error raised when policy is violated.

It can bear a HTTP status code, 403 is by default.

class zsl.resource.guard.ResourcePolicy

Declarative policy class.

Every CRUD method has is corespondent can_method__before and can_method__after where method can be one of (create, read, update, delete). __before method will get the CRUD method parameters and __after will get the CRUD method result as parameter. On returning Access.ALLOW access is granted. It should return Access.CONTINUE when the policy is not met, but is not broken, i. e. it is not its responsibility to decide. On returning Access.DENY or raising a PolicyException policy is broken and access is immediately denied.

The default implementation of these method lookup for corresponding attribute can_method, so can_read = Access.ALLOW will allow access for reading without the declaration of can_read__before or can_read__after. default attribute is used if can_method attribute is not declared. For more complex logic it can be declared as a property, see examples:

class SimplePolicy(ResourcePolicy):
    '''Allow read and create'''

    default = Access.ALLOW
    can_delete = Access.CONTINUE
    can_update = Access.CONTINUE


class AdminPolicy(ResourcePolicy):
    '''Only admin has access'''

    @inject(user_service=UserService)
    def __init__(self, user_service):
        self._user_service = user_service

    @property
    def default(self):
        if self._user_service.current_user.is_admin:
            return Access.ALLOW

can_create__after(*args, **kwargs)

Check create method after executing.

can_create__before(*args, **kwargs)

Check create method before executing.

can_delete__after(*args, **kwargs)

Check delete method after executing.

can_delete__before(*args, **kwargs)

Check delete method before executing.

can_read__after(*args, **kwargs)

Check read method after executing.

can_read__before(*args, **kwargs)

Check read method before executing.

can_update__after(*args, **kwargs)

Check update method after executing.

can_update__before(*args, **kwargs)

Check update method before executing.

default = 3

zsl.resource.guard.default_error_handler(e: PolicyViolationError, *_: Any) → ResourceResult

Default policy violation error handler.

It will create an empty resource result with an error HTTP code.

class zsl.resource.guard.guard

Guard decorator.

This decorator wraps the CRUD methods with security checks before and after CRUD method execution, so that the response can be stopped or manipulated. The original CRUD methods are renamed to guarded_method, where method can be [create, read, update, delete], so by using a GuardedResource as a base, you can still redeclare the guarded_methods and won’t loose the security checks.

It takes a list of policies, which will be always checked before and after executing the CRUD method.

Policy is met, when it returns Access.ALLOW, on Access.CONTINUE it will continue to check others and on Access.DENY or raising a PolicyViolationError access will be restricted. If there is no policy which grants the access a PolicyViolationError is raised and access will be restricted.

Guard can have a custom exception handlers or method wrappers to _wrap the CRUD method around.

class Policy(ResourcePolicy):
    default = Access.DENY
    can_read = Access.ALLOW  # allow only read


@guard([Policy()])
class GuardedResource(GuardedMixin):
    def read(self, param, args, data):
        return resources[param]


class SpecificResource(GuardedResource):
    # override GuardedResource.read, but with its security checks
    def guarded_read(self, param, args, data):
        return specific_resources[param]

exception_handlers = [<function default_error_handler>]

method_wrappers = []

resource_methods = ['create', 'read', 'update', 'delete']

zsl.resource.guard.transactional_error_handler(e: Any, rv: Any, _: SessionFactory) → Any

Re-raise a violation error to be handled in the _nested_transactional.

class zsl.resource.guard.transactional_guard

Security guard for ModelResource.

This add a transactional method wrapper and error handler which calls the rollback on PolicyViolationError.

exception_handlers = [<function default_error_handler>, <function transactional_error_handler>]

method_wrappers = [<function transactional>, <function _nested_transactional>]

zsl.resource.json_server_resource

class zsl.resource.json_server_resource.JsonServerResource(model_cls=None)

Model resource implementation to correspond with json-server.

This implements the same REST interface which json-server (https://github.com/typicode/json-server) uses. It transforms the given input arguments into ModelResource-like and then adds metadata to result.

create(*args, **kwargs)

Adds created http status response and location link.

delete(params, args, data)

Supports only singular delete and adds proper http status.

read(params, args, data)

Modifies the parameters and adds metadata for read results.

to_filter(query, arg)

Json-server filter using the _or_ operator.

update(*args, **kwargs)

Modifies the parameters and adds metadata for update results.

Currently it does not support PUT method, which works as replacing the resource. This is somehow questionable in relation DB.

zsl.resource.model_resource – REST for a DB model.

Resources provide a way how to directly access DB tables (raw models) and perform CRUD operations upon them. The default classes of the models should be overridden to provide more logic or restrictions if wanted.

The basic way to use them is as follows:

• The models can be defined in the __exposer__.py which is set up at settings.

• To override just override the basic methods - create, read, update, delete.

class zsl.resource.model_resource.CachedModelResource(model_cls, cache_module, id_helper, logger, timeout='short')

The cached resource - uses redis to cache the resource for the given amount of seconds.

create(params, args, data)

POST /resource/model_cls/ data

Create new resource

delete(params, args, data)

DELETE /resource/model_cls/[params]?[args]

delete resource/s

invalidate()

Invalidates all the data associated with this model

update(params, args, data)

PUT /resource/model_cls/[params:id] data

Update resource/s

class zsl.resource.model_resource.ModelResource(model_cls=None)

static _create_context(params: dict, args: list, data: dict) → ResourceQueryContext

Creates the resource query context - this an object holding the data alongside the querying of the resource. This object is always present as a parameter for each method during the query and users.py are free to create own properties so that they can optimize and perform the query (so the subsequent methods have an access to the already precomputed data).

_create_one(ctx)

Creates an instance to be saved when a model is created.

_save_one(model, ctx)

Saves the created instance.

_delete_one(row_id, ctx)

Deletes row by the given id – row_id. The method first created the query using the method _create_delete_one_query() and then executes it.

Parameters:

• row_id (int) – Identifier of the deleted row.

• ctx (ResourceQueryContext) – The context of this delete query.

_create_delete_one_query(row_id, ctx)

Delete row by id query creation.

Parameters:

• row_id (int) – Identifier of the deleted row.

• ctx (ResourceQueryContext) – The context of this delete query.

create(params: str, args: dict, data: dict) → AppModel

POST /resource/model_cls/ data

Create new resource

delete(params: str, args: dict, data: dict) → None

DELETE /resource/model_cls/[params]?[args]

delete resource/s

read(params: str = None, args: dict = None, data: dict = None) → List[AppModel] | AppModel

GET /resource/model_cls/[params:id]?[args:{limit,offset,page,per_page,filter_by,order_by,related,fields}]

Get resource/s

:param params :type params list :param args :type args dict :param data :type data: dict

update(params: str, args: dict, data: dict) → List[AppModel] | AppModel

PUT /resource/model_cls/[params:id] data

Update resource/s

class zsl.resource.model_resource.ModelResourceBase(model_cls=None)

ModelResource works only for tables with a single-column identifier (key).

__init__(model_cls=None)

Create Model CRUD resource for model_cls

add_related(query, related)

set_ordering(query, arg)

to_filter(query, arg)

class zsl.resource.model_resource.ReadOnlyResourceMixin

The mixin to be used to forbid the update/delete and create operations. Remember the Python’s MRO and place this mixin at the right place in the inheritance declaration.

static create(params, args, data)

Raises exception.

Just raises ReadOnlyResourceUpdateOperationException to indicate that this method is not available.

Raises:

ReadOnlyResourceUpdateOperationException – when accessed

static update(params, args, data)

Raises exception.

Just raises ReadOnlyResourceUpdateOperationException to indicate that this method is not available.

Raises:

ReadOnlyResourceUpdateOperationException – when accessed

static delete(params, args, data)

Raises exception.

Just raises ReadOnlyResourceUpdateOperationException to indicate that this method is not available.

Raises:

ReadOnlyResourceUpdateOperationException – when accessed

OPERATION_CREATE = 'create'

OPERATION_DELETE = 'delete'

OPERATION_UPDATE = 'update'

static create(params, args, data)

Raises exception.

Just raises ReadOnlyResourceUpdateOperationException to indicate that this method is not available.

Raises:

ReadOnlyResourceUpdateOperationException – when accessed

static delete(params, args, data)

Raises exception.

Just raises ReadOnlyResourceUpdateOperationException to indicate that this method is not available.

Raises:

ReadOnlyResourceUpdateOperationException – when accessed

static update(params, args, data)

Raises exception.

Just raises ReadOnlyResourceUpdateOperationException to indicate that this method is not available.

Raises:

ReadOnlyResourceUpdateOperationException – when accessed

exception zsl.resource.model_resource.ReadOnlyResourceUpdateOperationException(operation)

get_operation()

property operation

class zsl.resource.model_resource.ResourceQueryContext(params: dict, args: list, data: dict)

The context of the resource query.

• holds the parameters and arguments of the query,

• holds the related models which should be fetched (parsed from the arguments),

• holds the given filter and splits it to the given field array (parsed from the arguments)

__init__(params: dict, args: list, data: dict)

property args

Args are in the query part of the url ?related=&filter_by etc.

property data

Body of the request.

get_filter_by()

Filter argument - list of filters.

get_related()

Related argument - parsed as array, original argument containing the list of comma delimited models which should be fetched along with the resource.

get_row_id()

First parameter, if given, else None. Handy for GET requests.

property params

Params are given as the part of the path in URL. For example GET /entities/1 will have. 1 in the params.

zsl.resource.model_resource.dict_pick(dictionary, allowed_keys)

Return a dictionary only with keys found in allowed_keys

zsl.resource.model_resource.page_to_offset(params)

Transforms page/per_page from params to limit/offset suitable for SQL.

Parameters:

params (dict) – The dictionary containing page and per_page values will be added the values limit and offset.

zsl.resource.resource_helper

zsl.resource.resource_helper.apply_related(model_cls, query, related_fields)

zsl.resource.resource_helper.create_related_tree(fields)

zsl.resource.resource_helper.filter_from_url_arg(model_cls, query, arg, query_operator=<function and_>, arg_types=None)

Parse filter URL argument arg and apply to query

Example: ‘column1<=value,column2==value’ -> query.filter(Model.column1 <= value, Model.column2 == value)

zsl.resource.resource_helper.flat_model(tree)

Flatten the tree into a list of properties adding parents as prefixes.

zsl.resource.resource_helper.model_tree(name, model_cls, visited=None)

Create a simple tree of model’s properties and its related models.

It traverse trough relations, but ignore any loops.

Parameters:

• name (str) – name of the model

• model_cls – model class

• visited (list or None) – set of visited models

Returns:

a dictionary where values are lists of string or other dictionaries

zsl.resource.resource_helper.order_from_url_arg(model_cls, query, arg)

zsl.resource.resource_helper.related_from_fields(fields)

zsl.router.method

class zsl.router.method.Performer(f)

set_responder(responder)

zsl.router.method.default_web_responder(rv)

zsl.router.method.get_method_packages(config)

zsl.router.method.identity_responder(rv)

zsl.router.method.route(path, app=None, config=None, **options)

zsl.router.method.set_default_responder(responder)

zsl.router.task

class zsl.router.task.PackageTaskRouterStrategy

can_route(path)

is_task_reloading()

route(path: str) → Callable

class zsl.router.task.PathTaskRouterStrategy(task_configuration: TaskConfiguration)

can_route(path)

route(path)

class zsl.router.task.RouterStrategy

can_route(path: str) → bool

route(path: str) → Callable

exception zsl.router.task.RoutingError(path)

property path

class zsl.router.task.TaskConfiguration

create_namespace(namespace: str) → TaskNamespace

property namespaces

class zsl.router.task.TaskNamespace(namespace: str, task_configuration: TaskConfiguration)

add_packages(packages: List[str]) → TaskNamespace

Adds an automatic resolution of urls into tasks. :param packages: The url will determine package/module and the class. :return: self

add_routes(routes: Dict[str, Callable]) → TaskNamespace

Adds the detailed mapping of urls to tasks.

Parameters:

routes – Mapping which defines how urls of the namespace are mapped to tasks. Each url (string) is mapped to a callable which creates the task instance.

Returns:

self

get_configuration() → TaskConfiguration

get_packages() → List[str]

get_routes() → Dict[str, Callable]

property namespace: str

class zsl.router.task.TaskRouter(config: Config, task_configuration: TaskConfiguration)

route(path: str) → Tuple[Any, Callable]

Returns the task handling the given request path.

zsl.service.service

class zsl.service.service.Service(app=None, engine=None)

Main service class.

class zsl.service.service.TransactionalSupportMixin

This mixin allows the objects to access transactional holder.

zsl.service.service.transactional(f: Callable) → Callable

Decorator for making a function transactional.

Parameters:

f – Function to decorate.

Returns:

Decorated function.

from zsl.service import transactional

class MyService(Service):
    @transactional
    def my_function(self):
        self._repository.create(self._orm, 2, 3, 4)

zsl.service.service.tx_session(service: TransactionalSupportMixin) → Generator[Session, None, None]

Context manager for transactional session.

Parameters:

service – Service instance.

Returns:

Session.

from zsl.service import tx_session

class MyService(Service):
    def my_method(self):
        with tx_session(self) as session:
            self._repository.create(session, 2, 3, 4)

zsl.task.job_context

class zsl.task.job_context.DelegatingJobContext(job: Job, task: object, task_callable: Callable)

stop_delegating()

class zsl.task.job_context.HeadersDict

Location: str

class zsl.task.job_context.Job(data: JobData)

property is_valid: bool

Validity of job’s data.

Getter:

Returns if job’s data are valid

Type:

bool

property path: str

Job’s path.

Getter:

Returns job’s path

Type:

str

property payload: dict[str, Any]

Data part of job.

Getter:

Returns job’s payload

Type:

dict

class zsl.task.job_context.JobContext(job: Job, task: object, task_callable: Callable)

Job Context

classmethod get_current_context() → JobContext

property job: Job

property task: object

property task_callable: Callable

property task_data: TaskData

class zsl.task.job_context.JobData

A dictionary that represents the data associated with a job.

Variables:

• path (str) – The path to the job data.

• data (dict[str, Any]) – The data associated with the job.

Example:

>>> job_data = JobData(path='/path/to/job', data={'name': 'John', 'age': 30})

data: dict[str, Any]

path: str

class zsl.task.job_context.Responder

abstract respond(r: Response) → None

class zsl.task.job_context.ResponseDict

headers: HeadersDict

status_code: int

class zsl.task.job_context.StatusCodeResponder(status_code: int)

respond(r: Response) → None

class zsl.task.job_context.WebJobContext(path: str, data: dict, task: object, task_callable: Callable, request: Request)

add_responder(r: Responder)

get_web_request() → Request

notify_responders(response: Response)

zsl.task.job_context.add_responder(responder: Responder) → None

zsl.task.job_context.create_job(path: str, data: dict[str, Any]) → Job

zsl.task.job_context.delegating_job_context(job: Job, task: object, task_callable: Callable) → Generator[DelegatingJobContext, None, None]

zsl.task.job_context.web_task_redirect(location) → HeadersDict

zsl.task.task_data

class zsl.task.task_data.TaskData(payload, app=None, payload_type=<class 'str'>)

get_data()

property payload

transform_payload(f)

zsl.task.task_decorator

class zsl.task.task_decorator.CrossdomainWebTaskResponder(origin: str = None, methods: List[str] = None, allow_headers: str = None, expose_headers: str = None, max_age: int | timedelta = None, app: ServiceApplication = None, cors_config: CORSConfiguration = None)

source: https://github.com/fengsp/flask-snippets/blob/master/decorators/http_access_control.py

get_methods()

respond(response)

exception zsl.task.task_decorator.ForbiddenError

class zsl.task.task_decorator.MimeSetterWebTaskResponder(mime)

respond(r)

exception zsl.task.task_decorator.SecurityException(hashed_token)

get_hashed_token()

class zsl.task.task_decorator.WebTaskResponder(data)

respond(response)

zsl.task.task_decorator.append_get_parameters(accept_only_web: bool = True) → Callable

Task decorator which appends the GET data to the task data.

Parameters:

accept_only_web – Parameter which limits using this task only with web requests.

zsl.task.task_decorator.crossdomain(origin=None, methods=None, allow_headers=None, expose_headers=None, max_age=None)

zsl.task.task_decorator.error_and_result(f)

Format task result into json dictionary {‘data’: task return value} if no exception was raised during the task execution. If there was raised an exception during task execution, formats task result into dictionary {‘error’: exception message with traceback}.

zsl.task.task_decorator.error_and_result_decorator_inner_fn(f, web_only, *args, **kwargs)

zsl.task.task_decorator.file_upload(f)

Return list of werkzeug.datastructures.FileStorage objects - files to be uploaded

zsl.task.task_decorator.forbid_web_access(f)

Forbids running task using http request.

Parameters:

f – Callable

:return Callable

zsl.task.task_decorator.jsend_output(fail_exception_classes=None)

Format task result to json output in jsend specification format. See: https://github.com/omniti-labs. Task return value must be dict or None.

@param fail_exception_classes: exceptions which will produce ‘fail’ response status.

zsl.task.task_decorator.json_input(f)

Expects task input data in json format and parse this data.

zsl.task.task_decorator.json_output(f)

Format response to json and in case of web-request set response content type to ‘application/json’.

zsl.task.task_decorator.jsonp_wrap(callback_key='callback')

Format response to jsonp and add a callback to JSON data - a jsonp request

zsl.task.task_decorator.log_output(f)

Logs the output value.

zsl.task.task_decorator.required_data(*data)

Task decorator which checks if the given variables (indices) are stored inside the task data.

zsl.task.task_decorator.save_to_file(destination_filename, append=False)

Save the output value to file.

zsl.task.task_decorator.secured_task(f)

Secured task decorator.

zsl.task.task_decorator.web_error_and_result(f)

Same as error_and_result decorator, except: If no exception was raised during task execution, ONLY IN CASE OF WEB REQUEST formats task result into json dictionary {‘data’: task return value}

zsl.task.task_decorator.web_task(f)

Checks if the task is called through the web interface. Task return value should be in format {‘data’: …}.

zsl.task.task_decorator.xml_output(f)

Set content-type for response to WEB-REQUEST to ‘text/xml’

zsl.task.task_status

Status codes for REST services

zsl.tasks.asl.db_test_task

Created on 24.12.2012

class zsl.tasks.zsl.db_test_task.DbTestTask(db, app)

Connects to a database and executes a simple query. The result of the query should be 6.

<emph>Input</emph> No input.

<emph>Output</emph> Returns just a number 6.

@author: Martin Babka

perform(data)

zsl.tasks.asl.kill_worker_task

Created on 22.12.2012

class zsl.tasks.zsl.kill_worker_task.KillWorkerTask(app)

static perform(data)

zsl.tasks.asl.schedule_gearman_task

class zsl.tasks.zsl.schedule_celery_task.ScheduleCeleryTask

perform(data: TaskData) → str

zsl.tasks.zsl.schedule_celery_task.schedule_celery_task(path: str, data: TaskData) → None

zsl.tasks.asl.sum_task

Created on 22.12.2012

..moduleauthor:: Martin Babka

class zsl.tasks.zsl.sum_task.SumTask(app)

perform(data: TaskData) → str

zsl.tasks.asl.test_task

Created on 22.12.2012

..moduleauthor:: Martin Babka <babka@atteq.com>

class zsl.tasks.zsl.test_task.TestTask

perform(_data)

zsl.tasks.asl.version_task

Created on 24.12.2012

..moduleauthor:: Martin Babka

class zsl.tasks.zsl.version_task.VersionTask(app)

Shows the versions of ASL and the various used libraries.

perform(data)

zsl.tasks.asl.version_task

Created on 18.02.2019

..moduleauthor:: Julius Flimmel

class zsl.tasks.zsl.with_request_task.WithRequestTask(app)

class Request

perform(request)

zsl.testing

zsl.testing.db

This module allows for database unit testing. For how to use the database testing in practice, a sample, refer to Testing with database.

The module works in the following way (methods setUp, tearDown): 1. Each test runs in a single transaction. 2. This transaction is always called a rollback.

All the tests are run in a single parent transaction (setUpClass, tearDownClass): 1. In general initialization phase the session/transaction is created and it is kept during all the testing. Also the database schema is created. 2. After this the transaction is called rollback.

This means that the tests may be conducted in the in the memory database or a persistent one which is kept clean.

The module provides class TestSessionFactory - it always returns the same session. Also one should add DbTestModule to the test container when creating Zsl instance, see Testing with a Zsl instance.

class zsl.testing.db.DbTestCase

DbTestCase is a mixin to be used when testing with a database.

setUp(session_factory: TestSessionFactory = None) → None

tearDown(session_factory: TestSessionFactory = None) → None

class zsl.testing.db.DbTestModule

Module fixing the zsl.service.service.SessionFactory to our TestSessionFactory.

get_session_factory() → SessionFactory

get_test_session_factory(session_factory: SessionFactory) → SessionFactory

provide_transaction_holder_factory()

class zsl.testing.db.TestSessionFactory(session_holder: SessionHolder)

Factory always returning the single test transaction.

close_test_session()

create_session()

create_test_session(engine: Engine) → Session

class zsl.testing.db.TestTransactionHolder

begin()

close()

commit()

rollback()

class zsl.testing.db.TestTransactionHolderFactory

create_transaction_holder()

zsl.testing.http

This module allows for querying the API via HTTP.

class zsl.testing.http.HTTPTestCase

Extends TestCase with methods for easier testing of HTTP requests.

assertHTTPStatus(status: int | HTTPStatus, test_value: int, msg: AnyStr) → None

Assert HTTP status

Parameters:

• status – http status

• test_value – flask respond status

• msg – test message

assertJSONData(rv: Response, data: Any, msg: AnyStr) → None

extractResponseJSON(rv: Response) → Dict

getHTTPClient(app: ServiceApplication) → FlaskClient

getRequestContext(app)

requestTask(client: FlaskClient, task: str, data: dict, headers: dict = None) → Response

Request a task using POST and convert the given data to JSON.

Parameters:

• client – The client to ZSL which will be used for the request.

• task – Url which will be requested using POST method.

• data – Data which will be posted and first converted to JSON.

• headers – Dictionary of headers that’ll be appended the Content-Type: application/json header.

Returns:

Flask response.

zsl.testing.http.json_loads(str_: AnyStr) → Dict[str, str]

Parse JSON from Flask response which could be in bytes in Py3.

zsl.testing.test_utils

Test utilities.

class zsl.testing.test_utils.TestTaskData(payload: Any)

Data suitable when directly calling a task.

zsl.testing.test_utils.parent_module(module_name: str) → str

Return the parent module name for a module.

Parameters:

module_name (str) – module nam

Returns:

module’s parent name

Return type:

str

>>> parent_module('zsl.application.module')
'zsl.application'

zsl.testing.zsl

This module allows for unit testing with a Zsl instance. Use ZslTestCase as a mixin to create a predefined Zsl instance to be used while testing. Refer to unit testing section Testing with a Zsl instance for an example.

class zsl.testing.zsl.ZslTestCase

ZSL_TEST_CONFIGURATION: ZslTestConfiguration = None

classmethod setUpClass()

classmethod tearDownClass()

class zsl.testing.zsl.ZslTestConfiguration(app_name, version, container, profile, config_object)

app_name

Alias for field number 0

config_object

Alias for field number 4

container

Alias for field number 2

profile

Alias for field number 3

version

Alias for field number 1

zsl.utils.cache_helper

class zsl.utils.cache_helper.CacheDecorator(id_helper)

decorate(key_params, timeout, fn)

get_data_key(*args)

static get_decoder()

static get_encoder()

abstract get_wrapped_fn()

static is_caching(*args)

class zsl.utils.cache_helper.CacheModelDecorator(id_helper)

get_wrapped_fn()

class zsl.utils.cache_helper.CacheOutputDecorator(id_helper)

get_wrapped_fn()

class zsl.utils.cache_helper.CachePageDecorator(id_helper)

get_wrapped_fn()

zsl.utils.cache_helper.cache_filtered_page(filter_param='filter', pagination_param='pagination', sorter_param='sorter', timeout='default')

zsl.utils.cache_helper.cache_list(key_params, timeout='default')

Cache a page (slice) of a list of AppModels

zsl.utils.cache_helper.cache_model(key_params, timeout='default')

Caching decorator for app models in task.perform

zsl.utils.cache_helper.cache_output(key_params, timeout='default')

zsl.utils.cache_helper.cache_page(key_params, timeout='default')

Cache a page (slice) of a list of AppModels

zsl.utils.cache_helper.create_key_for_data(prefix, data, key_params)

From data params in task create corresponding key with help of key_params (defined in decorator)

zsl.utils.command_dispatcher

class zsl.utils.command_dispatcher.CommandDispatcher

A simple class for command dictionary. A command is a function which can take named parameters.

add_function(fn)

Add method or function to dispatcher. Can be use as a nice decorator.

Parameters:

fn (function) – function or method

Returns:

the same function

Return type:

function

bound(instance)

Return a new dispatcher, which will switch all command functions with bounded methods of given instance matched by name. It will match only regular methods.

Parameters:

instance (object) – object instance

Returns:

new Dispatcher

Return type:

CommandDispatcher

command(fn)

Add method or function to dispatcher. Can be use as a nice decorator.

Parameters:

fn (function) – function or method

Returns:

the same function

Return type:

function

execute_command(command, args=None)

Execute a command

Parameters:

• command (str) – name of the command

• args (dict) – optional named arguments for command

Returns:

the result of command

Raises:

KeyError – if command is not found

zsl.utils.date_helper

zsl.utils.date_helper.format_date_portable(ts)

zsl.utils.date_helper.format_date_relative(d)

zsl.utils.date_helper.format_datetime_portable(ts)

zsl.utils.date_helper.format_datetime_relative(dt)

zsl.utils.deploy.apiary_doc_generator

Generates the API documentation suitable for the apiary.io. It parses all the files and finds the apiary.io definitions in the documentary comments. Then outputs it to a file.

class zsl.utils.deploy.apiari_doc_generator.ApiaryDoc

docmodule(obj, name=None, *args)

Raise an exception for unimplemented types.

get_doc()

zsl.utils.deploy.apiari_doc_generator.generate_apiary_doc(task_router)

Generate apiary documentation.

Create a Apiary generator and add application packages to it.

Parameters:

task_router (TaskRouter) – task router, injected

Returns:

apiary generator

Return type:

ApiaryDoc

zsl.utils.deploy.integrator

zsl.utils.deploy.integrator.integrate_to_file(what, filename, start_line, end_line)

WARNING this is working every second run.. so serious bug Integrate content into a file withing “line marks”

zsl.utils.deploy.js_model_generator

class zsl.utils.deploy.js_model_generator.ModelGenerator(module, model_prefix='', collection_prefix='', model_fn='Atteq.bb.Model', collection_fn='Atteq.bb.Collection')

generate_model(model_name, model_plural=None)

generate_models(models)

zsl.utils.deploy.js_model_generator.generate_js_models(module: str, models: str, collection_prefix: str, model_prefix: str, model_fn: str, collection_fn: str, marker: str, integrate: bool, js_file: str) → str | None

Generate models for Backbone Javascript applications.

Parameters:

• module – module from which models are imported

• models – model name, can be a tuple WineCountry/WineCountries as singular/plural

• model_prefix – namespace prefix for models (app.models.)

• collection_prefix – namespace prefix for collection (App.collections.)

• model_fn – name of model constructor (MyApp.bb.Model)

• collection_fn – name of collection constructor (MyApp.bb.Collection)

• marker – marker to indicate the auto generated code

• integrate – integrate to file

• js_file – file to integrate

Returns:

generated models or nothing if writing into a file

zsl.utils.deploy.js_model_generator.parse_model_arg(models: list[str]) → list[str]

Parse the model argument definition.

zsl.utils.dict_to_object_conversion.extend_object_by_dict(target, dict_data, hints=None)

Extends the target object with data from the provided dictionary, using hints to handle specific data types and related objects.

Parameters:

• target (object) – The object to extend with data from the dictionary.

• dict_data (dict) – The dictionary containing data to extend the target object with.

• hints (dict, optional) –

  A dictionary containing hints on how to handle specific fields. The hints should be organized in the following way:

  {
      'date_data': {
          'field_name': 'date_format',
          ...
      },
      'datetime_data': {
          'field_name': 'datetime_format',
          ...
      },
      'related_fields': {
          'field_name': {
              'cls': RelatedClass,
              'hints': {
                  'date_data': {...},
                  'datetime_data': {...},
                  'related_fields': {...}
              }
          },
          ...
      }
  }
  

  The ‘date_data’ and ‘datetime_data’ keys contain dictionaries with field names as keys and date or datetime formats as values. This helps to correctly parse date and datetime strings in the provided dictionary. The ‘related_fields’ key contains a dictionary with field names as keys and dictionaries as values. Each dictionary should have a ‘cls’ key, which should be a related class, and an optional ‘hints’ key containing another hints dictionary, used for nested related objects.

Returns:

None. The function modifies the target object in-place.

zsl.utils.date_helper

zsl.utils.documentation.documentation_link(chapter: str) → str

Creates a link to the documentation.

This method is useful for showing a link to the ZSL documentation in case of any misconfiguration, etc. :param chapter: Chapter name in to which the link points. Use underscores instead of spaces. :return: The absolute link to the documentation.

zsl.utils.email_helper

zsl.utils.email_helper.send_email(sender, receivers, subject, text=None, html=None, charset='utf-8', config=None)

Sends an email.

Parameters:

• sender – Sender as string or None for default got from config.

• receivers – String or array of recipients.

• subject – Subject.

• text – Plain text message.

• html – Html message.

• charset – Charset.

• config – Current configuration

zsl.utils.file_helper

zsl.utils.file_helper.makedirs(path)

Behaves like mkdir -p <path>. Without failure if the path exists.

zsl.utils.http.get_http_status_code_value(status_code: int | HTTPStatus) → int

Py2/3 status code.

zsl.utils.import_helper

zsl.utils.import_helper.fetch_class(full_class_name)

Fetches the given class.

Parameters:

full_class_name (string) – Name of the class to be fetched.

zsl.utils.injection_helper

zsl.utils.injection_helper.bind(interface, to=None, scope=None)

zsl.utils.injection_helper.inject(**bindings)

Decorator for injecting parameters for ASL objects.

zsl.utils.injection_helper.instantiate(cls)

zsl.utils.injection_helper.simple_bind(binder: Binder, cls: Type, scope: Scope) → None

zsl.utils.model_helper

Helper module for working with models.

zsl.utils.model_helper.update_model(raw_model, app_model, forbidden_keys=None, inverse=False)

Updates the raw_model according to the values in the app_model.

Parameters:

• raw_model – Raw model which gets updated.

• app_model – App model holding the data.

• forbidden_keys (list) – Data/attributes which will not be updated.

• inverse – If the value is True all app_model attributes which are contained in the raw_model are updated. If the value is False all raw_model properties which are in the app_model will be updated.

zsl.utils.nginx_push_helper

Helper for nginx push stream module https://github.com/wandenberg/nginx-push-stream-module

class zsl.utils.nginx_push_helper.NginxPusher(server_path, channel_prefix=None)

channel_path(channel_id)

delete_channel(channel_id)

Deletes channel

push(channel_id, data)

Push message with POST data for channel_id

push_msg(channel_id, msg)

Push msg for given channel_id. If msg is not string, it will be urlencoded

push_object(channel_id, obj)

Push obj for channel_id. obj will be encoded as JSON in the request.

zsl.utils.params_helper

Helpers for parameter handling

exception zsl.utils.params_helper.RequestException

Exception raised on bad request

zsl.utils.params_helper.msg_err_missing_params(*params)

zsl.utils.params_helper.required_params(data, *r_params)

Check if given parameters are in the given dict, if not raise an exception.

Parameters:

• data (dict) – data to check

• r_params – required parameters

Raises:

RequestException – if params not in data

zsl.utils.params_helper.safe_args(fn, args)

Check if args as a dictionary has the required parameters of fn function and filter any waste parameters so fn can be safely called with them.

Parameters:

• fn (Callable) – function object

• args (dict) – dictionary of parameters

Returns:

dictionary to be used as named params for the fn

Return type:

dict

zsl.utils.php_helper

Module with functions to help with dealing with code written in PHP.

zsl.utils.php_helper.bool_to_str(boolean)

Convert boolean to string like PHP.

Parameters:

boolean (bool) – boolean value

Returns:

string representation

Return type:

str

zsl.utils.redis_helper

Helper module for working with redis.

class zsl.utils.redis_helper.Keymaker(keys, prefix=None, config=None)

Keymaker is a class to generate an object to generate Redis keys.

Example:

>>> article_keys = Keymaker(prefix='articles', keys={'full_article': 'full', 'short_article': 'short'})
>>> article_keys.full_article('today', 'ID214')
'$PROJECT_PREFIX:articles:full:today:ID214'

zsl.utils.redis_helper.redis_key(*args)

Create a string key from a hierarchical structure.

Parameters:

args (list(str)) – list of keys to denote hierarchy

Returns:

string representation of args

Return type:

str

Example:

>>> redis_key(['project', 'module', 'class'])
'project:module:class'

zsl.utils.reflection_helper

Helper module for OOP operations.

zsl.utils.reflection_helper.add_mixin(base_class, mixin_class)

zsl.utils.reflection_helper.extend(instance, new_class)

Adds new_class to the ancestors of instance.

Parameters:

• instance – Instance that will have a new ancestor.

• new_class – Ancestor.

zsl.utils.reflection_helper.is_list(v)

zsl.utils.reflection_helper.is_scalar(v)

zsl.utils.reflection_helper.proxy_object_to_delegate(proxy_object, delegate_object)

zsl.utils.request_helper

Helper module for working with request.

zsl.utils.request_helper.args_to_dict(args)

Converts request arguments to a simple dictionary. If an argument appears multiple times, the resulting dictionary will contain list of all of its values.

Parameters:

args (request.args) – flask’s request args object

Returns:

a simple dict from args

Return type:

dict

zsl.utils.resource_helper

Helper module for resource management.

exception zsl.utils.resource_helper.MethodNotImplementedException

Exception raised on missing method

zsl.utils.resource_helper.create_model_resource(resource_map, name, app=None)

Create a model resource from a dict resource_map {‘resource name’: (‘model package’, ‘model class’)}

Parameters:

• resource_map (dict(str, tuple(str))) – dict with resource descriptions

• name – name of the concrete resource

• app (Zsl) – current application, injected

Returns:

instantiated model resource

zsl.utils.resource_helper.get_method(resource, method)

Test and return a method by name on resource.

Parameters:

• resource (object) – resource object

• method (str) – method name

Returns:

bounded method

Raises:

MethodNotImplementedException – when method not found

zsl.utils.resource_helper.get_resource_task(resource_path: str, config: Config = None) → Callable[[str, Dict, Dict], List[AppModel] | AppModel | ResourceResult]

Search and return a bounded method for given path.

Parameters:

• resource_path (str) – resource path

• config (Config) – current configuration, injected

Returns:

bounded method or None when not found

Raises:

NameError – when can’t find given resource by path

zsl.utils.resource_helper.parse_resource_path(path)

Split the path to its elements.

Parameters:

path (str) – URL path

Returns:

name and rest of the path

Return type:

tuple(str, list(str))

zsl.utils.rss

Helper function for handling rss.

zsl.utils.rss.complex_el_from_dict(parent, data, key)

Create element from a dict definition and add it to parent.

Parameters:

• parent (Element) – parent element

• data – dictionary with elements definitions, it can be a simple {element_name: ‘element_value’} or complex {element_name: {_attr: {name: value, name1: value1}, _text: ‘text’}}

• key – element name and key in data

Returns:

created element

zsl.utils.rss.element_from_dict(parent, data, element)

Create element to parent and sets its value to data[element], which will be removed from the data.

Parameters:

• parent (Element) – parent element

• data (dict(str, str)) – dictionary where data[element] is desired value

• element (str) – name of the new element

Returns:

created element

zsl.utils.rss.rss_create(channel, articles)

Create RSS xml feed.

Parameters:

• channel (dict(str, str)) – channel info [title, link, description, language]

• articles (list(dict(str,str))) – list of articles, an article is a dictionary with some required fields [title, description, link] and any optional, which will result to <dict_key>dict_value</dict_key>

Returns:

root element

Return type:

ElementTree.Element

zsl.utils.security_helper

Helper module with function dealing with security.

zsl.utils.security_helper.compute_token(random_token, config)

Compute a hash of the given token with a preconfigured secret.

Parameters:

random_token (str) – random token

Returns:

hashed token

Return type:

str

zsl.utils.security_helper.generate_security_data()

Generate security token - a random token and its hashed version salted with a secret.

Returns:

random and hashed token

Return type:

dict(str, str)

zsl.utils.security_helper.verify_security_data(security)

Verify an untrusted security token.

Parameters:

security (dict) – security token

Returns:

True if valid

Return type:

bool

zsl.utils.security_helper.wrap_plain_data_as_secure(data)

Wrap task data with security token.

Parameters:

data – data to be wrapped

Returns:

wrapped data with security token

Return type:

dict

zsl.utils.string_helper

zsl.utils.string_helper.addslashes(s, escaped_chars=None)

Add slashes for given characters. Default is for \ and '.

Parameters:

• s – string

• escaped_chars – list of characters to prefix with a slash \

Returns:

string with slashed characters

Return type:

str

Example:

>>> addslashes("'")
"\'"

zsl.utils.string_helper.camelcase_to_underscore(name)

Transform string from camelCase to underscore_string.

Parameters:

name (str) – string in camelcase

Returns:

string with underscores

Return type:

str

Example:

>>> camelcase_to_underscore('camelCase')
'camel_case'

zsl.utils.string_helper.et_node_to_string(et_node, default='')

Simple method to get stripped text from node or default string if None is given.

Parameters:

• et_node (xml.etree.ElementTree.Element, None) – Element or None

• default (str) – string returned if None is given, default ''

Returns:

text from node or default

Return type:

str

zsl.utils.string_helper.generate_random_string(size=6, chars='ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789')

Generate random string.

Parameters:

• size (int) – Length of the returned string. Default is 6.

• chars (str) – List of the usable characters. Default is string.ascii_uppercase + string.digits.

Returns:

The random string.

Return type:

str

zsl.utils.string_helper.join_list()

Concatenates the upper-cased values using the given delimiter if the given values variable is a list. Otherwise it is just returned. :param values: List of strings or string . :param delimiter: The delimiter used to join the values. :return: The concatenation or identity.

zsl.utils.string_helper.strip_html_tags(s)

Remove all html tags from string.

Parameters:

s (str) – string

Returns:

string with stripped html tags

Return type:

str

Example:

>>> strip_html_tags('<h1>hello <i>world</i></h1>')
'hello world'

zsl.utils.string_helper.strip_html_tags_unicode(s)

Remove all html tags from string.

Deprecated since version Use: strip_html_tags() instead.

Parameters:

s (str) – string

Returns:

string with stripped html tags

Return type:

str

zsl.utils.string_helper.underscore_to_camelcase(value, first_upper=True)

Transform string from underscore_string to camelCase.

Parameters:

• value (str) – string with underscores

• first_upper – the result will have its first character in upper case

Returns:

string in CamelCase or camelCase according to the first_upper

Return type:

str

Example:

>>> underscore_to_camelcase('camel_case')
'CamelCase'
>>> underscore_to_camelcase('camel_case', False)
'camelCase'

zsl.utils.string_helper.xstr(s)

If s is None return empty string.

Parameters:

s – string

Returns:

s or an empty string if s is None

Return type:

str

Example:

>>> xstr(None)
''

zsl.utils.string_helper.xunicode(s)

If s is None return empty string

Deprecated since version Use: xstr() instead.

Parameters:

s – string

Returns:

s or an empty string if s in None

Return type:

str

zsl.utils.task_helper

Helper module for task management.

zsl.utils.task_helper.get_callable(task)

Return the perform method on given task.

Parameters:

task – task instance

Returns:

bound perform method

zsl.utils.task_helper.run_task(task_cls, task_data)

Instantiate and run the perform method od given task data.

Parameters:

• task_cls – task class

• task_data (TaskData) – task data

Returns:

task’s result

zsl.utils.task_helper.run_task_json(task_cls, task_data)

Instantiate and run the perform method od given task data.

Parameters:

• task_cls – task class

• task_data (TaskData) – task data

Returns:

task’s result

zsl.utils.testing

zsl.utils.testing.load_and_run_tests(config)

Load application test package and run it using testing util.

Parameters:

config (Config) – application configuration, injected

zsl.utils.testing.set_test_responder()

zsl.utils.type_helper

Helper module for handling types.

zsl.utils.type_helper.not_empty_list(l)

Check if given value is a non-empty list.

Parameters:

l – value

Returns:

True on non-empty list, False otherwise

Return type:

bool

Example:

>>> not_empty_list([1, 2, 3])
True
>>> not_empty_list(None)
False

zsl.utils.url_helper

Helper module for URL handling.

zsl.utils.url_helper.slugify(value, allow_unicode=False)

Normalizes string, converts to lowercase, removes non-alpha characters, and converts spaces to hyphens.

Parameters:

• value – string

• allow_unicode (bool) – allow utf8 characters

Returns:

slugified string

Return type:

str

Example:

>>> slugify('pekná líščička')
'pekna-liscicka'

zsl.utils.url_helper.urlencode(query)

Encode string to be used in urls (percent encoding).

Parameters:

query (str) – string to be encoded

Returns:

urlencoded string

Return type:

str

Example:

>>> urlencode('pekná líščička')
'pekn%C3%A1%20l%C3%AD%C5%A1%C4%8Di%C4%8Dka'

zsl.utils.warnings.deprecated(func)

This is a decorator which can be used to mark functions as deprecated. It will result in a warning being emitted when the function is used.

zsl.utils.xml_helper – xml helpers

Helper functions for working with XML and ElementTree.

exception zsl.utils.xml_helper.NotValidXmlException

Exception raised on invalid xml

zsl.utils.xml_helper.attrib_to_dict(element, *args, **kwargs)

For an ElementTree element extract specified attributes. If an attribute does not exists, its value will be None.

attrib_to_dict(element, ‘attr_a’, ‘attr_b’) -> {‘attr_a’: ‘value’, ‘attr_a’: ‘value’}

Mapping between xml attributes and dictionary keys is done with kwargs. attrib_to_dict(element, my_new_name = ‘xml_atribute_name’, ..)

zsl.utils.xml_helper.create_el(name, text=None, attrib=None)

Create element with given attributes and set element.text property to given text value (if text is not None)

Parameters:

• name (str) – element name

• text (str) – text node value

• attrib (dict) – attributes

Returns:

xml element

Return type:

Element

zsl.utils.xml_helper.element_to_int(element, attribute=None)

Convert element object to int. If attribute is not given, convert element.text.

Parameters:

• element – ElementTree element

• attribute (str) – attribute name

Returns:

integer

Return type:

int

zsl.utils.xml_helper.get_xml_root(xml_path)

Load and parse an xml by given xml_path and return its root.

Parameters:

xml_path (str) – URL to a xml file

Returns:

xml root

zsl.utils.xml_helper.msg_err_missing_attributes(tag, *attributes)

Format message for missing attributes.

Parameters:

• tag – tag name

• attributes – list of attributes

Returns:

message

Return type:

str

zsl.utils.xml_helper.msg_err_missing_children(tag, *children)

Format message for missing children.

Parameters:

• tag – tag name

• children – list of children

Returns:

message

Return type:

str

zsl.utils.xml_helper.required_attributes(element, *attributes)

Check element for required attributes. Raise NotValidXmlException on error.

Parameters:

• element – ElementTree element

• attributes – list of attributes names to check

Raises:

NotValidXmlException – if some argument is missing

zsl.utils.xml_helper.required_elements(element, *children)

Check element (xml.etree.ElementTree.Element) for required children, defined as XPath. Raise NotValidXmlException on error.

Parameters:

• element – ElementTree element

• children – list of XPaths to check

Raises:

NotValidXmlException – if some child is missing

zsl.utils.xml_helper.required_items(element, children, attributes)

Check an xml element to include given attributes and children.

Parameters:

• element – ElementTree element

• children – list of XPaths to check

• attributes – list of attributes names to check

Raises:

• NotValidXmlException – if some argument is missing

• NotValidXmlException – if some child is missing

zsl.utils.xml_to_json – xml helpers

Helper functions for simpler parsing xml into object with schemas.

exception zsl.utils.xml_to_json.NotCompleteXmlException

Exception raised during parsing an invalid XML.

exception zsl.utils.xml_to_json.XmlToJsonException

Exception raised during converting xml to json.

zsl.utils.xml_to_json.xml_to_json(element, definition, required=False)

Convert XML (ElementTree) to dictionary from a definition schema.

Definition schema can be a simple string - XPath or @attribute for direct extraction or a complex one described by

• dictionary {key: 'xpath or @attribute', second: 'complex definition'} required parameters can be marked with * at the end

• list [xpath, [definition]] - create a list of all elements found by xpath, parse the parts with given definition if provided as second argument

• Callable - parse the element by given function, can be handy as a part of complex definition

Parameters:

• element (ElementTree.Element) – ElementTree element

• definition (Union[str, tuple, dict, list, Callable]) – schema for the json

• required (bool) – parsed value should be not None

Returns:

parsed xml

Return type:

Union[dict, str, list]