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.

Receiving Kili events

Instead of on_submit and on_review, a plugin can receive any Kili event: decorate a method of PluginHandler with on_kili_event, and annotate its event parameter with the events to receive. The method is called with each of them, parsed into its kili_events model — typed fields, completion in your IDE. The Kili events documentation lists every event, with its model and its fields.

from kili_events import AssetSkippedEvent, AssetWorkflowEvent, LabelWorkflowEvent

from kili.plugins import PluginCore, on_kili_event


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

    @on_kili_event
    def on_skip(self, event: AssetSkippedEvent) -> None:
        self.logger.info(f"Asset {event.payload.asset_id} skipped by {event.user_id}")

    @on_kili_event
    def on_workflow(self, event: AssetWorkflowEvent | LabelWorkflowEvent) -> None:
        self.logger.info(f"Received {event.event} on project {event.project_id}")

The annotation is the subscription, and is one of:

Annotation	Receives	Example
an event model	that event	`AssetSkippedEvent`
a family of events	every event of the family, including those Kili adds to it later	`AssetWorkflowEvent`: every `asset.workflow.…` event
a union of them	every event of each	`Union[AssetSkippedEvent, LabelWorkflowEvent]`

The union can also be written AssetWorkflowEvent | LabelWorkflowEvent, as in on_workflow. AnyEvent receives every event. Events that share no family, such as every ….started event, are received with a union of their models or families. on_kili_event takes no arguments, and is the only decorator of the method.

Every method whose annotation matches an event receives it, in the alphabetical order of their names; when one raises, the others still run, its error is logged, and the run fails with the first error. Upload the plugin without event_matcher: the events it receives are read from the annotations.

kili.upload_plugin(plugin_path="./my_plugin/", plugin_name="my_plugin")

The upload reads your code without running it, so:

the decorated methods are those of the PluginHandler class of the plugin file (of main.py for a folder) — not the ones it inherits — and plain methods taking self and the event, not async ones;
the annotation is made of names imported from kili_events when the class is defined (not under if TYPE_CHECKING: — with ruff, list kili.plugins.on_kili_event in runtime-evaluated-decorators), or a module-level alias of them — anything else stops the upload with the reason;
Kili never sends plugins its system events, nor the events of no project (assetStorage.integration.…): an annotation naming a model or a family made only of those stops the upload;
a plugin receives events either with on_kili_event or with on_submit, on_review, on_custom_interface_click and on_send_back_to_queue, not both.

A family is uploaded as the pattern of its folder (asset.workflow.*), so an event Kili adds to it later reaches the plugin without changing its code. The method receives it once the plugin runs with a version of kili_events that knows it: on Kili's cloud, the packages are installed when the plugin is uploaded, so update the plugin (kili.update_plugin); on-premise, they are those of the plugins runner. Until then, like any event the plugin's version of kili_events does not know, it is logged and skipped; an event that does not match its model is logged as an error and skipped.

The event carries its subject (event.event), organization_id, project_id, user_id and payload, as its model declares them — no event id or timestamp.

Migrating from `on_event`

Overriding on_event with an event_matcher still works, but is deprecated in favour of on_kili_event. on_event now receives the whole event: the fields it received before are under payload["payload"], beside event, organizationId, projectId and userId. Its event_matcher takes subjects and patterns, where * stands for one or more names (label.workflow.*, project.workflow.bulkAction.*.started).

Before:

class PluginHandler(PluginCore):
    def on_event(self, payload: dict) -> None:
        if payload["event"] == "asset.skipped":
            self.logger.info(f"Asset {payload['assetId']} skipped")

kili.upload_plugin(plugin_path="./my_plugin/", event_matcher=["asset.skipped"])

After, with on_kili_event:

class PluginHandler(PluginCore):
    @on_kili_event
    def on_skip(self, event: AssetSkippedEvent) -> None:
        self.logger.info(f"Asset {event.payload.asset_id} skipped")

kili.upload_plugin(plugin_path="./my_plugin/")

Or after, keeping on_event:

class PluginHandler(PluginCore):
    def on_event(self, payload: dict) -> None:
        if payload["event"] == "asset.skipped":
            self.logger.info(f"Asset {payload['payload']['assetId']} skipped")

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 :

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
on_kili_event needs the version of kili that introduces it, and kili-events: until your plugins-runner ships them, receive events with on_submit and on_review. 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_send_back_to_queue(self, asset_id: str)

