Plugins toolkit reference

As well as using the variables made available to them by implementing plugin interfaces, plugins will likely want to be able to use parts of the CKAN core library. To allow this, CKAN provides a stable set of classes and functions that plugins can use safe in the knowledge that this interface will remain stable, backward-compatible and with clear deprecation guidelines when new versions of CKAN are released. This interface is available in CKAN’s plugins toolkit.

class ckan.plugins.toolkit.BaseModel

Base class for SQLAlchemy declarative models.

Models extending BaseModel class are attached to the SQLAlchemy’s metadata object automatically:

from ckan.plugins import toolkit

class ExtModel(toolkit.BaseModel):
    __tablename__ = "ext_model"
    id = Column(String(50), primary_key=True)
    ...
class ckan.plugins.toolkit.CkanVersionException

Exception raised by requires_ckan_version() if the required CKAN version is not available.

class ckan.plugins.toolkit.DefaultDatasetForm

The default implementatinon of IDatasetForm.

This class serves two purposes:

  1. It provides a base class for plugin classes that implement IDatasetForm to inherit from, so they can inherit the default behavior and just modify the bits they need to.

  2. It is used as the default fallback plugin when no registered IDatasetForm plugin handles the given dataset type and no other plugin has registered itself as the fallback plugin.

Note

DefaultDatasetForm doesn’t call implements(), because we don’t want it being registered.

class ckan.plugins.toolkit.DefaultGroupForm

Provides a default implementation of the pluggable Group controller behaviour.

This class has 2 purposes:

  • it provides a base class for IGroupForm implementations to use if only a subset of the method hooks need to be customised.

  • it provides the fallback behaviour if no plugin is setup to provide the fallback behaviour.

Note

this isn’t a plugin implementation. This is deliberate, as we don’t want this being registered.

class ckan.plugins.toolkit.DefaultOrganizationForm

Provides a default implementation of the pluggable Group controller behaviour.

This class has 2 purposes:

  • it provides a base class for IGroupForm implementations to use if only a subset of the method hooks need to be customised.

  • it provides the fallback behaviour if no plugin is setup to provide the fallback behaviour.

Note

this isn’t a plugin implementation. This is deliberate, as we don’t want this being registered.

class ckan.plugins.toolkit.HelperError

Raised if an attempt to access an undefined helper is made.

Normally, this would be a subclass of AttributeError, but Jinja2 will catch and ignore them. We want this to be an explicit failure re #2908.

class ckan.plugins.toolkit.Invalid

Exception raised by some validator, converter and dictization functions when the given value is invalid.

class ckan.plugins.toolkit.NotAuthorized

Exception raised when the user is not authorized to call the action.

For example package_create() raises NotAuthorized if the user is not authorized to create packages.

class ckan.plugins.toolkit.ObjectNotFound

Exception raised by logic functions when a given object is not found.

For example package_show() raises ObjectNotFound if no package with the given id exists.

class ckan.plugins.toolkit.StopOnError

error to stop validations for a particualar key

class ckan.plugins.toolkit.UnknownValidator

Exception raised when a requested validator function cannot be found.

class ckan.plugins.toolkit.ValidationError

Exception raised by action functions when validating their given data_dict fails.

ckan.plugins.toolkit._(*args: 'Any', **kwargs: 'Any') 'str'

Translates a string to the current locale.

The _() function is a reference to the ugettext() function. Everywhere in your code where you want strings to be internationalized (made available for translation into different languages), wrap them in the _() function, eg.:

msg = toolkit._("Hello")

Returns the localized unicode string.

ckan.plugins.toolkit.abort(status_code: 'int', detail: 'str' = '', headers: 'Optional[dict[str, Any]]' = None, comment: 'Optional[str]' = None) 'NoReturn'

Abort the current request immediately by returning an HTTP exception.

This is a wrapper for flask.abort() that adds some CKAN custom behavior, including allowing IAuthenticator plugins to alter the abort response, and showing flash messages in the web interface.

ckan.plugins.toolkit.add_public_directory(config_: 'CKANConfig', relative_path: 'str')

Add a path to the extra_public_paths config setting.

The path is relative to the file calling this function.

Webassets addition: append directory to webassets load paths in order to correctly rewrite relative css paths and resolve public urls.

ckan.plugins.toolkit.add_resource(path: 'str', name: 'str')

Add a WebAssets library to CKAN.

