Skip to content

Plugin module

Plugins structure

A plugin is an uploaded Python script triggered by an event. It can be defined as either :

  • a single python file with everything inside
  • a plugin module (a folder) containing multiple python files and a non mandatory requirements.txt file listing all the dependencies you need for you plugin.

In the case of the module type plugin, at the root of the folder a file named main.py is strictly necessary, as it serves as the entrypoint of the plugin. In this main.py file, you can import what you need from other python files in the folder. The structure of the folder can be the following (the only constraint being the presence of the main.py file):

plugin_folder
|__ main.py
|__ other_file.py
|__ requirements.txt
|
|___helpers
    |__ helper.py

The plugin you are going to upload has to contain a class PluginHandler(PluginCore) (in the case of the module type plugin it has to be inside main.py) that implements two methods for the different types of events:

  • on_submit
  • on_review

These methods have a predefined set of parameters:

  • the label submitted (a dictionary containing the fields of the GraphQL type Label)
  • the asset_id of the asset labeled

You can add custom methods in your class as well.

Moreover, some attributes are directly available in the class:

  • self.kili
  • self.project_id

Therefore, the skeleton of the plugin (of main.py in the case of the module type plugin) should look like this:

from typing import Dict
import numpy as np

from kili.plugins import PluginCore

def custom_function():
    # Do something...

class PluginHandler(PluginCore):
    """Custom plugin"""

    def custom_method(self):
        # Do something...

    def on_review(self, label: Dict, asset_id: str) -> None:
        """Dedicated handler for Review action"""
        # Do something...

    def on_submit(self, label: Dict, asset_id: str) -> None:
        """Dedicated handler for Submit action"""
        # Do something...

Note

The plugins run has some limitations, it can use a maximum of 512 MB of ram and will timeout after 60 sec of run.

Model for Plugins

Kili Plugin core class

Parameters:

Name Type Description Default
kili Kili

kili instance that plugins can make use of

 required
project_id str

the project on which plugin is ran

 required

Implements:

on_submit(self, label: Label, asset_id: str)
on_review(self, label: Label, asset_id: str)

Warning : if using a custom init, be sure to call super().init()

Source code in kili/services/plugins/model.py 
class PluginCore:
    """
    Kili Plugin core class

    Args:
        kili: kili instance that plugins can make use of
        project_id: the project on which plugin is ran

    Implements:

        on_submit(self, label: Label, asset_id: str)
        on_review(self, label: Label, asset_id: str)

    # Warning : if using a custom init, be sure to call super().__init__()
    """

    logger: logging.Logger
    kili: Kili
    project_id: str

    def __init__(
        self, kili: Kili, project_id: str, logger: Optional[logging.Logger] = None
    ) -> None:
        self.kili = kili
        self.project_id = project_id
        if logger:
            self.logger = logger
        else:
            self.logger = get_logger()

    def on_submit(
        self,
        label: Dict,
        asset_id: str,
    ) -> None:
        """
        Handler for the submit action, triggered when a default label is submitted into Kili.

        Args:
            label: Label submitted to Kili: a dictionary containing the following fields:
                `id`, `labelType`, `numberOfAnnotations`, `authorId`, `modelName`, `jsonResponse`,
                `secondsToLabel`, `isSentBackToQueue`, `search` and some technical fields:
                `createdAt`, `updatedAt`, `version`, `isLatestReviewLabelForUser`,
                `isLatestLabelForUser`, `isLatestDefaultLabelForUser`, `readPermissionsFromProject`.
            asset_id: Id of the asset on which the label was submitted

        Example use:

            >>> def on_submit(self, label: Label, asset_id: str):
            >>>     json_response = label.get('jsonResponse')
            >>>     if label_is_respecting_business_rule(json_response):
            >>>         return
            >>>     else:
            >>>         self.kili.send_back_to_queue(asset_ids=[asset_id])

        """
        # pylint: disable=unused-argument
        self.logger.warning("Method not implemented. Define a custom on_submit on your plugin")
        pass  # pylint: disable=unnecessary-pass

    def on_review(
        self,
        label: Dict,
        asset_id: str,
    ) -> None:
        """
        Handler for the review action, triggered when a default label is reviewed on Kili

        Args:
            label: Label submitted to Kili: a dictionary containing the following fields:
                `id`, `labelType`, `numberOfAnnotations`, `authorId`, `modelName`, `jsonResponse`,
                `secondsToLabel`, `isSentBackToQueue`, `search` and `reviewedLabel` (dictionary
                that has a field `id` representing the id of the original label that was reviewed).
                It also contains some technical fields: `createdAt`, `updatedAt`, `version`,
                `isLatestReviewLabelForUser`, `isLatestLabelForUser`, `isLatestDefaultLabelForUser`,
                `readPermissionsFromProject`.
            asset_id: Id of the asset on which the label was submitted

        Example use:

            >>> def on_review(self, label: Label, asset_id: str):
            >>>     json_response = label.get('jsonResponse')
            >>>     if label_is_respecting_business_rule(json_response):
            >>>         return
            >>>     else:
            >>>         self.kili.send_back_to_queue(asset_ids=[asset_id])
        """
        # pylint: disable=unused-argument
        self.logger.warning("Method not implemented. Define a custom on_review on your plugin")
        pass  # pylint: disable=unnecessary-pass

    def on_custom_interface_click(
        self,
        label: Dict,
        asset_id: str,
    ) -> None:
        """
        Handler for the custom interface click action.

        **Warning**: This handler is in beta and is still in active development,
        so it should not be used.

        Args:
            label: label submitted to Kili
            asset_id: id of the asset on which the action is called
        """
        # pylint: disable=unused-argument
        self.logger.warning("Handler is in active development.")