or methods decorated with @on_kili_event, which receive the Kili events their parameter's annotation names, as kili_events models.

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_send_back_to_queue(self, asset_id: str)

    or methods decorated with `@on_kili_event`, which receive the Kili events their
    parameter's annotation names, as kili_events models.

    !!! 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")

    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` and `search` (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")

    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_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.")

    def on_event(
        self,
        payload: dict,
    ) -> None:
        """Receives every event of the plugin, and calls the methods subscribed to it.

        The event is parsed into its kili_events model, then passed to each method decorated
        with `@on_kili_event` whose annotation it is an instance of. Decorate methods rather than
        overriding this one.

        Args:
            payload: The event as Kili sends it: `event` (its subject, e.g. `asset.skipped`),
                `organizationId`, `projectId`, `userId` and `payload`, for the fields the
                event declares.
        """
        try:
            event = parse_event(payload)
        except UnknownEventError as error:
            self.logger.warning(f"{error} The event is ignored.")
            return
        except ValidationError as error:
            self.logger.error(
                f"The event '{payload.get('event')}' does not match its kili_events model, so no"
                f" method receives it: {error}"
            )
            return

        subscriptions = subscribed_methods(type(self))
        handlers = [name for name, models in subscriptions.items() if isinstance(event, models)]
        if not handlers:
            self.logger.warning(
                f"No method of the plugin receives the event '{event.event}' (methods decorated"
                f" with @on_kili_event: {', '.join(subscriptions) or 'none'})."
            )
            return

        # Each method runs even when another one fails, then the run fails with the first error.
        errors: list[Exception] = []
        for name in handlers:
            try:
                getattr(self, name)(event)
            except Exception as error:  # pylint: disable=broad-exception-caught
                self.logger.error(f"{name} failed on '{event.event}': {error!r}")
                errors.append(error)
        if errors:
            raise errors[0]

`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_event(self, payload)`

Receives every event of the plugin, and calls the methods subscribed to it.

The event is parsed into its kili_events model, then passed to each method decorated with @on_kili_event whose annotation it is an instance of. Decorate methods rather than overriding this one.

Parameters:

Name	Type	Description	Default
`payload`	`dict`	The event as Kili sends it: `event` (its subject, e.g. `asset.skipped`), `organizationId`, `projectId`, `userId` and `payload`, for the fields the event declares.	required

Source code in kili/services/plugins/model.py

def on_event(
    self,
    payload: dict,
) -> None:
    """Receives every event of the plugin, and calls the methods subscribed to it.

    The event is parsed into its kili_events model, then passed to each method decorated
    with `@on_kili_event` whose annotation it is an instance of. Decorate methods rather than
    overriding this one.

    Args:
        payload: The event as Kili sends it: `event` (its subject, e.g. `asset.skipped`),
            `organizationId`, `projectId`, `userId` and `payload`, for the fields the
            event declares.
    """
    try:
        event = parse_event(payload)
    except UnknownEventError as error:
        self.logger.warning(f"{error} The event is ignored.")
        return
    except ValidationError as error:
        self.logger.error(
            f"The event '{payload.get('event')}' does not match its kili_events model, so no"
            f" method receives it: {error}"
        )
        return

    subscriptions = subscribed_methods(type(self))
    handlers = [name for name, models in subscriptions.items() if isinstance(event, models)]
    if not handlers:
        self.logger.warning(
            f"No method of the plugin receives the event '{event.event}' (methods decorated"
            f" with @on_kili_event: {', '.join(subscriptions) or 'none'})."
        )
        return

    # Each method runs even when another one fails, then the run fails with the first error.
    errors: list[Exception] = []
    for name in handlers:
        try:
            getattr(self, name)(event)
        except Exception as error:  # pylint: disable=broad-exception-caught
            self.logger.error(f"{name} failed on '{event.event}': {error!r}")
            errors.append(error)
    if errors:
        raise errors[0]

`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` and `search` (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` and `search` (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")

`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")

Queries

Set of Plugins queries.

Source code in kili/entrypoints/queries/plugins/__init__.py

@for_all_methods(log_call, exclude=["__init__"])
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:
        """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[reportArgumentType]
            "",
            plugin_name,
            verbose,
            self.http_client,
            event_matcher=None,
        ).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://api-docs.kili-technology.com/types/objects/plugin) 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:
    """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[reportArgumentType]
        "",
        plugin_name,
        verbose,
        self.http_client,
        event_matcher=None,
    ).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`	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://api-docs.kili-technology.com/types/objects/plugin) 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

@for_all_methods(log_call, exclude=["__init__"])
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,
        event_matcher: Optional[list[str]] = None,
        **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
            event_matcher: Events for which a plugin overriding `on_event` is called. A plugin
                using `@on_kili_event` leaves it out: its events come from the annotations.
            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[reportArgumentType]
            plugin_path,
            plugin_name,
            verbose,
            self.http_client,
            event_matcher,
        ).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,
        event_matcher: 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`].
            event_matcher: List of events for which the webhook should be called.

        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[reportArgumentType]
            webhook_url,
            plugin_name,
            header,
            verbose,
            handler_types,
            event_matcher,
        ).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,
        event_matcher: 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`]
            event_matcher: List of events for which the webhook should be called.

        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[reportArgumentType]
            new_webhook_url,
            plugin_name,
            new_header,
            verbose,
            handler_types,
            event_matcher,
        ).update_webhook()

    @typechecked
    def activate_plugin_on_project(self, plugin_name: str, project_id: str) -> Optional[str]:
        """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:
        """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,
        event_matcher: Optional[list[str]] = None,
        **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
            event_matcher: Events for which a plugin overriding `on_event` is called. A plugin
                using `@on_kili_event` leaves it out: its events come from the annotations.
            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[reportArgumentType]
            plugin_path,
            plugin_name,
            verbose,
            self.http_client,
            event_matcher,
        ).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]:
    """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, event_matcher=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`
`event_matcher`	`Optional[list[str]]`	List of events for which the webhook should be called.	`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,
    event_matcher: 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`].
        event_matcher: List of events for which the webhook should be called.

    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[reportArgumentType]
        webhook_url,
        plugin_name,
        header,
        verbose,
        handler_types,
        event_matcher,
    ).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:
    """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, event_matcher=None, **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`
`event_matcher`	`Optional[list[str]]`	Events for which a plugin overriding `on_event` is called. A plugin using `@on_kili_event` leaves it out: its events come from the annotations.	`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,
    event_matcher: Optional[list[str]] = None,
    **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
        event_matcher: Events for which a plugin overriding `on_event` is called. A plugin
            using `@on_kili_event` leaves it out: its events come from the annotations.
        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[reportArgumentType]
        plugin_path,
        plugin_name,
        verbose,
        self.http_client,
        event_matcher,
    ).update_plugin()