WebAssets libraries are directories containing static resource files (e.g. CSS, JavaScript or image files) that can be compiled into WebAsset Bundles.

See Theming guide for more details.

ckan.plugins.toolkit.add_template_directory(config_: 'CKANConfig', relative_path: 'str')

Add a path to the extra_template_paths config setting.

The path is relative to the file calling this function.

ckan.plugins.toolkit.asbool(obj: 'Any') 'bool'

Convert a string (e.g. 1, true, True) into a boolean.

Example:

assert asbool("yes") is True
ckan.plugins.toolkit.asint(obj: 'Any') 'int'

Convert a string into an int.

Example:

assert asint("111") == 111
ckan.plugins.toolkit.aslist(obj: 'Any', sep: 'Optional[str]' = None, strip: 'bool' = True) 'Any'

Convert a space-separated string into a list.

Example:

assert aslist("a b c") == ["a", "b", "c"]
ckan.plugins.toolkit.auth_allow_anonymous_access(action: 'Decorated') 'Decorated'

Flag an auth function as not requiring a logged in user

This means that check_access won’t automatically raise a NotAuthorized exception if an authenticated user is not provided in the context. (The auth function can still return False if for some reason access is not granted).

ckan.plugins.toolkit.auth_disallow_anonymous_access(action: 'Decorated') 'Decorated'

Flag an auth function as requiring a logged in user

This means that check_access will automatically raise a NotAuthorized exception if an authenticated user is not provided in the context, without calling the actual auth function.

ckan.plugins.toolkit.auth_sysadmins_check(action: 'Decorated') 'Decorated'

A decorator that prevents sysadmins from being automatically authorized to call an action function.

Normally sysadmins are allowed to call any action function (for example when they’re using the Action API or the web interface), if the user is a sysadmin the action function’s authorization function will not even be called.

If an action function is decorated with this decorator, then its authorization function will always be called, even if the user is a sysadmin.

ckan.plugins.toolkit.base

The base functionality for web-views.

Provides functions for rendering templates, aborting the request, etc.

ckan.plugins.toolkit.blanket

Quick implementations of simple plugin interfaces.

Blankets allow to reduce boilerplate code in plugins by simplifying the way common interfaces are registered.

For instance, this is how template helpers are generally added using the ITemplateHelpers interface:

from ckan import plugins as p
from ckanext.myext import helpers


class MyPlugin(p.SingletonPlugin):

    p.implements(ITemplateHelpers)

    def get_helpers(self):

        return {
            'my_ext_custom_helper_1': helpers.my_ext_custom_helper_1,
            'my_ext_custom_helper_2': helpers.my_ext_custom_helper_2,
        }

The same pattern is used for IActions, IAuthFunctions, etc.

With Blankets, assuming that you have created your module in the expected path with the expected name (see below), you can automate the registration of your helpers using the corresponding blanket decorator from the plugins toolkit:

@p.toolkit.blanket.helpers
class MyPlugin(p.SingletonPlugin):
    pass

The following table lists the available blanket decorators, the interface they implement and the default source where the blanket will automatically look for items to import:

Decorator

Interfaces

Default source

toolkit.blanket.helpers

ITemplateHelpers

ckanext.myext.helpers

toolkit.blanket.auth_functions

IAuthFunctions

ckanext.myext.logic.auth

toolkit.blanket.actions

IActions

ckanext.myext.logic.action

toolkit.blanket.validators

IValidators

ckanext.myext.logic.validators

toolkit.blanket.blueprints

IBlueprint

ckanext.myext.logic.views

toolkit.blanket.cli

IClick

ckanext.myext.cli

toolkit.blanket.config_declarations

IConfigDeclaration

ckanext/myext/config_declaration.{json,yaml,toml}

Note

By default, all local module members, whose __name__/name doesn’t start with an underscore are exported. If the module has __all__ list, only members listed inside this list will be exported.

If your extension uses a different naming convention for your modules, it is still possible to use blankets by passing the relevant module as a parameter to the decorator:

import ckanext.myext.custom_actions as custom_module

@p.toolkit.blanket.actions(custom_module)
class MyPlugin(p.SingletonPlugin):
    pass

Note

The config_declarations blanket is an exception. Instead of a module object it accepts path to the JSON, YAML or TOML file with the config declarations.

You can also pass a function that produces the artifacts required by the interface:

def all_actions():
    return {'ext_action': ext_action}

