Plugin module
Plugins structure
A plugin is a set of :
- a
python
file - a non mandatory
requirements.txt
file listing all the dependencies you need for you plugin.
The plugin you are going to upload has to contain a class
and two methods for the different types of events:
on_submit
on_review
You can add custom methods in your class as well.
Some attributes are available in the class:
self.kili
self.project_id
The skeleton of the plugin should look like this:
from kili.plugins import PluginCore
from kili.types import Label
import numpy as np
def custom_function():
# Do something...
class PluginHandler(PluginCore):
"""Custom plugin"""
def custom_method(self):
# Do something...
def on_review(self, label: Label, asset_id: str) -> None:
"""Dedicated handler for Review action"""
# Do something...
def on_submit(self, label: Label, asset_id: str) -> None:
"""Dedicated handler for Submit action"""
# Do something...
Note
The plugins run has some limitations, it can use a maximum of 512mb of ram and will timeout after 60sec of run
Model for Plugins
Kili Plugin core class
Parameters:
Name | Type | Description | Default |
---|---|---|---|
kili |
Kili |
kili instance that plugins can make use of |
required |
project_id |
str |
the project on which plugin is ran |
required |
Implements:
on_submit(self, label: Label, asset_id: str)
on_review(self, label: Label, asset_id: str)
Warning : if using a custom init, be sure to call super().init()
Source code in kili/services/plugins/model.py
class PluginCore:
"""
Kili Plugin core class
Args:
kili: kili instance that plugins can make use of
project_id: the project on which plugin is ran
Implements:
on_submit(self, label: Label, asset_id: str)
on_review(self, label: Label, asset_id: str)
# Warning : if using a custom init, be sure to call super().__init__()
"""
logger: logging.Logger
kili: Kili
project_id: str
def __init__(
self, kili: Kili, project_id: str, logger: Optional[logging.Logger] = None
) -> None:
self.kili = kili
self.project_id = project_id
if logger:
self.logger = logger
else:
self.logger = logging.getLogger()
def on_submit(
self,
label: Label,
asset_id: str,
) -> None:
"""
Handler for the submit action, triggered when a default label is submitted into Kili.
Args:
label: label submitted to Kili
asset_id: id of the asset on which the label was submitted
Example use:
>>> def on_submit(self, label: Label, asset_id: str):
>>> json_response = label.get('jsonResponse')
>>> if label_is_respecting_business_rule(json_response):
>>> return
>>> else:
>>> self.kili.send_back_to_queue(asset_ids=[asset_id])
"""
# pylint: disable=unused-argument
self.logger.warning("Method not implemented. Define a custom on_submit on your plugin")
pass # pylint: disable=unnecessary-pass
def on_review(
self,
label: Label,
asset_id: str,
) -> None:
"""
Handler for the review action, triggered when a default label is reviewed on Kili
Args:
label: label submitted to Kili
asset_id: id of the asset on which the label was submitted
Example use:
>>> def on_review(self, label: Label, asset_id: str):
>>> json_response = label.get('jsonResponse')
>>> if label_is_respecting_business_rule(json_response):
>>> return
>>> else:
>>> self.kili.send_back_to_queue(asset_ids=[asset_id])
"""
# pylint: disable=unused-argument
self.logger.warning("Method not implemented. Define a custom on_review on your plugin")
pass # pylint: disable=unnecessary-pass
on_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 |
Label |
label submitted to Kili |
required |
asset_id |
str |
id of the asset on which the label was submitted |
required |
Example use:
>>> def on_review(self, label: Label, asset_id: str):
>>> json_response = label.get('jsonResponse')
>>> if label_is_respecting_business_rule(json_response):
>>> return
>>> else:
>>> self.kili.send_back_to_queue(asset_ids=[asset_id])
Source code in kili/services/plugins/model.py
def on_review(
self,
label: Label,
asset_id: str,
) -> None:
"""
Handler for the review action, triggered when a default label is reviewed on Kili
Args:
label: label submitted to Kili
asset_id: id of the asset on which the label was submitted
Example use:
>>> def on_review(self, label: Label, asset_id: str):
>>> json_response = label.get('jsonResponse')
>>> if label_is_respecting_business_rule(json_response):
>>> return
>>> else:
>>> self.kili.send_back_to_queue(asset_ids=[asset_id])
"""
# pylint: disable=unused-argument
self.logger.warning("Method not implemented. Define a custom on_review on your plugin")
pass # pylint: disable=unnecessary-pass
on_submit(self, label, asset_id)
Handler for the submit action, triggered when a default label is submitted into Kili.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
label |
Label |
label submitted to Kili |
required |
asset_id |
str |
id of the asset on which the label was submitted |
required |
Example use:
>>> def on_submit(self, label: Label, asset_id: str):
>>> json_response = label.get('jsonResponse')
>>> if label_is_respecting_business_rule(json_response):
>>> return
>>> else:
>>> self.kili.send_back_to_queue(asset_ids=[asset_id])
Source code in kili/services/plugins/model.py
def on_submit(
self,
label: Label,
asset_id: str,
) -> None:
"""
Handler for the submit action, triggered when a default label is submitted into Kili.
Args:
label: label submitted to Kili
asset_id: id of the asset on which the label was submitted
Example use:
>>> def on_submit(self, label: Label, asset_id: str):
>>> json_response = label.get('jsonResponse')
>>> if label_is_respecting_business_rule(json_response):
>>> return
>>> else:
>>> self.kili.send_back_to_queue(asset_ids=[asset_id])
"""
# pylint: disable=unused-argument
self.logger.warning("Method not implemented. Define a custom on_submit on your plugin")
pass # pylint: disable=unnecessary-pass
Queries
Set of Plugins queries.
Source code in kili/queries/plugins/__init__.py
class QueriesPlugins:
"""Set of Plugins queries."""
# pylint: disable=too-many-arguments,too-many-locals
def __init__(self, auth: KiliAuth):
"""Initialize the subclass.
Args:
auth: KiliAuth object
"""
self.auth = auth
@typechecked
def get_plugin_logs(
self,
project_id: str,
plugin_name: str,
start_date: Optional[datetime] = None,
limit: Optional[int] = None,
skip: Optional[int] = None,
):
# pylint: disable=line-too-long
"""Get paginated logs of a plugin on a project.
Args:
project_id: Identifier of the project
plugin_name: Name of the plugin
start_date: Datetime used to get the logs from, if not provided, it will be the plugin's creation date
limit: Limit for pagination, if not provided, it will be 100
skip: Skip for pagination, if not provided, it will be 0
Returns:
A result array which contains the logs of the plugin,
or an error message.
Examples:
>>> kili.get_plugin_logs(project_id="my_project_id", plugin_name="my_plugin_name", start_date="1970/01/01")
"""
where = PluginLogsWhere(
project_id=project_id, plugin_name=plugin_name, start_date=start_date
)
options = QueryOptions(
first=limit, skip=skip or 0, disable_tqdm=False
) # disable tqm is not implemnted for this query
pretty_result = PluginQuery(self.auth.client).get_logs(where, options)
return json.dumps(pretty_result, sort_keys=True, indent=4)
@typechecked
def get_plugin_status(
self,
plugin_name: str,
verbose: bool = True,
):
"""Update a plugin with new code.
Args:
plugin_name: Name of the plugin
verbose: If false, minimal logs are displayed
Returns:
The status of the plugin if query was successful or an error message otherwise.
Examples:
>>> kili.get_plugin_status(plugin_name="my_plugin_name")
"""
result = PluginUploader(self.auth, "", plugin_name, verbose).get_plugin_runner_status()
return result
# pylint: disable=dangerous-default-value
@typechecked
def list_plugins(
self,
fields: List[str] = [
"name",
"projectIds",
"id",
"createdAt",
"updatedAt",
],
):
# pylint: disable=line-too-long
"""List all plugins from your organization
Args:
fields: All the fields to request among the possible fields for the plugins
See [the documentation](https://docs.kili-technology.com/reference/graphql-api#plugins) for all possible fields.
Returns:
A result array which contains all the plugins from your organization,
or an error message.
Examples:
>>> kili.list_plugins()
>>> kili.list_plugins(fields=['name'])
"""
return PluginQuery(self.auth.client).list(fields=fields)
get_plugin_logs(self, project_id, plugin_name, start_date=None, limit=None, skip=None)
Get paginated logs of a plugin on a project.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project_id |
str |
Identifier of the project |
required |
plugin_name |
str |
Name of the plugin |
required |
start_date |
Optional[datetime.datetime] |
Datetime used to get the logs from, if not provided, it will be the plugin's creation date |
None |
limit |
Optional[int] |
Limit for pagination, if not provided, it will be 100 |
None |
skip |
Optional[int] |
Skip for pagination, if not provided, it will be 0 |
None |
Returns:
Type | Description |
---|---|
A result array which contains the logs of the plugin, or an error message. |
Examples:
>>> kili.get_plugin_logs(project_id="my_project_id", plugin_name="my_plugin_name", start_date="1970/01/01")
Source code in kili/queries/plugins/__init__.py
@typechecked
def get_plugin_logs(
self,
project_id: str,
plugin_name: str,
start_date: Optional[datetime] = None,
limit: Optional[int] = None,
skip: Optional[int] = None,
):
# pylint: disable=line-too-long
"""Get paginated logs of a plugin on a project.
Args:
project_id: Identifier of the project
plugin_name: Name of the plugin
start_date: Datetime used to get the logs from, if not provided, it will be the plugin's creation date
limit: Limit for pagination, if not provided, it will be 100
skip: Skip for pagination, if not provided, it will be 0
Returns:
A result array which contains the logs of the plugin,
or an error message.
Examples:
>>> kili.get_plugin_logs(project_id="my_project_id", plugin_name="my_plugin_name", start_date="1970/01/01")
"""
where = PluginLogsWhere(
project_id=project_id, plugin_name=plugin_name, start_date=start_date
)
options = QueryOptions(
first=limit, skip=skip or 0, disable_tqdm=False
) # disable tqm is not implemnted for this query
pretty_result = PluginQuery(self.auth.client).get_logs(where, options)
return json.dumps(pretty_result, sort_keys=True, indent=4)
get_plugin_status(self, plugin_name, verbose=True)
Update a plugin with new code.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
plugin_name |
str |
Name of the plugin |
required |
verbose |
bool |
If false, minimal logs are displayed |
True |
Returns:
Type | Description |
---|---|
The status of the plugin if query was successful or an error message otherwise. |
Examples:
>>> kili.get_plugin_status(plugin_name="my_plugin_name")
Source code in kili/queries/plugins/__init__.py
@typechecked
def get_plugin_status(
self,
plugin_name: str,
verbose: bool = True,
):
"""Update a plugin with new code.
Args:
plugin_name: Name of the plugin
verbose: If false, minimal logs are displayed
Returns:
The status of the plugin if query was successful or an error message otherwise.
Examples:
>>> kili.get_plugin_status(plugin_name="my_plugin_name")
"""
result = PluginUploader(self.auth, "", plugin_name, verbose).get_plugin_runner_status()
return result
list_plugins(self, fields=['name', 'projectIds', 'id', 'createdAt', 'updatedAt'])
List all plugins from your organization
Parameters:
Name | Type | Description | Default |
---|---|---|---|
fields |
List[str] |
All the fields to request among the possible fields for the plugins See the documentation for all possible fields. |
['name', 'projectIds', 'id', 'createdAt', 'updatedAt'] |
Returns:
Type | Description |
---|---|
A result array which contains all the plugins from your organization, or an error message. |
Examples:
>>> kili.list_plugins()
>>> kili.list_plugins(fields=['name'])
Source code in kili/queries/plugins/__init__.py
@typechecked
def list_plugins(
self,
fields: List[str] = [
"name",
"projectIds",
"id",
"createdAt",
"updatedAt",
],
):
# pylint: disable=line-too-long
"""List all plugins from your organization
Args:
fields: All the fields to request among the possible fields for the plugins
See [the documentation](https://docs.kili-technology.com/reference/graphql-api#plugins) for all possible fields.
Returns:
A result array which contains all the plugins from your organization,
or an error message.
Examples:
>>> kili.list_plugins()
>>> kili.list_plugins(fields=['name'])
"""
return PluginQuery(self.auth.client).list(fields=fields)
Mutations
Set of Plugins mutations.
Source code in kili/mutations/plugins/__init__.py
class MutationsPlugins:
"""Set of Plugins mutations."""
# pylint: disable=too-many-arguments,too-many-locals
def __init__(self, auth: KiliAuth):
"""Initialize the subclass.
Args:
auth: KiliAuth object
"""
self.auth = auth
@typechecked
def upload_plugin(
self,
plugin_path: Optional[str] = None,
plugin_name: Optional[str] = None,
verbose: bool = True,
**kwargs
):
# pylint: disable=line-too-long
"""Uploads a plugin.
Args:
plugin_path : Path to your plugin. Either a folder containing a main.py (mandatory) and a requirements.txt (optional) or a .py file
plugin_name: name of your plugin, if not provided, it will be the name from your file
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.upload_plugin(plugin_path="./path/to/my/folder")
>>> kili.upload_plugin(plugin_path="./path/to/my/file.py")
"""
if kwargs.get("file_path"):
raise TypeError(
'"file_path" has been deprecated for "plugin_path", please use "plugin_path" instead'
)
if not plugin_path:
raise TypeError('"plugin_path is nullish, please provide a value')
return PluginUploader(self.auth, plugin_path, plugin_name, verbose).create_plugin()
@typechecked
def create_webhook(
self,
webhook_url: str,
plugin_name: str,
header: Optional[str] = None,
verbose: bool = True,
):
# pylint: disable=line-too-long
"""Create a webhook linked to Kili's events.
Args:
webhook_url: URL receiving post requests on events on Kili
plugin_name: name of your plugin
header: Authorization header to access the routes
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.create_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
"""
return WebhookUploader(
self.auth, webhook_url, plugin_name, header, verbose
).create_webhook()
@typechecked
def update_webhook(
self,
new_webhook_url: str,
plugin_name: str,
new_header: Optional[str] = None,
verbose: bool = True,
):
# pylint: disable=line-too-long
"""Update a webhook linked to Kili's events.
Args:
new_webhook_url: New URL receiving post requests on events on Kili
plugin_name: name of your plugin
new_header: Authorization header to access the routes
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.update_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
"""
return WebhookUploader(
self.auth, new_webhook_url, plugin_name, new_header, verbose
).update_webhook()
@typechecked
def activate_plugin_on_project(
self,
plugin_name: str,
project_id: str,
):
# pylint: disable=line-too-long
"""Activates a plugin on a project.
Args:
plugin_name: Name of the plugin
project_id: Identifier of the project
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.activate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
"""
return activate_plugin(self.auth, plugin_name, project_id)
@typechecked
def deactivate_plugin_on_project(
self,
plugin_name: str,
project_id: str,
):
# pylint: disable=line-too-long
"""Activates a plugin on a project.
Args:
plugin_name: Name of the plugin
project_id: Identifier of the project
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.deactivate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
"""
return deactivate_plugin(self.auth, plugin_name, project_id)
@typechecked
def delete_plugin(
self,
plugin_name: str,
):
# pylint: disable=line-too-long
"""Deletes a plugin.
Args:
plugin_name: Name of the plugin
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.delete_plugin(plugin_name="my_plugin_name")
"""
return delete_plugin(self.auth, plugin_name)
@typechecked
def update_plugin(
self,
plugin_path: Optional[str] = None,
plugin_name: Optional[str] = None,
verbose: bool = True,
**kwargs
):
"""Update a plugin with new code.
Args:
plugin_path : Path to your plugin. Either:
- a folder containing a main.py (mandatory) and a requirements.txt (optional)
- a .py file
plugin_name: Name of the plugin
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.update_plugin(plugin_name="my_plugin_name")
"""
if kwargs.get("file_path"):
raise TypeError(
""" "file_path" has been deprecated for "plugin_path",
please use "plugin_path" instead"""
)
if not plugin_path:
raise TypeError('"plugin_path is nullish, please provide a value')
if not plugin_name:
raise TypeError('"plugin_name is nullish, please provide a value')
return PluginUploader(self.auth, plugin_path, plugin_name, verbose).update_plugin()
activate_plugin_on_project(self, plugin_name, project_id)
Activates a plugin on a project.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
plugin_name |
str |
Name of the plugin |
required |
project_id |
str |
Identifier of the project |
required |
verbose |
If false, minimal logs are displayed |
required |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.activate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
Source code in kili/mutations/plugins/__init__.py
@typechecked
def activate_plugin_on_project(
self,
plugin_name: str,
project_id: str,
):
# pylint: disable=line-too-long
"""Activates a plugin on a project.
Args:
plugin_name: Name of the plugin
project_id: Identifier of the project
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.activate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
"""
return activate_plugin(self.auth, plugin_name, project_id)
create_webhook(self, webhook_url, plugin_name, header=None, verbose=True)
Create a webhook linked to Kili's events.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
webhook_url |
str |
URL receiving post requests on events on Kili |
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 |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.create_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
Source code in kili/mutations/plugins/__init__.py
@typechecked
def create_webhook(
self,
webhook_url: str,
plugin_name: str,
header: Optional[str] = None,
verbose: bool = True,
):
# pylint: disable=line-too-long
"""Create a webhook linked to Kili's events.
Args:
webhook_url: URL receiving post requests on events on Kili
plugin_name: name of your plugin
header: Authorization header to access the routes
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.create_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
"""
return WebhookUploader(
self.auth, webhook_url, plugin_name, header, verbose
).create_webhook()
deactivate_plugin_on_project(self, plugin_name, project_id)
Activates a plugin on a project.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
plugin_name |
str |
Name of the plugin |
required |
project_id |
str |
Identifier of the project |
required |
verbose |
If false, minimal logs are displayed |
required |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.deactivate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
Source code in kili/mutations/plugins/__init__.py
@typechecked
def deactivate_plugin_on_project(
self,
plugin_name: str,
project_id: str,
):
# pylint: disable=line-too-long
"""Activates a plugin on a project.
Args:
plugin_name: Name of the plugin
project_id: Identifier of the project
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.deactivate_plugin_on_project(plugin_name="my_plugin_name", project_id="my_project_id")
"""
return deactivate_plugin(self.auth, plugin_name, project_id)
delete_plugin(self, plugin_name)
Deletes a plugin.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
plugin_name |
str |
Name of the plugin |
required |
verbose |
If false, minimal logs are displayed |
required |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.delete_plugin(plugin_name="my_plugin_name")
Source code in kili/mutations/plugins/__init__.py
@typechecked
def delete_plugin(
self,
plugin_name: str,
):
# pylint: disable=line-too-long
"""Deletes a plugin.
Args:
plugin_name: Name of the plugin
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.delete_plugin(plugin_name="my_plugin_name")
"""
return delete_plugin(self.auth, plugin_name)
update_plugin(self, plugin_path=None, plugin_name=None, verbose=True, **kwargs)
Update a plugin with new code.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
plugin_path |
Path to your plugin. Either: - a folder containing a main.py (mandatory) and a requirements.txt (optional) - a .py file |
None |
|
plugin_name |
Optional[str] |
Name of the plugin |
None |
verbose |
bool |
If false, minimal logs are displayed |
True |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.update_plugin(plugin_name="my_plugin_name")
Source code in kili/mutations/plugins/__init__.py
@typechecked
def update_plugin(
self,
plugin_path: Optional[str] = None,
plugin_name: Optional[str] = None,
verbose: bool = True,
**kwargs
):
"""Update a plugin with new code.
Args:
plugin_path : Path to your plugin. Either:
- a folder containing a main.py (mandatory) and a requirements.txt (optional)
- a .py file
plugin_name: Name of the plugin
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.update_plugin(plugin_name="my_plugin_name")
"""
if kwargs.get("file_path"):
raise TypeError(
""" "file_path" has been deprecated for "plugin_path",
please use "plugin_path" instead"""
)
if not plugin_path:
raise TypeError('"plugin_path is nullish, please provide a value')
if not plugin_name:
raise TypeError('"plugin_name is nullish, please provide a value')
return PluginUploader(self.auth, plugin_path, plugin_name, verbose).update_plugin()
update_webhook(self, new_webhook_url, plugin_name, new_header=None, verbose=True)
Update a webhook linked to Kili's events.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
new_webhook_url |
str |
New URL receiving post requests on events on Kili |
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 |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.update_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
Source code in kili/mutations/plugins/__init__.py
@typechecked
def update_webhook(
self,
new_webhook_url: str,
plugin_name: str,
new_header: Optional[str] = None,
verbose: bool = True,
):
# pylint: disable=line-too-long
"""Update a webhook linked to Kili's events.
Args:
new_webhook_url: New URL receiving post requests on events on Kili
plugin_name: name of your plugin
new_header: Authorization header to access the routes
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.update_webhook(webhook_url='https://my-custom-url-publicly-accessible/', plugin_name='my webhook', header='...')
"""
return WebhookUploader(
self.auth, new_webhook_url, plugin_name, new_header, verbose
).update_webhook()
upload_plugin(self, plugin_path=None, plugin_name=None, verbose=True, **kwargs)
Uploads a plugin.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
plugin_path |
Path to your plugin. Either a folder containing a main.py (mandatory) and a requirements.txt (optional) or a .py file |
None |
|
plugin_name |
Optional[str] |
name of your plugin, if not provided, it will be the name from your file |
None |
verbose |
bool |
If false, minimal logs are displayed |
True |
Returns:
Type | Description |
---|---|
A result object which indicates if the mutation was successful, or an error message. |
Examples:
>>> kili.upload_plugin(plugin_path="./path/to/my/folder")
>>> kili.upload_plugin(plugin_path="./path/to/my/file.py")
Source code in kili/mutations/plugins/__init__.py
@typechecked
def upload_plugin(
self,
plugin_path: Optional[str] = None,
plugin_name: Optional[str] = None,
verbose: bool = True,
**kwargs
):
# pylint: disable=line-too-long
"""Uploads a plugin.
Args:
plugin_path : Path to your plugin. Either a folder containing a main.py (mandatory) and a requirements.txt (optional) or a .py file
plugin_name: name of your plugin, if not provided, it will be the name from your file
verbose: If false, minimal logs are displayed
Returns:
A result object which indicates if the mutation was successful,
or an error message.
Examples:
>>> kili.upload_plugin(plugin_path="./path/to/my/folder")
>>> kili.upload_plugin(plugin_path="./path/to/my/file.py")
"""
if kwargs.get("file_path"):
raise TypeError(
'"file_path" has been deprecated for "plugin_path", please use "plugin_path" instead'
)
if not plugin_path:
raise TypeError('"plugin_path is nullish, please provide a value')
return PluginUploader(self.auth, plugin_path, plugin_name, verbose).create_plugin()