on_review(self, label, asset_id)

Handler for the review action, triggered when a default label is reviewed on Kili

Parameters:

Name Type Description Default
label Dict

Label submitted to Kili: a dictionary containing the following fields: id, labelType, numberOfAnnotations, authorId, modelName, jsonResponse, secondsToLabel, isSentBackToQueue, search and reviewedLabel (dictionary that has a field id representing the id of the original label that was reviewed). It also contains some technical fields: createdAt, updatedAt, version, isLatestReviewLabelForUser, isLatestLabelForUser, isLatestDefaultLabelForUser, readPermissionsFromProject.

 required
asset_id str

Id of the asset on which the label was submitted

 required

Example use:

>>> def on_review(self, label: Label, asset_id: str):
>>>     json_response = label.get('jsonResponse')
>>>     if label_is_respecting_business_rule(json_response):
>>>         return
>>>     else:
>>>         self.kili.send_back_to_queue(asset_ids=[asset_id])
Source code in kili/services/plugins/model.py 
def on_review(
    self,
    label: Dict,
    asset_id: str,
) -> None:
    """
    Handler for the review action, triggered when a default label is reviewed on Kili

    Args:
        label: Label submitted to Kili: a dictionary containing the following fields:
            `id`, `labelType`, `numberOfAnnotations`, `authorId`, `modelName`, `jsonResponse`,
            `secondsToLabel`, `isSentBackToQueue`, `search` and `reviewedLabel` (dictionary
            that has a field `id` representing the id of the original label that was reviewed).
            It also contains some technical fields: `createdAt`, `updatedAt`, `version`,
            `isLatestReviewLabelForUser`, `isLatestLabelForUser`, `isLatestDefaultLabelForUser`,
            `readPermissionsFromProject`.
        asset_id: Id of the asset on which the label was submitted

    Example use:

        >>> def on_review(self, label: Label, asset_id: str):
        >>>     json_response = label.get('jsonResponse')
        >>>     if label_is_respecting_business_rule(json_response):
        >>>         return
        >>>     else:
        >>>         self.kili.send_back_to_queue(asset_ids=[asset_id])
    """
    # pylint: disable=unused-argument
    self.logger.warning("Method not implemented. Define a custom on_review on your plugin")
    pass  # pylint: disable=unnecessary-pass