@p.toolkit.blanket.actions(all_actions)
class MyPlugin(p.SingletonPlugin):
    pass

Or just a dict with the items required by the interface:

all_actions = {'ext_action': ext_action}

@p.toolkit.blanket.actions(all_actions)
class MyPlugin(p.SingletonPlugin):
    pass
ckan.plugins.toolkit.c

The Pylons template context object.

[Deprecated]: Use toolkit.g instead.

This object is used to pass request-specific information to different parts of the code in a thread-safe way (so that variables from different requests being executed at the same time don’t get confused with each other).

Any attributes assigned to c are available throughout the template and application code, and are local to the current request.

ckan.plugins.toolkit.chained_action(func: 'ChainedAction') 'ChainedAction'

Decorator function allowing action function to be chained.

This allows a plugin to modify the behaviour of an existing action function. A Chained action function must be defined as action_function(original_action, context, data_dict) where the first parameter will be set to the action function in the next plugin or in core ckan. The chained action may call the original_action function, optionally passing different values, handling exceptions, returning different values and/or raising different exceptions to the caller.

Usage:

from ckan.plugins.toolkit import chained_action

@chained_action
@side_effect_free
def package_search(original_action, context, data_dict):
    return original_action(context, data_dict)
Parameters:

func (callable) – chained action function

Returns:

chained action function

Return type:

callable

ckan.plugins.toolkit.chained_auth_function(func: 'ChainedAuthFunction') 'ChainedAuthFunction'

Decorator function allowing authentication functions to be chained.

This chain starts with the last chained auth function to be registered and ends with the original auth function (or a non-chained plugin override version). Chained auth functions must accept an extra parameter, specifically the next auth function in the chain, for example:

auth_function(next_auth, context, data_dict).

The chained auth function may call the next_auth function, optionally passing different values, handling exceptions, returning different values and/or raising different exceptions to the caller.

Usage:

from ckan.plugins.toolkit import chained_auth_function

@chained_auth_function
@auth_allow_anonymous_access
def user_show(next_auth, context, data_dict=None):
    return next_auth(context, data_dict)
Parameters:

func (callable) – chained authentication function

Returns:

chained authentication function

Return type:

callable

ckan.plugins.toolkit.chained_helper(func: 'Helper') 'Helper'

Decorator function allowing helper functions to be chained.

This chain starts with the first chained helper to be registered and ends with the original helper (or a non-chained plugin override version). Chained helpers must accept an extra parameter, specifically the next helper in the chain, for example:

helper(next_helper, *args, **kwargs).

The chained helper function may call the next_helper function, optionally passing different values, handling exceptions, returning different values and/or raising different exceptions to the caller.

Usage:

from ckan.plugins.toolkit import chained_helper

@chained_helper
def ckan_version(next_func, **kw):

    return next_func(**kw)
Parameters:

func (callable) – chained helper function

Returns:

chained helper function

Return type:

callable

ckan.plugins.toolkit.check_access(action: 'str', context: 'Context', data_dict: 'Optional[dict[str, Any]]' = None) 'Literal[True]'

Calls the authorization function for the provided action

This is the only function that should be called to determine whether a user (or an anonymous request) is allowed to perform a particular action.

The function accepts a context object, which should contain a ‘user’ key with the name of the user performing the action, and optionally a dictionary with extra data to be passed to the authorization function.

For example:

check_access('package_update', context, data_dict)

If not already there, the function will add an auth_user_obj key to the context object with the actual User object (in case it exists in the database). This check is only performed once per context object.

Raise NotAuthorized if the user is not authorized to call the named action function.

If the user is authorized to call the action, return True.

Parameters:
  • action (string) – the name of the action function, eg. 'package_create'

  • context (dict)

  • data_dict (dict)

Raises:

NotAuthorized if the user is not authorized to call the named action

ckan.plugins.toolkit.check_ckan_version(min_version: 'Optional[str]' = None, max_version: 'Optional[str]' = None)

Return True if the CKAN version is greater than or equal to min_version and less than or equal to max_version, return False otherwise.

If no min_version is given, just check whether the CKAN version is less than or equal to max_version.

If no max_version is given, just check whether the CKAN version is greater than or equal to min_version.

Parameters:
  • min_version (string) – the minimum acceptable CKAN version, eg. '2.1'

  • max_version (string) – the maximum acceptable CKAN version, eg. '2.3'

ckan.plugins.toolkit.ckan