`update_webhook(self, new_webhook_url, plugin_name, new_header=None, verbose=True, handler_types=None, event_matcher=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`
`event_matcher`	`Optional[list[str]]`	List of events for which the webhook should be called.	`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,
    event_matcher: 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`]
        event_matcher: List of events for which the webhook should be called.

    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[reportArgumentType]
        new_webhook_url,
        plugin_name,
        new_header,
        verbose,
        handler_types,
        event_matcher,
    ).update_webhook()

`upload_plugin(self, plugin_path=None, plugin_name=None, verbose=True, event_matcher=None, **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`
`event_matcher`	`Optional[list[str]]`	Events for which a plugin overriding `on_event` is called. A plugin using `@on_kili_event` leaves it out: its events come from the annotations.	`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,
    event_matcher: Optional[list[str]] = None,
    **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
        event_matcher: Events for which a plugin overriding `on_event` is called. A plugin
            using `@on_kili_event` leaves it out: its events come from the annotations.
        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[reportArgumentType]
        plugin_path,
        plugin_name,
        verbose,
        self.http_client,
        event_matcher,
    ).create_plugin()

Plugin module

Plugins structure

Receiving Kili events

Migrating from on_event

On-Premise deployment details

Model for Plugins

on_custom_interface_click(self, label, asset_id)

on_event(self, payload)

on_review(self, label, asset_id)

on_send_back_to_queue(self, asset_id)

on_submit(self, label, asset_id)

Queries

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

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

get_plugin_status(self, plugin_name, verbose=True)

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

Mutations

activate_plugin_on_project(self, plugin_name, project_id)

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

deactivate_plugin_on_project(self, plugin_name, project_id)

delete_plugin(self, plugin_name)

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

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

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

Migrating from `on_event`

`on_custom_interface_click(self, label, asset_id)`

`on_event(self, payload)`

`on_review(self, label, asset_id)`

`on_send_back_to_queue(self, asset_id)`

`on_submit(self, label, asset_id)`

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

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

`get_plugin_status(self, plugin_name, verbose=True)`

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

`activate_plugin_on_project(self, plugin_name, project_id)`

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

`deactivate_plugin_on_project(self, plugin_name, project_id)`

`delete_plugin(self, plugin_name)`

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

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

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