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 (providing requirements.txt is not available for On-Premise deployments - see details below).

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.

On-Premise deployment details

The plugins for the on-premise deployments work exactly the same as the plugins for the SaaS version of Kili, with only a few small exceptions :

  1. It's not possible to add custom python packages to your plugin with the help of the requirements.txt file, but we selected a list of the most useful packages that you can directly use, including :
    • numpy, pandas, scikit-learn, opencv-python-headless, Pillow, requests, uuid and of course kili
  2. In order to save the logs during the execution of your plugin, you should only use the provided logger in the plugin class (the simple print function will not save the log). For an example, see the code below:
from logging import Logger
from typing import Dict
from kili.plugins import PluginCore

def custom_function(label: Dict, logger: Logger):
    logger.info("Custom function called")
    # Do something...

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

    def on_submit(self, label: Dict, asset_id: str) -> None:
        """Dedicated handler for Submit action"""
        self.logger.info("On Submit called")
        custom_function(label, self.logger)

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: Dict, asset_id: str)
on_review(self, label: Dict, asset_id: str)
on_custom_interface_click(self, label: Dict, asset_id: str)
on_project_updated(self, settings_updated: List[Dict])
on_send_back_to_queue(self, 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: Dict, asset_id: str)
        on_review(self, label: Dict, asset_id: str)
        on_custom_interface_click(self, label: Dict, asset_id: str)
        on_project_updated(self, settings_updated: List[Dict])
        on_send_back_to_queue(self, 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
            ```python
            def on_submit(self, label: Dict, 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")
        # pylint: disable=unnecessary-ellipsis

    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
            ```python
            def on_review(self, label: Dict, 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")
        # pylint: disable=unnecessary-ellipsis

    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,
            it should be used with caution.

        Args:
            label: Label submitted to Kili: a dictionary containing the following fields:
                `id`, `jsonResponse`.
            asset_id: id of the asset on which the action is called

        !!! example
            ```python
            def on_custom_interface_click(self, label: Dict, asset_id: str):
                json_response = label.get('jsonResponse')`
                label_id = label.get('id')
                issue = label_is_respecting_business_rule(json_response)
                if !issue:
                    return
                else:
                    self.kili.create_issues(
                            project_id=self.project_id,
                            label_id_array=[label_id],
                            text_array=[issue]
                        )
            ```
        """
        # pylint: disable=unused-argument
        self.logger.warning("Handler is in active development.")

    def on_project_updated(
        self,
        settings_updated: List[Dict],
    ) -> None:
        """Handler for the project updated action.

        Triggered when a project setting is updated on Kili.

        !!! warning
            This handler is in beta and is still in active development,
            it should be used with caution.

        Args:
            settings_updated: Settings updated on the project a list of
                dictionary containing the following fields:
                `key`, `newValue`, `oldValue`.
                !!! note
                    key is one of the following: 'canNavigateBetweenAssets',
                    'canSkipAsset', 'consensusTotCoverage', 'description',
                    'inputType', 'instructions', 'isAnonymized', 'jsonInterface',
                    'metadataTypes', 'minConsensusSize', 'reviewCoverage',
                    'title', 'archivedAt', 'useHoneyPot'

        !!! example
            ```python
            def on_project_updated(self, settings_updated: List[Dict]):
                for setting in settings_updated:
                    self.logger.info(setting)
                    # this will print:
                    # {'key': 'description', 'newValue': 'new desc', 'oldValue': 'old desc'}
            ```
        """
        # pylint: disable=unused-argument
        self.logger.warning("Handler is in active development.")

    def on_send_back_to_queue(
        self,
        asset_id: str,
    ) -> None:
        """Handler for send back to queue.

        Triggered when an asset is sent back to queue

        !!! warning
            This handler is in beta and is still in active development,
            it should be used with caution.

        Args:
            asset_id: Id of the asset on which was sent back to queue

        !!! example
            ```python
            def on_send_back_to_queue(self, asset_id: str):
                self.logger.info(f"Asset {asset_id} was sent back to queue")
            ```
        """
        # pylint: disable=unused-argument
        self.logger.warning("Handler is in active development.")

on_custom_interface_click(self, label, asset_id)

Handler for the custom interface click action.

Warning

This handler is in beta and is still in active development, it should be used with caution.

Parameters:

Name Type Description Default
label Dict

Label submitted to Kili: a dictionary containing the following fields: id, jsonResponse.

required
asset_id str

id of the asset on which the action is called

required

Example

def on_custom_interface_click(self, label: Dict, asset_id: str):
    json_response = label.get('jsonResponse')`
    label_id = label.get('id')
    issue = label_is_respecting_business_rule(json_response)
    if !issue:
        return
    else:
        self.kili.create_issues(
                project_id=self.project_id,
                label_id_array=[label_id],
                text_array=[issue]
            )
Source code in kili/services/plugins/model.py
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,
        it should be used with caution.

    Args:
        label: Label submitted to Kili: a dictionary containing the following fields:
            `id`, `jsonResponse`.
        asset_id: id of the asset on which the action is called

    !!! example
        ```python
        def on_custom_interface_click(self, label: Dict, asset_id: str):
            json_response = label.get('jsonResponse')`
            label_id = label.get('id')
            issue = label_is_respecting_business_rule(json_response)
            if !issue:
                return
            else:
                self.kili.create_issues(
                        project_id=self.project_id,
                        label_id_array=[label_id],
                        text_array=[issue]
                    )
        ```
    """
    # pylint: disable=unused-argument
    self.logger.warning("Handler is in active development.")

on_project_updated(self, settings_updated)

Handler for the project updated action.

Triggered when a project setting is updated on Kili.

Warning

This handler is in beta and is still in active development, it should be used with caution.

Parameters:

Name Type Description Default
settings_updated List[Dict]

Settings updated on the project a list of dictionary containing the following fields: key, newValue, oldValue.

Note

key is one of the following: 'canNavigateBetweenAssets', 'canSkipAsset', 'consensusTotCoverage', 'description', 'inputType', 'instructions', 'isAnonymized', 'jsonInterface', 'metadataTypes', 'minConsensusSize', 'reviewCoverage', 'title', 'archivedAt', 'useHoneyPot'

required

Example

def on_project_updated(self, settings_updated: List[Dict]):
    for setting in settings_updated:
        self.logger.info(setting)
        # this will print:
        # {'key': 'description', 'newValue': 'new desc', 'oldValue': 'old desc'}
Source code in kili/services/plugins/model.py
def on_project_updated(
    self,
    settings_updated: List[Dict],
) -> None:
    """Handler for the project updated action.

    Triggered when a project setting is updated on Kili.

    !!! warning
        This handler is in beta and is still in active development,
        it should be used with caution.

    Args:
        settings_updated: Settings updated on the project a list of
            dictionary containing the following fields:
            `key`, `newValue`, `oldValue`.
            !!! note
                key is one of the following: 'canNavigateBetweenAssets',
                'canSkipAsset', 'consensusTotCoverage', 'description',
                'inputType', 'instructions', 'isAnonymized', 'jsonInterface',
                'metadataTypes', 'minConsensusSize', 'reviewCoverage',
                'title', 'archivedAt', 'useHoneyPot'

    !!! example
        ```python
        def on_project_updated(self, settings_updated: List[Dict]):
            for setting in settings_updated:
                self.logger.info(setting)
                # this will print:
                # {'key': 'description', 'newValue': 'new desc', 'oldValue': 'old desc'}
        ```
    """
    # 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

def on_review(self, label: Dict, 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
        ```python
        def on_review(self, label: Dict, 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")
    # pylint: disable=unnecessary-ellipsis

on_send_back_to_queue(self, asset_id)

Handler for send back to queue.

Triggered when an asset is sent back to queue

Warning

This handler is in beta and is still in active development, it should be used with caution.

Parameters:

Name Type Description Default
asset_id str

Id of the asset on which was sent back to queue

required

Example

def on_send_back_to_queue(self, asset_id: str):
    self.logger.info(f"Asset {asset_id} was sent back to queue")
Source code in kili/services/plugins/model.py
def on_send_back_to_queue(
    self,
    asset_id: str,
) -> None:
    """Handler for send back to queue.

    Triggered when an asset is sent back to queue

    !!! warning
        This handler is in beta and is still in active development,
        it should be used with caution.

    Args:
        asset_id: Id of the asset on which was sent back to queue

    !!! example
        ```python
        def on_send_back_to_queue(self, asset_id: str):
            self.logger.info(f"Asset {asset_id} was sent back to queue")
        ```
    """
    # pylint: disable=unused-argument
    self.logger.warning("Handler is in active development.")

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

def on_submit(self, label: Dict, 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
        ```python
        def on_submit(self, label: Dict, 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")
    # pylint: disable=unnecessary-ellipsis

Queries

Set of Plugins queries.

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

    # pylint: disable=too-many-arguments

    @typechecked
    def get_plugin_build_errors(
        self,
        plugin_name: str,
        start_date: Optional[datetime] = None,
        limit: int = 100,
        skip: int = 0,
    ) -> str:
        # pylint: disable=line-too-long
        """Get paginated build errors of a plugin.

        Args:
            plugin_name: Name of the plugin
            start_date: Datetime used to get the build errors 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 build errors of the plugin,
                or an error message.

        Examples:
            >>> kili.get_plugin_build_errors(plugin_name="my_plugin_name", start_date="1970/01/01")
        """
        where = PluginBuildErrorsWhere(plugin_name=plugin_name, start_date=start_date)
        options = QueryOptions(
            first=limit, skip=skip, disable_tqdm=False
        )  # disable tqm is not implemented for this query
        pretty_result = PluginQuery(self.graphql_client, self.http_client).get_build_errors(
            where, options
        )
        return json.dumps(pretty_result, sort_keys=True, indent=4)

    @typechecked
    def get_plugin_logs(
        self,
        project_id: str,
        plugin_name: str,
        start_date: Optional[datetime] = None,
        limit: int = 100,
        skip: int = 0,
    ) -> str:
        # 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, disable_tqdm=False
        )  # disable tqm is not implemented for this query
        pretty_result = PluginQuery(self.graphql_client, self.http_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,
    ) -> str:
        """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")
        """
        return PluginUploader(
            self,  # pyright: ignore[reportGeneralTypeIssues]
            "",
            plugin_name,
            verbose,
            self.http_client,
        ).get_plugin_runner_status()

    @typechecked
    def list_plugins(
        self,
        fields: ListOrTuple[str] = ("name", "projectIds", "id", "createdAt", "updatedAt"),
    ) -> List[Dict]:
        # 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.graphql_client, self.http_client).list(fields=fields)

get_plugin_build_errors(self, plugin_name, start_date=None, limit=100, skip=0)

Get paginated build errors of a plugin.

Parameters:

Name Type Description Default
plugin_name str

Name of the plugin

required
start_date Optional[datetime.datetime]

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

None
limit int

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

100
skip int

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

0

Returns:

Type Description
str

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

Examples:

>>> kili.get_plugin_build_errors(plugin_name="my_plugin_name", start_date="1970/01/01")
Source code in kili/entrypoints/queries/plugins/__init__.py
def get_plugin_build_errors(
    self,
    plugin_name: str,
    start_date: Optional[datetime] = None,
    limit: int = 100,
    skip: int = 0,
) -> str:
    # pylint: disable=line-too-long
    """Get paginated build errors of a plugin.

    Args:
        plugin_name: Name of the plugin
        start_date: Datetime used to get the build errors 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 build errors of the plugin,
            or an error message.

    Examples:
        >>> kili.get_plugin_build_errors(plugin_name="my_plugin_name", start_date="1970/01/01")
    """
    where = PluginBuildErrorsWhere(plugin_name=plugin_name, start_date=start_date)
    options = QueryOptions(
        first=limit, skip=skip, disable_tqdm=False
    )  # disable tqm is not implemented for this query
    pretty_result = PluginQuery(self.graphql_client, self.http_client).get_build_errors(
        where, options
    )
    return json.dumps(pretty_result, sort_keys=True, indent=4)

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

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 int

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

100
skip int

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

0

Returns:

Type Description
str

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/entrypoints/queries/plugins/__init__.py
def get_plugin_logs(
    self,
    project_id: str,
    plugin_name: str,
    start_date: Optional[datetime] = None,
    limit: int = 100,
    skip: int = 0,
) -> str:
    # 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, disable_tqdm=False
    )  # disable tqm is not implemented for this query
    pretty_result = PluginQuery(self.graphql_client, self.http_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
str

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/entrypoints/queries/plugins/__init__.py
def get_plugin_status(
    self,
    plugin_name: str,
    verbose: bool = True,
) -> str:
    """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")
    """
    return PluginUploader(
        self,  # pyright: ignore[reportGeneralTypeIssues]
        "",
        plugin_name,
        verbose,
        self.http_client,
    ).get_plugin_runner_status()

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

List all plugins from your organization.

Parameters:

Name Type Description Default
fields Union[List[str], Tuple[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
List[Dict]

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/entrypoints/queries/plugins/__init__.py
def list_plugins(
    self,
    fields: ListOrTuple[str] = ("name", "projectIds", "id", "createdAt", "updatedAt"),
) -> List[Dict]:
    # 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.graphql_client, self.http_client).list(fields=fields)

Mutations

Set of Plugins mutations.

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

    @typechecked
    def upload_plugin(
        self,
        plugin_path: Optional[str] = None,
        plugin_name: Optional[str] = None,
        verbose: bool = True,
        **kwargs,  # pylint: disable=missing-param-doc
    ) -> LiteralString:
        """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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
            plugin_path,
            plugin_name,
            verbose,
            self.http_client,
        ).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,
    ) -> str:
        # 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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
            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,
    ) -> str:
        # 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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
            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) -> Optional[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

        Returns:
            A string 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, plugin_name, project_id)

    @typechecked
    def deactivate_plugin_on_project(self, plugin_name: str, project_id: str) -> 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

        Returns:
            A string 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, plugin_name, project_id)

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

        Args:
            plugin_name: Name of the plugin

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

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

    @typechecked
    def update_plugin(
        self,
        plugin_path: Optional[str] = None,
        plugin_name: Optional[str] = None,
        verbose: bool = True,
        **kwargs,  # pylint: disable=missing-param-doc
    ) -> LiteralString:
        """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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
            plugin_path,
            plugin_name,
            verbose,
            self.http_client,
        ).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

Returns:

Type Description
Optional[str]

A string 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/entrypoints/mutations/plugins/__init__.py
def activate_plugin_on_project(self, plugin_name: str, project_id: str) -> Optional[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

    Returns:
        A string 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, 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
str

A string 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/entrypoints/mutations/plugins/__init__.py
def create_webhook(
    self,
    webhook_url: str,
    plugin_name: str,
    header: Optional[str] = None,
    verbose: bool = True,
    handler_types: Optional[List[str]] = None,
) -> str:
    # 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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
        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

Returns:

Type Description
str

A string 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/entrypoints/mutations/plugins/__init__.py
def deactivate_plugin_on_project(self, plugin_name: str, project_id: str) -> 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

    Returns:
        A string 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, 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

Returns:

Type Description
str

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

Examples:

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

    Args:
        plugin_name: Name of the plugin

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

    Examples:
        >>> kili.delete_plugin(plugin_name="my_plugin_name")
    """
    return delete_plugin(self, 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
typing_extensions.LiteralString

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

Examples:

>>> kili.update_plugin(plugin_name="my_plugin_name")
Source code in kili/entrypoints/mutations/plugins/__init__.py
def update_plugin(
    self,
    plugin_path: Optional[str] = None,
    plugin_name: Optional[str] = None,
    verbose: bool = True,
    **kwargs,  # pylint: disable=missing-param-doc
) -> LiteralString:
    """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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
        plugin_path,
        plugin_name,
        verbose,
        self.http_client,
    ).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
str

A string 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/entrypoints/mutations/plugins/__init__.py
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,
) -> str:
    # 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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
        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 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 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
typing_extensions.LiteralString

A string 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/entrypoints/mutations/plugins/__init__.py
def upload_plugin(
    self,
    plugin_path: Optional[str] = None,
    plugin_name: Optional[str] = None,
    verbose: bool = True,
    **kwargs,  # pylint: disable=missing-param-doc
) -> LiteralString:
    """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 string 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,  # pyright: ignore[reportGeneralTypeIssues]
        plugin_path,
        plugin_name,
        verbose,
        self.http_client,
    ).create_plugin()