ckan package itself.

ckan.plugins.toolkit.config

The CKAN configuration object.

It stores the configuration values defined in the CKAN configuration file, eg:

title = toolkit.config.get("ckan.site_title")
ckan.plugins.toolkit.current_user
ckan.plugins.toolkit.enqueue_job(fn: 'Callable[..., Any]', args: 'Optional[Union[tuple[Any], list[Any], None]]' = None, kwargs: 'Optional[dict[str, Any]]' = None, title: 'Optional[str]' = None, queue: 'str' = 'default', rq_kwargs: 'Optional[dict[str, Any]]' = None) 'Job'

Enqueue a job to be run in the background.

Parameters:
  • fn (function) – Function to be executed in the background

  • args (list) – List of arguments to be passed to the function. Pass an empty list if there are no arguments (default).

  • kwargs (dict) – Dict of keyword arguments to be passed to the function. Pass an empty dict if there are no keyword arguments (default).

  • title (string) – Optional human-readable title of the job.

  • queue (string) – Name of the queue. If not given then the default queue is used.

  • rq_kwargs (dict) – Dict of keyword arguments that will get passed to the RQ enqueue_call invocation (eg timeout, depends_on, ttl etc).

Returns:

The enqueued job.

Return type:

rq.job.Job

ckan.plugins.toolkit.error_shout(exception: 'Any') 'None'

Report CLI error with a styled message.

ckan.plugins.toolkit.fresh_context(context: 'Context', **kwargs: 'Any') 'Context'

Copy just the minimum fields into a new context for cases in which we reuse the context and we want a clean version with minimum fields

ckan.plugins.toolkit.g

The Flask global object.

This object is used to pass request-specific information to different parts of the code in a thread-safe way (so that variables from different requests being executed at the same time don’t get confused with each other).

Any attributes assigned to g are available throughout the template and application code, and are local to the current request.

It is a bad pattern to pass variables to the templates using the g object. Pass them explicitly from the view functions as extra_vars, eg:

return toolkit.render(
    'myext/package/read.html',
    extra_vars={
        u'some_var': some_value,
        u'some_other_var': some_other_value,
    }
)
ckan.plugins.toolkit.get_action(action: 'str') 'Action'

Return the named ckan.logic.action function.

For example get_action('package_create') will normally return the ckan.logic.action.create.package_create() function.

For documentation of the available action functions, see Action API reference.

You should always use get_action() instead of importing an action function directly, because IActions plugins can override action functions, causing get_action() to return a plugin-provided function instead of the default one.

Usage:

import ckan.plugins.toolkit as toolkit

# Call the package_create action function:
toolkit.get_action('package_create')(context, data_dict)

As the context parameter passed to an action function is commonly:

context = {'model': ckan.model, 'session': ckan.model.Session,
           'user': user}

an action function returned by get_action() will automatically add these parameters to the context if they are not defined. This is especially useful for plugins as they should not really be importing parts of ckan eg ckan.model and as such do not have access to model or model.Session.

If a context of None is passed to the action function then the default context dict will be created.

Note

Many action functions modify the context dict. It can therefore not be reused for multiple calls of the same or different action functions.

Parameters:

action (string) – name of the action function to return, eg. 'package_create'

Returns:

the named action function

Return type:

callable

ckan.plugins.toolkit.get_converter(validator: 'str') 'Union[Validator, ValidatorFactory]'

Return a validator function by name.

Parameters:

validator (string) – the name of the validator function to return, eg. 'package_name_exists'

Raises:

UnknownValidator if the named validator is not found

Returns:

the named validator function

Return type:

types.FunctionType

ckan.plugins.toolkit.get_endpoint() 'Union[tuple[str, str], tuple[None, None]]'

Returns tuple in format: (blueprint, view).

ckan.plugins.toolkit.get_or_bust(data_dict: 'dict[str, Any]', keys: 'Union[str, Iterable[str]]') 'Union[Any, tuple[Any, ...]]'

Return the value(s) from the given data_dict for the given key(s).

Usage:

single_value = get_or_bust(data_dict, 'a_key')
value_1, value_2 = get_or_bust(data_dict, ['key1', 'key2'])
Parameters:
  • data_dict (dictionary) – the dictionary to return the values from

  • keys (either a string or a list) – the key(s) for the value(s) to return

Returns:

a single value from the dict if a single key was given, or a tuple of values if a list of keys was given