on_submit(self, label, asset_id)

Handler for the submit action, triggered when a default label is submitted into Kili.

Parameters:

Name Type Description Default
label Dict

Label submitted to Kili: a dictionary containing the following fields: id, labelType, numberOfAnnotations, authorId, modelName, jsonResponse, secondsToLabel, isSentBackToQueue, search and some technical fields: createdAt, updatedAt, version, isLatestReviewLabelForUser, isLatestLabelForUser, isLatestDefaultLabelForUser, readPermissionsFromProject.

 required
asset_id str

Id of the asset on which the label was submitted

 required

Example use:

>>> def on_submit(self, label: Label, asset_id: str):
>>>     json_response = label.get('jsonResponse')
>>>     if label_is_respecting_business_rule(json_response):
>>>         return
>>>     else:
>>>         self.kili.send_back_to_queue(asset_ids=[asset_id])
Source code in kili/services/plugins/model.py 
def on_submit(
    self,
    label: Dict,
    asset_id: str,
) -> None:
    """
    Handler for the submit action, triggered when a default label is submitted into Kili.

    Args:
        label: Label submitted to Kili: a dictionary containing the following fields:
            `id`, `labelType`, `numberOfAnnotations`, `authorId`, `modelName`, `jsonResponse`,
            `secondsToLabel`, `isSentBackToQueue`, `search` and some technical fields:
            `createdAt`, `updatedAt`, `version`, `isLatestReviewLabelForUser`,
            `isLatestLabelForUser`, `isLatestDefaultLabelForUser`, `readPermissionsFromProject`.
        asset_id: Id of the asset on which the label was submitted

    Example use:

        >>> def on_submit(self, label: Label, asset_id: str):
        >>>     json_response = label.get('jsonResponse')
        >>>     if label_is_respecting_business_rule(json_response):
        >>>         return
        >>>     else:
        >>>         self.kili.send_back_to_queue(asset_ids=[asset_id])

    """
    # pylint: disable=unused-argument
    self.logger.warning("Method not implemented. Define a custom on_submit on your plugin")
    pass  # pylint: disable=unnecessary-pass

Queries

Set of Plugins queries.

Source code in kili/queries/plugins/__init__.py 
class QueriesPlugins:
    """Set of Plugins queries."""

    # pylint: disable=too-many-arguments

    def __init__(self, auth: KiliAuth):
        """Initialize the subclass.

        Args:
            auth: KiliAuth object
        """
        self.auth = auth

    @typechecked
    def get_plugin_logs(
        self,
        project_id: str,
        plugin_name: str,
        start_date: Optional[datetime] = None,
        limit: Optional[int] = None,
        skip: Optional[int] = None,
    ):
        # pylint: disable=line-too-long
        """Get paginated logs of a plugin on a project.
        Args:
            project_id: Identifier of the project
            plugin_name: Name of the plugin
            start_date: Datetime used to get the logs from, if not provided, it will be the plugin's creation date
            limit: Limit for pagination, if not provided, it will be 100
            skip: Skip for pagination, if not provided, it will be 0
        Returns:
            A result array which contains the logs of the plugin,
                or an error message.
        Examples:
            >>> kili.get_plugin_logs(project_id="my_project_id", plugin_name="my_plugin_name", start_date="1970/01/01")
        """

        where = PluginLogsWhere(
            project_id=project_id, plugin_name=plugin_name, start_date=start_date
        )
        options = QueryOptions(
            first=limit, skip=skip or 0, disable_tqdm=False
        )  # disable tqm is not implemnted for this query
        pretty_result = PluginQuery(self.auth.client).get_logs(where, options)
        return json.dumps(pretty_result, sort_keys=True, indent=4)

    @typechecked
    def get_plugin_status(
        self,
        plugin_name: str,
        verbose: bool = True,
    ):
        """Update a plugin with new code.

        Args:
            plugin_name: Name of the plugin
            verbose: If false, minimal logs are displayed

        Returns:
            The status of the plugin if query was successful or an error message otherwise.

        Examples:
            >>> kili.get_plugin_status(plugin_name="my_plugin_name")
        """

        result = PluginUploader(self.auth, "", plugin_name, verbose).get_plugin_runner_status()
        return result

    # pylint: disable=dangerous-default-value
    @typechecked
    def list_plugins(
        self,
        fields: List[str] = [
            "name",
            "projectIds",
            "id",
            "createdAt",
            "updatedAt",
        ],
    ):
        # pylint: disable=line-too-long
        """List all plugins from your organization
        Args:
            fields: All the fields to request among the possible fields for the plugins
                See [the documentation](https://docs.kili-technology.com/reference/graphql-api#plugins) for all possible fields.

        Returns:
            A result array which contains all the plugins from your organization,
                or an error message.
        Examples:
            >>> kili.list_plugins()
            >>> kili.list_plugins(fields=['name'])
        """
        return PluginQuery(self.auth.client).list(fields=fields)

