Project Workflow module
Client presentation methods for project workflow.
Source code in kili/presentation/client/project_workflow.py
class ProjectWorkflowClientMethods(BaseClientMethods):
"""Client presentation methods for project workflow."""
@typechecked
def update_project_workflow(
self,
project_id: str,
enforce_step_separation: Optional[bool] = None,
create_steps: Optional[list[WorkflowStepCreate]] = None,
update_steps: Optional[list[WorkflowStepUpdate]] = None,
delete_steps: Optional[list[str]] = None,
) -> dict[str, Any]:
"""Update properties of a project workflow.
Args:
project_id: Id of the project.
enforce_step_separation: Prevents the same user from being assigned to
multiple steps in the workflow for a same asset,
ensuring independent review and labeling processes
create_steps: List of steps to create in the project workflow.
update_steps: List of steps to update in the project workflow.
delete_steps: List of step IDs or names to delete from the project workflow.
Returns:
A dict with the changed properties which indicates if the mutation was successful,
else an error message.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).update_project_workflow(
project_id=ProjectId(project_id),
enforce_step_separation=enforce_step_separation,
create_steps=create_steps,
update_steps=update_steps,
delete_steps=delete_steps,
)
@typechecked
def get_steps(
self,
project_id: str,
fields: ListOrTuple[str] = (
"steps.type",
"steps.name",
"steps.id",
"steps.assignees.email",
"steps.assignees.id",
),
) -> list[dict[str, Any]]:
"""Get steps in a project workflow.
Args:
project_id: Id of the project.
fields: All the fields to request among the possible fields for the project.
See the documentation for all possible fields.
Returns:
A dict with the steps of the project workflow.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).get_steps(
project_id=ProjectId(project_id), fields=fields
)
@typechecked
def add_reviewers_to_step(
self, project_id: str, step_name: str, emails: list[str]
) -> list[str]:
"""Add reviewers to a specific step.
Args:
project_id: Id of the project.
step_name: Name of the step.
emails: List of emails to add.
Returns:
A list with the added emails.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).add_reviewers_to_step(
project_id=project_id, step_name=step_name, emails=emails
)
@typechecked
def remove_reviewers_from_step(
self, project_id: str, step_name: str, emails: list[str]
) -> list[str]:
"""Remove reviewers from a specific step.
Args:
project_id: Id of the project.
step_name: Name of the step.
emails: List of emails to remove.
Returns:
A list with the removed emails.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).remove_reviewers_from_step(
project_id=project_id, step_name=step_name, emails=emails
)
@typechecked
def copy_workflow_from_project(
self,
destination_project_id: str,
source_project_id: str,
) -> dict[str, Any]:
"""Copy the workflow from a source project to a destination project.
Copies all workflow steps with their configurations (consensus, coverage,
sendBackStepId) from the source project. Assignees are not copied.
The destination project must have no labels. Existing workflow steps in the
destination project will be deleted and replaced by the source workflow.
Args:
destination_project_id: Id of the destination project to copy the workflow to.
source_project_id: Id of the source project to copy the workflow from.
Returns:
A dict with the workflow data which indicates if the mutation was successful,
else an error message.
Raises:
ValueError: If the source project has no workflow steps, or if the
destination project already has labels.
Examples:
>>> kili.copy_workflow_from_project(
... destination_project_id="destination_project_id",
... source_project_id="source_project_id",
... )
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).copy_workflow_from_project(
source_project_id=ProjectId(source_project_id),
destination_project_id=ProjectId(destination_project_id),
)
add_reviewers_to_step(self, project_id, step_name, emails)
Add reviewers to a specific step.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_id |
str |
Id of the project. |
required |
step_name |
str |
Name of the step. |
required |
emails |
list |
List of emails to add. |
required |
Returns:
| Type | Description |
|---|---|
list |
A list with the added emails. |
Source code in kili/presentation/client/project_workflow.py
def add_reviewers_to_step(
self, project_id: str, step_name: str, emails: list[str]
) -> list[str]:
"""Add reviewers to a specific step.
Args:
project_id: Id of the project.
step_name: Name of the step.
emails: List of emails to add.
Returns:
A list with the added emails.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).add_reviewers_to_step(
project_id=project_id, step_name=step_name, emails=emails
)
copy_workflow_from_project(self, destination_project_id, source_project_id)
Copy the workflow from a source project to a destination project.
Copies all workflow steps with their configurations (consensus, coverage, sendBackStepId) from the source project. Assignees are not copied.
The destination project must have no labels. Existing workflow steps in the destination project will be deleted and replaced by the source workflow.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
destination_project_id |
str |
Id of the destination project to copy the workflow to. |
required |
source_project_id |
str |
Id of the source project to copy the workflow from. |
required |
Returns:
| Type | Description |
|---|---|
dict |
A dict with the workflow data which indicates if the mutation was successful, else an error message. |
Exceptions:
| Type | Description |
|---|---|
ValueError |
If the source project has no workflow steps, or if the destination project already has labels. |
Examples:
>>> kili.copy_workflow_from_project(
... destination_project_id="destination_project_id",
... source_project_id="source_project_id",
... )
Source code in kili/presentation/client/project_workflow.py
def copy_workflow_from_project(
self,
destination_project_id: str,
source_project_id: str,
) -> dict[str, Any]:
"""Copy the workflow from a source project to a destination project.
Copies all workflow steps with their configurations (consensus, coverage,
sendBackStepId) from the source project. Assignees are not copied.
The destination project must have no labels. Existing workflow steps in the
destination project will be deleted and replaced by the source workflow.
Args:
destination_project_id: Id of the destination project to copy the workflow to.
source_project_id: Id of the source project to copy the workflow from.
Returns:
A dict with the workflow data which indicates if the mutation was successful,
else an error message.
Raises:
ValueError: If the source project has no workflow steps, or if the
destination project already has labels.
Examples:
>>> kili.copy_workflow_from_project(
... destination_project_id="destination_project_id",
... source_project_id="source_project_id",
... )
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).copy_workflow_from_project(
source_project_id=ProjectId(source_project_id),
destination_project_id=ProjectId(destination_project_id),
)
get_steps(self, project_id, fields=('steps.type', 'steps.name', 'steps.id', 'steps.assignees.email', 'steps.assignees.id'))
Get steps in a project workflow.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_id |
str |
Id of the project. |
required |
fields |
Union[list[str], tuple[str, ...]] |
All the fields to request among the possible fields for the project. See the documentation for all possible fields. |
('steps.type', 'steps.name', 'steps.id', 'steps.assignees.email', 'steps.assignees.id') |
Returns:
| Type | Description |
|---|---|
list |
A dict with the steps of the project workflow. |
Source code in kili/presentation/client/project_workflow.py
def get_steps(
self,
project_id: str,
fields: ListOrTuple[str] = (
"steps.type",
"steps.name",
"steps.id",
"steps.assignees.email",
"steps.assignees.id",
),
) -> list[dict[str, Any]]:
"""Get steps in a project workflow.
Args:
project_id: Id of the project.
fields: All the fields to request among the possible fields for the project.
See the documentation for all possible fields.
Returns:
A dict with the steps of the project workflow.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).get_steps(
project_id=ProjectId(project_id), fields=fields
)
remove_reviewers_from_step(self, project_id, step_name, emails)
Remove reviewers from a specific step.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_id |
str |
Id of the project. |
required |
step_name |
str |
Name of the step. |
required |
emails |
list |
List of emails to remove. |
required |
Returns:
| Type | Description |
|---|---|
list |
A list with the removed emails. |
Source code in kili/presentation/client/project_workflow.py
def remove_reviewers_from_step(
self, project_id: str, step_name: str, emails: list[str]
) -> list[str]:
"""Remove reviewers from a specific step.
Args:
project_id: Id of the project.
step_name: Name of the step.
emails: List of emails to remove.
Returns:
A list with the removed emails.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).remove_reviewers_from_step(
project_id=project_id, step_name=step_name, emails=emails
)
update_project_workflow(self, project_id, enforce_step_separation=None, create_steps=None, update_steps=None, delete_steps=None)
Update properties of a project workflow.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_id |
str |
Id of the project. |
required |
enforce_step_separation |
Optional[bool] |
Prevents the same user from being assigned to multiple steps in the workflow for a same asset, ensuring independent review and labeling processes |
None |
create_steps |
Optional[list[kili.domain.project.WorkflowStepCreate]] |
List of steps to create in the project workflow. |
None |
update_steps |
Optional[list[kili.domain.project.WorkflowStepUpdate]] |
List of steps to update in the project workflow. |
None |
delete_steps |
Optional[list[str]] |
List of step IDs or names to delete from the project workflow. |
None |
Returns:
| Type | Description |
|---|---|
dict |
A dict with the changed properties which indicates if the mutation was successful, else an error message. |
Source code in kili/presentation/client/project_workflow.py
def update_project_workflow(
self,
project_id: str,
enforce_step_separation: Optional[bool] = None,
create_steps: Optional[list[WorkflowStepCreate]] = None,
update_steps: Optional[list[WorkflowStepUpdate]] = None,
delete_steps: Optional[list[str]] = None,
) -> dict[str, Any]:
"""Update properties of a project workflow.
Args:
project_id: Id of the project.
enforce_step_separation: Prevents the same user from being assigned to
multiple steps in the workflow for a same asset,
ensuring independent review and labeling processes
create_steps: List of steps to create in the project workflow.
update_steps: List of steps to update in the project workflow.
delete_steps: List of step IDs or names to delete from the project workflow.
Returns:
A dict with the changed properties which indicates if the mutation was successful,
else an error message.
"""
return ProjectWorkflowUseCases(self.kili_api_gateway).update_project_workflow(
project_id=ProjectId(project_id),
enforce_step_separation=enforce_step_separation,
create_steps=create_steps,
update_steps=update_steps,
delete_steps=delete_steps,
)