Raises:

ckan.logic.ValidationError if one of the given keys is not in the given dictionary

ckan.plugins.toolkit.get_validator(validator: 'str') 'Union[Validator, ValidatorFactory]'

Return a validator function by name.

Parameters:

validator (string) – the name of the validator function to return, eg. 'package_name_exists'

Raises:

UnknownValidator if the named validator is not found

Returns:

the named validator function

Return type:

types.FunctionType

ckan.plugins.toolkit.h

Collection of CKAN native and extension-provided helpers.

class ckan.plugins.toolkit.literal

Represents an HTML literal.

ckan.plugins.toolkit.login_user(user, remember=False, duration=None, force=False, fresh=True)

Logs a user in. You should pass the actual user object to this. If the user’s is_active property is False, they will not be logged in unless force is True.

This will return True if the log in attempt succeeds, and False if it fails (i.e. because the user is inactive).

Parameters:
  • user (object) – The user object to log in.

  • remember (bool) – Whether to remember the user after their session expires. Defaults to False.

  • duration (datetime.timedelta) – The amount of time before the remember cookie expires. If None the value set in the settings is used. Defaults to None.

  • force (bool) – If the user is inactive, setting this to True will log them in regardless. Defaults to False.

  • fresh (bool) – setting this to False will log in the user with a session marked as not “fresh”. Defaults to True.

ckan.plugins.toolkit.logout_user()

Logs a user out. (You do not need to pass the actual user.) This will also clean up the remember me cookie if it exists.

ckan.plugins.toolkit.mail_recipient(recipient_name: 'str', recipient_email: 'str', subject: 'str', body: 'str', body_html: 'Optional[str]' = None, headers: 'Optional[dict[str, Any]]' = None, attachments: 'Optional[Iterable[Attachment]]' = None) 'None'

Sends an email to a an email address.

Note

You need to set up the Email settings to able to send emails.

Parameters:
  • recipient_name – the name of the recipient

  • recipient_email – the email address of the recipient

  • subject (string) – the email subject

  • body (string) – the email body, in plain text

  • body_html (string) – the email body, in html format (optional)

Headers:

extra headers to add to email, in the form {‘Header name’: ‘Header value’}

Type:

dict

Attachments:

a list of tuples containing file attachments to add to the email. Tuples should contain the file name and a file-like object pointing to the file contents:

[
    ('some_report.csv', file_object),
]

Optionally, you can add a third element to the tuple containing the media type. If not provided, it will be guessed using the mimetypes module:

[
    ('some_report.csv', file_object, 'text/csv'),
]
Type:

list

ckan.plugins.toolkit.mail_user(recipient: 'model.User', subject: 'str', body: 'str', body_html: 'Optional[str]' = None, headers: 'Optional[dict[str, Any]]' = None, attachments: 'Optional[Iterable[Attachment]]' = None) 'None'

Sends an email to a CKAN user.

You need to set up the Email settings to able to send emails.

Parameters:

recipient (a model.User object) – a CKAN user object

For further parameters see mail_recipient().

ckan.plugins.toolkit.missing
ckan.plugins.toolkit.navl_validate(data: 'dict[str, Any]', schema: 'dict[str, Any]', context: 'Optional[Context]' = None) 'tuple[dict[str, Any], dict[str, Any]]'

Validate an unflattened nested dict against a schema.

ckan.plugins.toolkit.redirect_to(*args: 'Any', **kw: 'Any') 'Response'

Issue a redirect: return an HTTP response with a 302 Moved header.

This is a wrapper for flask.redirect() that maintains the user’s selected language when redirecting.

The arguments to this function identify the route to redirect to, they’re the same arguments as ckan.plugins.toolkit.url_for() accepts, for example:

import ckan.plugins.toolkit as toolkit

# Redirect to /dataset/my_dataset.
return toolkit.redirect_to('dataset.read',
                    id='my_dataset')

Or, using a named route:

return toolkit.redirect_to('dataset.read', id='changed')

If given a single string as argument, this redirects without url parsing