get_plugin_logs(self, project_id, plugin_name, start_date=None, limit=None, skip=None)

Get paginated logs of a plugin on a project.

Parameters:

Name Type Description Default
project_id str

Identifier of the project

 required
plugin_name str

Name of the plugin

 required
start_date Optional[datetime.datetime]

Datetime used to get the logs from, if not provided, it will be the plugin's creation date

 None
limit Optional[int]

Limit for pagination, if not provided, it will be 100

 None
skip Optional[int]

Skip for pagination, if not provided, it will be 0

 None

Returns:

Type Description

A result array which contains the logs of the plugin, or an error message.

Examples:

>>> kili.get_plugin_logs(project_id="my_project_id", plugin_name="my_plugin_name", start_date="1970/01/01")
Source code in kili/queries/plugins/__init__.py 
@typechecked
def get_plugin_logs(
    self,
    project_id: str,
    plugin_name: str,
    start_date: Optional[datetime] = None,
    limit: Optional[int] = None,
    skip: Optional[int] = None,
):
    # pylint: disable=line-too-long
    """Get paginated logs of a plugin on a project.
    Args:
        project_id: Identifier of the project
        plugin_name: Name of the plugin
        start_date: Datetime used to get the logs from, if not provided, it will be the plugin's creation date
        limit: Limit for pagination, if not provided, it will be 100
        skip: Skip for pagination, if not provided, it will be 0
    Returns:
        A result array which contains the logs of the plugin,
            or an error message.
    Examples:
        >>> kili.get_plugin_logs(project_id="my_project_id", plugin_name="my_plugin_name", start_date="1970/01/01")
    """

    where = PluginLogsWhere(
        project_id=project_id, plugin_name=plugin_name, start_date=start_date
    )
    options = QueryOptions(
        first=limit, skip=skip or 0, disable_tqdm=False
    )  # disable tqm is not implemnted for this query
    pretty_result = PluginQuery(self.auth.client).get_logs(where, options)
    return json.dumps(pretty_result, sort_keys=True, indent=4)

get_plugin_status(self, plugin_name, verbose=True)

Update a plugin with new code.

Parameters:

Name Type Description Default
plugin_name str

Name of the plugin

 required
verbose bool

If false, minimal logs are displayed

 True

Returns:

Type Description

The status of the plugin if query was successful or an error message otherwise.

Examples:

>>> kili.get_plugin_status(plugin_name="my_plugin_name")
Source code in kili/queries/plugins/__init__.py 
@typechecked
def get_plugin_status(
    self,
    plugin_name: str,
    verbose: bool = True,
):
    """Update a plugin with new code.

    Args:
        plugin_name: Name of the plugin
        verbose: If false, minimal logs are displayed

    Returns:
        The status of the plugin if query was successful or an error message otherwise.

    Examples:
        >>> kili.get_plugin_status(plugin_name="my_plugin_name")
    """

    result = PluginUploader(self.auth, "", plugin_name, verbose).get_plugin_runner_status()
    return result

