Plugin module
Plugins structure
A plugin is an uploaded Python script triggered by an event. It can be defined as either :
- a single
pythonfile with everything inside - a plugin module (a folder) containing multiple
pythonfiles and a non mandatoryrequirements.txtfile listing all the dependencies you need for you plugin (providingrequirements.txtis 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_submiton_review
These methods have a predefined set of parameters:
- the
labelsubmitted (a dictionary containing the fields of the GraphQL type Label) - the
asset_idof the asset labeled
You can add custom methods in your class as well.
Moreover, some attributes are directly available in the class:
self.kiliself.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 :
- It's not possible to add custom python packages to your plugin with the help of the
requirements.txtfile, 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,uuidand of coursekili
- 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
printfunction 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)
on_event(self, payload: Dict)
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)
on_event(self, payload: Dict)
!!! 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`, `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")
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.")
def on_event(
self,
payload: Dict,
) -> None:
"""Handler for all events, triggered when an event is triggered.
Args:
payload: Dict.
"""
# pylint: disable=unused-argument
self.logger.warning("Method not implemented. Define a custom on_event on your plugin")
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:
|
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)
Handler for all events, triggered when an event is triggered.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload |
Dict |
Dict. |
required |
Source code in kili/services/plugins/model.py
def on_event(
self,
payload: Dict,
) -> None:
"""Handler for all events, triggered when an event is triggered.
Args:
payload: Dict.
"""
# pylint: disable=unused-argument
self.logger.warning("Method not implemented. Define a custom on_event on your plugin")
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:
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:
|
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")
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:
|
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[reportGeneralTypeIssues]
"",
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[reportGeneralTypeIssues]
"",
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[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://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: List of events for which the plugin should be called.
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,
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[reportGeneralTypeIssues]
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[reportGeneralTypeIssues]
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: List of events names and/or globs for which the plugin should be called.
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,
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:
|
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: |
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[reportGeneralTypeIssues]
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:
|
None |
plugin_name |
Optional[str] |
Name of the plugin |
None |
event_matcher |
Optional[List[str]] |
List of events names and/or globs for which the plugin should be called. |
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: List of events names and/or globs for which the plugin should be called.
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,
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 |
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: |
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[reportGeneralTypeIssues]
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:
|
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]] |
List of events for which the plugin should be called. |
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: List of events for which the plugin should be called.
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,
event_matcher,
).create_plugin()