return toolkit.redirect_to(’http://example.com’) return toolkit.redirect_to(‘/dataset’) return toolkit.redirect_to(‘/some/other/path’)

ckan.plugins.toolkit.render(template_name: 'str', extra_vars: 'Optional[dict[str, Any]]' = None) 'str'

Render a template and return the output.

This is CKAN’s main template rendering function.

Params template_name:

relative path to template inside registered tpl_dir

Params extra_vars:

additional variables available in template

ckan.plugins.toolkit.render_snippet(template: 'str', data: 'Optional[dict[str, Any]]' = None)

Render a template snippet and return the output.

See Theming guide.

ckan.plugins.toolkit.request

Flask request object.

A new request object is created for each HTTP request. It has methods and attributes for getting things like the request headers, query-string variables, request body variables, cookies, the request URL, etc.

ckan.plugins.toolkit.requires_ckan_version(min_version: 'str', max_version: 'Optional[str]' = None)

Raise CkanVersionException if the CKAN version is not greater than or equal to min_version and less then or equal to max_version.

If no max_version is given, just check whether the CKAN version is greater than or equal to min_version.

Plugins can call this function if they require a certain CKAN version, other versions of CKAN will crash if a user tries to use the plugin with them.

Parameters:
  • min_version (string) – the minimum acceptable CKAN version, eg. '2.1'

  • max_version (string) – the maximum acceptable CKAN version, eg. '2.3'

ckan.plugins.toolkit.side_effect_free(action: 'Decorated') 'Decorated'

A decorator that marks the given action function as side-effect-free.

Action functions decorated with this decorator can be called with an HTTP GET request to the Action API. Action functions that don’t have this decorator must be called with a POST request.

If your CKAN extension defines its own action functions using the IActions plugin interface, you can use this decorator to make your actions available with GET requests instead of just with POST requests.

Example:

import ckan.plugins.toolkit as toolkit

@toolkit.side_effect_free
def my_custom_action_function(context, data_dict):
    ...

(Then implement IActions to register your action function with CKAN.)

ckan.plugins.toolkit.signals

Contains ckan and ckanext namespaces for signals as well as a bunch of predefined core-level signals.

Check Signals for extra detais.

ckan.plugins.toolkit.ungettext(*args: 'Any', **kwargs: 'Any') 'str'

Translates a string with plural forms to the current locale.

Mark a string for translation that has pural forms in the format ungettext(singular, plural, n). Returns the localized unicode string of the pluralized value.

Mark a string to be localized as follows:

msg = toolkit.ungettext("Mouse", "Mice", len(mouses))
ckan.plugins.toolkit.url_for(*args: 'Any', **kw: 'Any') 'str'

Return the URL for an endpoint given some parameters.

This is a wrapper for flask.url_for() and routes.url_for() that adds some extra features that CKAN needs.

To build a URL for a Flask view, pass the name of the blueprint and the view function separated by a period ., plus any URL parameters:

url_for('api.action', ver=3, logic_function='status_show')
# Returns /api/3/action/status_show

For a fully qualified URL pass the _external=True parameter. This takes the ckan.site_url and ckan.root_path settings into account:

url_for('api.action', ver=3, logic_function='status_show',
        _external=True)
# Returns http://example.com/api/3/action/status_show

URLs built by Pylons use the Routes syntax:

url_for(controller='my_ctrl', action='my_action', id='my_dataset')
# Returns '/dataset/my_dataset'

Or, using a named route:

url_for('dataset.read', id='changed')
# Returns '/dataset/changed'

Use qualified=True for a fully qualified URL when targeting a Pylons endpoint.

For backwards compatibility, an effort is made to support the Pylons syntax when building a Flask URL, but this support might be dropped in the future, so calls should be updated.

ckan.plugins.toolkit.validate_action_data(schema_func: 'Callable[[], Schema]', can_skip_validator: 'bool' = False) 'Callable[[Action], Action]'

A decorator that validates an action function against a given schema.

Example:

def schema_func():
    return {
        "a": [get_validator("int_validator")],
        "__extras": [get_validator("ignore")]
    }

@validate_action_data(schema_function)
def my_action(context, data_dict):
    return data_dict

data = {"a": "1", "b": "2"}
assert my_action({}, data) == {"a": 1}
ckan.plugins.toolkit.validator_args(fn: Callable[..., ForwardRef('dict[str, Union[list[Validator], Schema]]')]) Callable[[], ForwardRef('dict[str, Union[list[Validator], Schema]]')]

Collect validator names from argument names and pass them to wrapped function.

Example:

@validator_args
def schema_function(not_empty, ignore):
    return not_empty, ignore

ne, ig = schema_function()
assert ne is get_validator("not_empty")
assert ig is get_validator("ignore")