list_plugins(self, fields=['name', 'projectIds', 'id', 'createdAt', 'updatedAt'])

List all plugins from your organization

Parameters:

Name Type Description Default
fields List[str]

All the fields to request among the possible fields for the plugins See the documentation for all possible fields.

 ['name', 'projectIds', 'id', 'createdAt', 'updatedAt']

Returns:

Type Description

A result array which contains all the plugins from your organization, or an error message.

Examples:

>>> kili.list_plugins()
>>> kili.list_plugins(fields=['name'])
Source code in kili/queries/plugins/__init__.py 
@typechecked
def list_plugins(
    self,
    fields: List[str] = [
        "name",
        "projectIds",
        "id",
        "createdAt",
        "updatedAt",
    ],
):
    # pylint: disable=line-too-long
    """List all plugins from your organization
    Args:
        fields: All the fields to request among the possible fields for the plugins
            See [the documentation](https://docs.kili-technology.com/reference/graphql-api#plugins) for all possible fields.

    Returns:
        A result array which contains all the plugins from your organization,
            or an error message.
    Examples:
        >>> kili.list_plugins()
        >>> kili.list_plugins(fields=['name'])
    """
    return PluginQuery(self.auth.client).list(fields=fields)

Mutations

Set of Plugins mutations.

Source code in kili/mutations/plugins/__init__.py 
class MutationsPlugins:
    """Set of Plugins mutations."""

    def __init__(self, auth: KiliAuth):
        """Initialize the subclass.

        Args:
            auth: KiliAuth object
        """
        self.auth = auth

    @typechecked
    def upload_plugin(
        self,
        plugin_path: Optional[str] = None,
        plugin_name: Optional[str] = None,
        verbose: bool = True,
        **kwargs
    ):
        """Uploads a plugin.

        Args:
            plugin_path : Path to your plugin. Either:

                - a folder containing a main.py (mandatory) and a requirements.txt (optional)
                - a .py file
            plugin_name: name of your plugin, if not provided, it will be the name from your file
            verbose: If false, minimal logs are displayed

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.upload_plugin(plugin_path="./path/to/my/folder")
            >>> kili.upload_plugin(plugin_path="./path/to/my/file.py")
        """

        if kwargs.get("file_path"):
            raise TypeError(
                '"file_path" has been deprecated for "plugin_path", please use "plugin_path"'
                " instead"
            )

        if not plugin_path:
            raise TypeError('"plugin_path is nullish, please provide a value')

        return PluginUploader(self.auth, plugin_path, plugin_name, verbose).create_plugin()

    @typechecked
    def create_webhook(
        self,
        webhook_url: str,
        plugin_name: str,
        header: Optional[str] = None,
        verbose: bool = True,
        handler_types: Optional[List[str]] = None,
    ):
        # pylint: disable=line-too-long,too-many-arguments
        """
        Create a webhook linked to Kili's events.
        For a complete example, refer to the notebook `webhooks_example` on kili repo

        Args:
            webhook_url: URL receiving post requests on events on Kili. The payload will be the following:

                - eventType: the type of event called
                - logPayload:
                    - runId: a unique identifier of the run for observability
                    - projectId: the Kili project the webhook is called on
                - payload: the event produced, for example for `onSubmit` event:
                    - label: the label produced
                    - asset_id: the asset on which the label is produced
            plugin_name: name of your plugin
            header: Authorization header to access the routes
            verbose: If false, minimal logs are displayed
            handler_types: List of actions for which the webhook should be called.
                Possible variants: `onSubmit`, `onReview`.
                By default, is [`onSubmit`, `onReview`].

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.create_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
        """

        return WebhookUploader(
            self.auth, webhook_url, plugin_name, header, verbose, handler_types
        ).create_webhook()

    @typechecked
    def update_webhook(
        self,
        new_webhook_url: str,
        plugin_name: str,
        new_header: Optional[str] = None,
        verbose: bool = True,
        handler_types: Optional[List[str]] = None,
    ):
        # pylint: disable=line-too-long,too-many-arguments
        """
        Update a webhook linked to Kili's events.
        For a complete example, refer to the notebook `webhooks_example` on kili repo

        Args:
            new_webhook_url: New URL receiving post requests on events on Kili. See `create_webhook` for the payload description
            plugin_name: name of your plugin
            new_header: Authorization header to access the routes
            verbose: If false, minimal logs are displayed
            handler_types: List of actions for which the webhook should be called.
                Possible variants: `onSubmit`, `onReview`.
                By default, is [`onSubmit`, `onReview`]

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.update_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
        """

        return WebhookUploader(
            self.auth, new_webhook_url, plugin_name, new_header, verbose, handler_types
        ).update_webhook()

    @typechecked
    def activate_plugin_on_project(
        self,
        plugin_name: str,
        project_id: str,
    ):
        # pylint: disable=line-too-long
        """Activates a plugin on a project.

        Args:
            plugin_name: Name of the plugin
            project_id: Identifier of the project
            verbose: If false, minimal logs are displayed

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.activate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
        """

        return activate_plugin(self.auth, plugin_name, project_id)

    @typechecked
    def deactivate_plugin_on_project(
        self,
        plugin_name: str,
        project_id: str,
    ):
        # pylint: disable=line-too-long
        """Activates a plugin on a project.

        Args:
            plugin_name: Name of the plugin
            project_id: Identifier of the project
            verbose: If false, minimal logs are displayed

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.deactivate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
        """

        return deactivate_plugin(self.auth, plugin_name, project_id)

    @typechecked
    def delete_plugin(
        self,
        plugin_name: str,
    ):
        """Deletes a plugin.

        Args:
            plugin_name: Name of the plugin
            verbose: If false, minimal logs are displayed

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.delete_plugin(plugin_name="my_plugin_name")
        """

        return delete_plugin(self.auth, plugin_name)

    @typechecked
    def update_plugin(
        self,
        plugin_path: Optional[str] = None,
        plugin_name: Optional[str] = None,
        verbose: bool = True,
        **kwargs
    ):
        """Update a plugin with new code.

        Args:
            plugin_path: Path to your plugin. Either:

                - a folder containing a main.py (mandatory) and a requirements.txt (optional)
                - a .py file
            plugin_name: Name of the plugin
            verbose: If false, minimal logs are displayed

        Returns:
            A result object which indicates if the mutation was successful,
                or an error message.

        Examples:
            >>> kili.update_plugin(plugin_name="my_plugin_name")
        """

        if kwargs.get("file_path"):
            raise TypeError(
                """ "file_path" has been deprecated for "plugin_path",
                please use "plugin_path" instead"""
            )

        if not plugin_path:
            raise TypeError('"plugin_path is nullish, please provide a value')

        if not plugin_name:
            raise TypeError('"plugin_name is nullish, please provide a value')

        return PluginUploader(self.auth, plugin_path, plugin_name, verbose).update_plugin()

activate_plugin_on_project(self, plugin_name, project_id)

Activates a plugin on a project.

Parameters:

Name Type Description Default
plugin_name str

Name of the plugin

 required
project_id str

Identifier of the project

 required
verbose

If false, minimal logs are displayed

 required

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.activate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def activate_plugin_on_project(
    self,
    plugin_name: str,
    project_id: str,
):
    # pylint: disable=line-too-long
    """Activates a plugin on a project.

    Args:
        plugin_name: Name of the plugin
        project_id: Identifier of the project
        verbose: If false, minimal logs are displayed

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.activate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
    """

    return activate_plugin(self.auth, plugin_name, project_id)

create_webhook(self, webhook_url, plugin_name, header=None, verbose=True, handler_types=None)

Create a webhook linked to Kili's events. For a complete example, refer to the notebook webhooks_example on kili repo

Parameters:

Name Type Description Default
webhook_url str

URL receiving post requests on events on Kili. The payload will be the following:

  • eventType: the type of event called
  • logPayload:
    • runId: a unique identifier of the run for observability
    • projectId: the Kili project the webhook is called on
  • payload: the event produced, for example for onSubmit event:
    • label: the label produced
    • asset_id: the asset on which the label is produced
 required
plugin_name str

name of your plugin

 required
header Optional[str]

Authorization header to access the routes

 None
verbose bool

If false, minimal logs are displayed

 True
handler_types Optional[List[str]]

List of actions for which the webhook should be called. Possible variants: onSubmit, onReview. By default, is [onSubmit, onReview].

 None

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.create_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def create_webhook(
    self,
    webhook_url: str,
    plugin_name: str,
    header: Optional[str] = None,
    verbose: bool = True,
    handler_types: Optional[List[str]] = None,
):
    # pylint: disable=line-too-long,too-many-arguments
    """
    Create a webhook linked to Kili's events.
    For a complete example, refer to the notebook `webhooks_example` on kili repo

    Args:
        webhook_url: URL receiving post requests on events on Kili. The payload will be the following:

            - eventType: the type of event called
            - logPayload:
                - runId: a unique identifier of the run for observability
                - projectId: the Kili project the webhook is called on
            - payload: the event produced, for example for `onSubmit` event:
                - label: the label produced
                - asset_id: the asset on which the label is produced
        plugin_name: name of your plugin
        header: Authorization header to access the routes
        verbose: If false, minimal logs are displayed
        handler_types: List of actions for which the webhook should be called.
            Possible variants: `onSubmit`, `onReview`.
            By default, is [`onSubmit`, `onReview`].

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.create_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
    """

    return WebhookUploader(
        self.auth, webhook_url, plugin_name, header, verbose, handler_types
    ).create_webhook()

deactivate_plugin_on_project(self, plugin_name, project_id)

Activates a plugin on a project.

Parameters:

Name Type Description Default
plugin_name str

Name of the plugin

 required
project_id str

Identifier of the project

 required
verbose

If false, minimal logs are displayed

 required

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.deactivate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def deactivate_plugin_on_project(
    self,
    plugin_name: str,
    project_id: str,
):
    # pylint: disable=line-too-long
    """Activates a plugin on a project.

    Args:
        plugin_name: Name of the plugin
        project_id: Identifier of the project
        verbose: If false, minimal logs are displayed

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.deactivate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
    """

    return deactivate_plugin(self.auth, plugin_name, project_id)

delete_plugin(self, plugin_name)

Deletes a plugin.

Parameters:

Name Type Description Default
plugin_name str

Name of the plugin

 required
verbose

If false, minimal logs are displayed

 required

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.delete_plugin(plugin_name="my_plugin_name")
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def delete_plugin(
    self,
    plugin_name: str,
):
    """Deletes a plugin.

    Args:
        plugin_name: Name of the plugin
        verbose: If false, minimal logs are displayed

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.delete_plugin(plugin_name="my_plugin_name")
    """

    return delete_plugin(self.auth, plugin_name)

update_plugin(self, plugin_path=None, plugin_name=None, verbose=True, **kwargs)

Update a plugin with new code.

Parameters:

Name Type Description Default
plugin_path Optional[str]

Path to your plugin. Either:

  • a folder containing a main.py (mandatory) and a requirements.txt (optional)
  • a .py file
 None
plugin_name Optional[str]

Name of the plugin

 None
verbose bool

If false, minimal logs are displayed

 True

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.update_plugin(plugin_name="my_plugin_name")
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def update_plugin(
    self,
    plugin_path: Optional[str] = None,
    plugin_name: Optional[str] = None,
    verbose: bool = True,
    **kwargs
):
    """Update a plugin with new code.

    Args:
        plugin_path: Path to your plugin. Either:

            - a folder containing a main.py (mandatory) and a requirements.txt (optional)
            - a .py file
        plugin_name: Name of the plugin
        verbose: If false, minimal logs are displayed

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.update_plugin(plugin_name="my_plugin_name")
    """

    if kwargs.get("file_path"):
        raise TypeError(
            """ "file_path" has been deprecated for "plugin_path",
            please use "plugin_path" instead"""
        )

    if not plugin_path:
        raise TypeError('"plugin_path is nullish, please provide a value')

    if not plugin_name:
        raise TypeError('"plugin_name is nullish, please provide a value')

    return PluginUploader(self.auth, plugin_path, plugin_name, verbose).update_plugin()

update_webhook(self, new_webhook_url, plugin_name, new_header=None, verbose=True, handler_types=None)

Update a webhook linked to Kili's events. For a complete example, refer to the notebook webhooks_example on kili repo

Parameters:

Name Type Description Default
new_webhook_url str

New URL receiving post requests on events on Kili. See create_webhook for the payload description

 required
plugin_name str

name of your plugin

 required
new_header Optional[str]

Authorization header to access the routes

 None
verbose bool

If false, minimal logs are displayed

 True
handler_types Optional[List[str]]

List of actions for which the webhook should be called. Possible variants: onSubmit, onReview. By default, is [onSubmit, onReview]

 None

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.update_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def update_webhook(
    self,
    new_webhook_url: str,
    plugin_name: str,
    new_header: Optional[str] = None,
    verbose: bool = True,
    handler_types: Optional[List[str]] = None,
):
    # pylint: disable=line-too-long,too-many-arguments
    """
    Update a webhook linked to Kili's events.
    For a complete example, refer to the notebook `webhooks_example` on kili repo

    Args:
        new_webhook_url: New URL receiving post requests on events on Kili. See `create_webhook` for the payload description
        plugin_name: name of your plugin
        new_header: Authorization header to access the routes
        verbose: If false, minimal logs are displayed
        handler_types: List of actions for which the webhook should be called.
            Possible variants: `onSubmit`, `onReview`.
            By default, is [`onSubmit`, `onReview`]

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.update_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
    """

    return WebhookUploader(
        self.auth, new_webhook_url, plugin_name, new_header, verbose, handler_types
    ).update_webhook()

upload_plugin(self, plugin_path=None, plugin_name=None, verbose=True, **kwargs)

Uploads a plugin.

Parameters:

Name Type Description Default
plugin_path

Path to your plugin. Either:

  • a folder containing a main.py (mandatory) and a requirements.txt (optional)
  • a .py file
 None
plugin_name Optional[str]

name of your plugin, if not provided, it will be the name from your file

 None
verbose bool

If false, minimal logs are displayed

 True

Returns:

Type Description

A result object which indicates if the mutation was successful, or an error message.

Examples:

>>> kili.upload_plugin(plugin_path="./path/to/my/folder")
>>> kili.upload_plugin(plugin_path="./path/to/my/file.py")
Source code in kili/mutations/plugins/__init__.py 
@typechecked
def upload_plugin(
    self,
    plugin_path: Optional[str] = None,
    plugin_name: Optional[str] = None,
    verbose: bool = True,
    **kwargs
):
    """Uploads a plugin.

    Args:
        plugin_path : Path to your plugin. Either:

            - a folder containing a main.py (mandatory) and a requirements.txt (optional)
            - a .py file
        plugin_name: name of your plugin, if not provided, it will be the name from your file
        verbose: If false, minimal logs are displayed

    Returns:
        A result object which indicates if the mutation was successful,
            or an error message.

    Examples:
        >>> kili.upload_plugin(plugin_path="./path/to/my/folder")
        >>> kili.upload_plugin(plugin_path="./path/to/my/file.py")
    """

    if kwargs.get("file_path"):
        raise TypeError(
            '"file_path" has been deprecated for "plugin_path", please use "plugin_path"'
            " instead"
        )

    if not plugin_path:
        raise TypeError('"plugin_path is nullish, please provide a value')

    return PluginUploader(self.auth, plugin_path, plugin_name, verbose).create_plugin()