from __future__ import annotations
import json
import string
import requests
from gazu.exception import (
TaskStatusNotFoundException,
TaskMustBeADictException,
)
from . import client as raw
from .client import KitsuClient
from .sorting import sort_by_name
from .helpers import (
download_file,
normalize_model_parameter,
normalize_list_of_models_for_links,
)
from .cache import cache
default = raw.default_client
[docs]
@cache
def all_task_statuses(client: KitsuClient = default) -> list[dict]:
"""
Returns:
list: Task statuses stored in database.
"""
task_statuses = raw.fetch_all("task-status", client=client)
return sort_by_name(task_statuses)
[docs]
@cache
def all_task_types(client: KitsuClient = default) -> list[dict]:
"""
Returns:
list: Task types stored in database.
"""
task_types = raw.fetch_all("task-types", client=client)
return sort_by_name(task_types)
[docs]
@cache
def all_task_types_for_project(
project: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Returns:
list: Task types stored in database.
"""
project = normalize_model_parameter(project)
task_types = raw.fetch_all(
"projects/%s/task-types" % project["id"], client=client
)
return sort_by_name(task_types)
[docs]
@cache
def all_task_statuses_for_project(
project: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Returns:
list: Task status stored in database.
"""
project = normalize_model_parameter(project)
task_statuses = raw.fetch_all(
"projects/%s/settings/task-status" % project["id"], client=client
)
return sort_by_name(task_statuses)
[docs]
@cache
def all_tasks_for_shot(
shot: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Args:
shot (str / dict): The shot dict or the shot ID.
Returns:
list: Tasks linked to given shot.
"""
shot = normalize_model_parameter(shot)
params = {}
if relations:
params = {"relations": True}
tasks = raw.fetch_all("shots/%s/tasks" % shot["id"], params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_concept(
concept: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Args:
concept (str / dict): The concept dict or the concept ID.
Returns:
list: Tasks linked to given concept.
"""
concept = normalize_model_parameter(concept)
params = {}
if relations:
params = {"relations": True}
tasks = raw.fetch_all(
"concepts/%s/tasks" % concept["id"], params, client=client
)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_edit(
edit: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Args:
edit (str / dict): The edit dict or the edit ID.
Returns:
list: Tasks linked to given edit.
"""
edit = normalize_model_parameter(edit)
params = {}
if relations:
params = {"relations": True}
tasks = raw.fetch_all("edits/%s/tasks" % edit["id"], params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_sequence(
sequence: str | dict,
relations: bool = False,
client: KitsuClient = default,
) -> list[dict]:
"""
Args:
sequence (str / dict): The sequence dict or the sequence ID.
Returns
list: Tasks linked to given sequence.
"""
sequence = normalize_model_parameter(sequence)
params = {}
if relations:
params = {"relations": True}
path = "sequences/%s/tasks" % sequence["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_scene(
scene: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Args:
sequence (str / dict): The scene dict or the scene ID.
Returns:
list: Tasks linked to given scene.
"""
scene = normalize_model_parameter(scene)
params = {}
if relations:
params = {"relations": True}
path = "scenes/%s/tasks" % scene["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_asset(
asset: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Args:
asset (str / dict): The asset dict or the asset ID.
Returns:
list: Tasks directly linked to given asset.
"""
asset = normalize_model_parameter(asset)
params = {}
if relations:
params = {"relations": True}
path = "assets/%s/tasks" % asset["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_episode(
episode: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Retrieve all tasks directly linked to given episode.
"""
episode = normalize_model_parameter(episode)
params = {}
if relations:
params = {"relations": True}
path = "episodes/%s/tasks" % episode["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_shot_tasks_for_sequence(
sequence: str | dict,
relations: bool = False,
client: KitsuClient = default,
) -> list[dict]:
"""
Retrieve all tasks directly linked to all shots of given sequence.
"""
sequence = normalize_model_parameter(sequence)
params = {}
if relations:
params = {"relations": True}
path = "sequences/%s/shot-tasks" % sequence["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_shot_tasks_for_episode(
episode: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Retrieve all tasks directly linked to all shots of given episode.
"""
episode = normalize_model_parameter(episode)
params = {}
if relations:
params = {"relations": True}
path = "episodes/%s/shot-tasks" % episode["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_assets_tasks_for_episode(
episode: str | dict, relations: bool = False, client: KitsuClient = default
) -> list[dict]:
"""
Retrieve all tasks directly linked to all assets of given episode.
"""
episode = normalize_model_parameter(episode)
params = {}
if relations:
params = {"relations": True}
path = "episodes/%s/asset-tasks" % episode["id"]
tasks = raw.fetch_all(path, params, client=client)
return sort_by_name(tasks)
[docs]
@cache
def all_tasks_for_task_status(
project: str | dict,
task_type: str | dict,
task_status: str | dict,
client: KitsuClient = default,
) -> list[dict]:
"""
Args:
project (str / dict): The project dict or the project ID.
task_type (str / dict): The task type dict or ID.
task_status (str / dict): The task status dict or ID.
Returns:
list: Tasks set at given status for given project and task type.
"""
project = normalize_model_parameter(project)
task_type = normalize_model_parameter(task_type)
task_status = normalize_model_parameter(task_status)
return raw.fetch_all(
"tasks",
{
"project_id": project["id"],
"task_type_id": task_type["id"],
"task_status_id": task_status["id"],
},
client=client,
)
[docs]
@cache
def all_tasks_for_task_type(
project: str | dict,
task_type: str | dict,
episode: str | dict | None = None,
client: KitsuClient = default,
) -> list[dict]:
"""
Args:
project (str / dict): The project dict or the project ID.
task_type (str / dict): The task type dict or ID.
episode_id (str / dict): The episode dict or ID.
Returns:
list: Tasks for given project and task type.
"""
project = normalize_model_parameter(project)
task_type = normalize_model_parameter(task_type)
params = {
"project_id": project["id"],
"task_type_id": task_type["id"],
}
if episode is not None:
episode = normalize_model_parameter(episode)
params["episode_id"] = episode["id"]
return raw.fetch_all("tasks", params, client=client)
[docs]
@cache
def all_task_types_for_shot(
shot: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Args:
shot (str / dict): The shot dict or the shot ID.
Returns
list: Task types of task linked to given shot.
"""
shot = normalize_model_parameter(shot)
path = "shots/%s/task-types" % shot["id"]
task_types = raw.fetch_all(path, client=client)
return sort_by_name(task_types)
[docs]
@cache
def all_task_types_for_concept(
concept: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Args:
concept (str / dict): The concept dict or the concept ID.
Returns
list: Task types of task linked to given concept.
"""
concept = normalize_model_parameter(concept)
path = "concepts/%s/task-types" % concept["id"]
task_types = raw.fetch_all(path, client=client)
return sort_by_name(task_types)
[docs]
@cache
def all_task_types_for_asset(
asset: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Args:
asset (str / dict): The asset dict or the asset ID.
Returns:
list: Task types of tasks related to given asset.
"""
asset = normalize_model_parameter(asset)
task_types = raw.fetch_all(
"assets/%s/task-types" % asset["id"], client=client
)
return sort_by_name(task_types)
[docs]
@cache
def all_task_types_for_scene(
scene: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Args:
scene (str / dict): The scene dict or the scene ID.
Returns:
list: Task types of tasks linked to given scene.
"""
scene = normalize_model_parameter(scene)
path = "scenes/%s/task-types" % scene["id"]
task_types = raw.fetch_all(path, client=client)
return sort_by_name(task_types)
[docs]
@cache
def all_task_types_for_sequence(
sequence: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Args:
sequence (str / dict): The sequence dict or the sequence ID.
Returns:
list: Task types of tasks linked directly to given sequence.
"""
sequence = normalize_model_parameter(sequence)
path = "sequences/%s/task-types" % sequence["id"]
task_types = raw.fetch_all(path, client=client)
return sort_by_name(task_types)
[docs]
@cache
def all_task_types_for_episode(
episode: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Returns:
list: Task types of tasks linked directly to given episode.
"""
episode = normalize_model_parameter(episode)
path = "episodes/%s/task-types" % episode["id"]
task_types = raw.fetch_all(path, client=client)
return sort_by_name(task_types)
[docs]
@cache
def all_tasks_for_entity_and_task_type(
entity: str | dict, task_type: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Args:
entity (str / dict): The entity dict or the entity ID.
task_type (str / dict): The task type dict or ID.
Returns:
list: Tasks for given entity or task type.
"""
entity = normalize_model_parameter(entity)
task_type = normalize_model_parameter(task_type)
task_type_id = task_type["id"]
entity_id = entity["id"]
path = "entities/%s/task-types/%s/tasks" % (entity_id, task_type_id)
return raw.fetch_all(path, client=client)
[docs]
@cache
def all_tasks_for_person(
person: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Returns:
list: Tasks that are not done for given person (only for open projects).
"""
person = normalize_model_parameter(person)
return raw.fetch_all("persons/%s/tasks" % person["id"], client=client)
[docs]
@cache
def all_done_tasks_for_person(
person: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Returns:
list: Tasks that are done for given person (only for open projects).
"""
person = normalize_model_parameter(person)
return raw.fetch_all("persons/%s/done-tasks" % person["id"], client=client)
[docs]
@cache
def get_task_by_entity(
entity: str | dict,
task_type: str | dict,
name: str = "main",
client: KitsuClient = default,
) -> dict | None:
"""
Args:
entity (str / dict): The entity dict or the entity ID.
task_type (str / dict): The task type dict or ID.
name (str): Name of the task to look for.
Returns:
Task matching given entity, task type and name.
"""
entity = normalize_model_parameter(entity)
task_type = normalize_model_parameter(task_type)
return raw.fetch_first(
"tasks",
{
"name": name,
"task_type_id": task_type["id"],
"entity_id": entity["id"],
},
client=client,
)
[docs]
@cache
def get_task_type(task_type_id: str, client: KitsuClient = default) -> dict:
"""
Args:
task_type_id (str): ID of claimed task type.
Returns:
dict: Task type matching given ID.
"""
return raw.fetch_one("task-types", task_type_id, client=client)
[docs]
@cache
def get_task_type_by_name(
task_type_name: str,
for_entity: str | None = None,
department: str | dict | None = None,
client: KitsuClient = default,
) -> dict | None:
"""
Args:
task_type_name (str): Name of claimed task type.
for_entity (str): The entity type for which the task type is created.
department (str / dict): The department for which the task type is created.
Returns:
dict: Task type object for given name, or None if none found.
"""
params = {"name": task_type_name}
if for_entity is not None:
params["for_entity"] = for_entity
if department is not None:
params["department_id"] = normalize_model_parameter(department)["id"]
return raw.fetch_first("task-types", params, client=client)
[docs]
@cache
def get_task_type_by_short_name(
task_type_short_name: str,
for_entity: str | None = None,
department: str | dict | None = None,
client: KitsuClient = default,
) -> dict | None:
"""
Args:
task_type_short_name (str): Short name of claimed task type.
for_entity (str): The entity type for which the task type is created.
department (str /dict): The department for which the task type is created.
Returns:
dict: Task type object for given name, or None if none found.
"""
params = {"short_name": task_type_short_name}
if for_entity is not None:
params["for_entity"] = for_entity
if department is not None:
params["department_id"] = normalize_model_parameter(department)["id"]
return raw.fetch_first("task-types", params, client=client)
[docs]
@cache
def get_task_by_path(
project: str | dict,
file_path: str,
entity_type: str = "shot",
client: KitsuClient = default,
) -> dict:
"""
Args:
project (str / dict): The project dict or the project ID.
file_path (str): The file path to find a related task.
entity_type (str): asset, shot or scene.
Returns:
dict: A task from given file path. This function requires context:
the project related to the given path and the related entity type.
"""
project = normalize_model_parameter(project)
data = {
"file_path": file_path,
"project_id": project["id"],
"type": entity_type,
}
return raw.post("data/tasks/from-path/", data, client=client)
[docs]
@cache
def get_default_task_status(client: KitsuClient = default) -> dict:
"""
Returns:
dict: The unique task status flagged with `is_default`.
"""
return raw.fetch_first(
"task-status", params={"is_default": True}, client=client
)
[docs]
@cache
def get_task_status(
task_status_id: str, client: KitsuClient = default
) -> dict:
"""
Args:
task_status_id (str): ID of claimed task status.
Returns:
dict: Task status matching given ID.
"""
return raw.fetch_one("task-status", task_status_id, client=client)
[docs]
@cache
def get_task_status_by_name(
name: str, client: KitsuClient = default
) -> dict | None:
"""
Args:
name (str / dict): The name of claimed task status.
Returns:
dict: Task status matching given name.
"""
return raw.fetch_first("task-status", {"name": name}, client=client)
[docs]
@cache
def get_task_status_by_short_name(
task_status_short_name: str, client: KitsuClient = default
) -> dict | None:
"""
Args:
short_name (str / dict): The short name of claimed task status.
Returns:
dict: Task status matching given short name.
"""
return raw.fetch_first(
"task-status", {"short_name": task_status_short_name}, client=client
)
[docs]
def remove_task_type(
task_type: str | dict, client: KitsuClient = default
) -> str:
"""
Remove given task type from database.
Args:
task_type (str / dict): The task type dict or ID.
"""
task_type = normalize_model_parameter(task_type)
return raw.delete(
"data/task-types/%s" % task_type["id"],
{"force": True},
client=client,
)
[docs]
def remove_task_status(
task_status: str | dict, client: KitsuClient = default
) -> str:
"""
Remove given task status from database.
Args:
task_status (str / dict): The task status dict or ID.
"""
task_status = normalize_model_parameter(task_status)
return raw.delete(
"data/task-status/%s" % task_status["id"],
{"force": True},
client=client,
)
[docs]
def update_task_type(task_type: dict, client: KitsuClient = default) -> dict:
"""
Update given task type into the API.
Args:
task_type (dict): The task type dict to update.
Returns:
dict: Updated task type.
"""
return raw.put(
"data/task-types/%s" % task_type["id"],
task_type,
client=client,
)
[docs]
def update_task_status(
task_status: dict, client: KitsuClient = default
) -> dict:
"""
Update given task status into the API.
Args:
task_status (dict): The task status dict to update.
Returns:
dict: Updated task status.
"""
return raw.put(
"data/task-status/%s" % task_status["id"],
task_status,
client=client,
)
[docs]
@cache
def get_task(task_id: str | dict, client: KitsuClient = default) -> dict:
"""
Args:
task_id (str): ID of claimed task.
Returns:
dict: Task matching given ID.
"""
task_id = normalize_model_parameter(task_id)
return raw.get("data/tasks/%s/full" % task_id["id"], client=client)
[docs]
def new_task(
entity: str | dict,
task_type: str | dict,
name: str = "main",
task_status: dict | None = None,
assigner: str | dict | None = None,
assignees: list[str | dict] = [],
client: KitsuClient = default,
) -> dict:
"""
Create a new task for given entity and task type.
Args:
entity (dict): Entity for which task is created.
task_type (dict): Task type of created task.
name (str): Name of the task (default is "main").
task_status (dict): The task status to set (default status is Todo).
assigner (dict): Person who assigns the task.
assignees (list): List of people assigned to the task.
Returns:
Created task.
"""
entity = normalize_model_parameter(entity)
task_type = normalize_model_parameter(task_type)
if task_status is None:
task_status = get_default_task_status(client=client)
data = {
"project_id": entity["project_id"],
"entity_id": entity["id"],
"task_type_id": task_type["id"],
"task_status_id": task_status["id"],
"assignees": normalize_list_of_models_for_links(assignees),
"name": name,
}
if assigner is not None:
data["assigner_id"] = normalize_model_parameter(assigner)["id"]
task = get_task_by_entity(entity, task_type, name, client=client)
if task is None:
task = raw.post("data/tasks", data, client=client)
return task
[docs]
def remove_task(task: str | dict, client: KitsuClient = default) -> str:
"""
Remove given task from database.
Args:
task (str / dict): The task dict or the task ID.
"""
task = normalize_model_parameter(task)
raw.delete("data/tasks/%s" % task["id"], {"force": True}, client=client)
[docs]
def start_task(
task: str | dict,
started_task_status: str | dict | None = None,
person: str | dict | None = None,
client: KitsuClient = default,
) -> dict:
"""
Create a comment to change task status to started_task_status
(by default WIP) and set its real start date to now.
Args:
task (str / dict): The task dict or the task ID.
started_task_status (str / dict): The task status dict or ID.
person (str / dict): The person dict or the person ID.
Returns:
dict: Created comment.
"""
if started_task_status is None:
started_task_status = get_task_status_by_short_name(
"wip", client=client
)
if started_task_status is None:
raise TaskStatusNotFoundException(
(
"started_task_status is None : 'wip' task status is "
"non-existent. You have to create it or to set an other "
"task status for started_task_status in the parameters "
"of the function."
)
)
return add_comment(task, started_task_status, person=person, client=client)
[docs]
def task_to_review(
task: str | dict,
person: str | dict,
comment: str,
revision: int = 1,
change_status: bool = True,
client: KitsuClient = default,
) -> dict:
"""
Deprecated.
Mark given task as pending, waiting for approval. Author is given through
the person argument.
Args:
task (str / dict): The task dict or the task ID.
person (str / dict): The person dict or the person ID.
comment (str): Comment text
revision (int): Force revision of related preview file
change_status (bool): If set to false, the task status is not changed.
Returns:
dict: Modified task
"""
task = normalize_model_parameter(task)
person = normalize_model_parameter(person)
path = "actions/tasks/%s/to-review" % task["id"]
data = {
"person_id": person["id"],
"comment": comment,
"revision": revision,
"change_status": change_status,
}
return raw.put(path, data, client=client)
[docs]
@cache
def get_time_spent(
task: str | dict, date: str | None = None, client: KitsuClient = default
) -> dict:
"""
Get the time spent by CG artists on a task at a given date if given. A field contains
the total time spent. Durations are given in minutes. Date format is
YYYY-MM-DD.
Args:
task (str / dict): The task dict or the task ID.
date (str): The date for which time spent is required.
Returns:
dict: A dict with person ID as key and time spent object as value.
"""
task = normalize_model_parameter(task)
path = "actions/tasks/%s/time-spents" % (task["id"])
if date is not None:
path += "/%s" % (date)
return raw.get(path, client=client)
[docs]
def set_time_spent(
task: str | dict,
person: str | dict,
date: str,
duration: int,
client: KitsuClient = default,
) -> dict:
"""
Set the time spent by a CG artist on a given task at a given date.
Args:
task (str / dict): The task dict or the task ID.
person (str / dict): The person who spent the time on given task.
date (str): The date ("YYYY-MM-DD") for which time spent must be set.
duration (int): The duration (in minutes) of the time spent on given task.
Returns:
dict: Created time spent entry.
"""
task = normalize_model_parameter(task)
person = normalize_model_parameter(person)
path = "actions/tasks/%s/time-spents/%s/persons/%s" % (
task["id"],
date,
person["id"],
)
return raw.post(path, {"duration": duration}, client=client)
[docs]
def add_time_spent(
task: str | dict,
person: str | dict,
date: str,
duration: int,
client: KitsuClient = default,
) -> dict:
"""
Add given duration to the already logged duration for given task and person
at a given date.
Args:
task (str / dict): The task dict or the task ID.
person (str / dict): The person who spent the time on given task.
date (str): The date ("YYYY-MM-DD") for which time spent must be set.
duration (int): The duration (in minutes) of the time spent on given task.
Returns:
dict: Updated time spent entry.
"""
task = normalize_model_parameter(task)
person = normalize_model_parameter(person)
path = "actions/tasks/%s/time-spents/%s/persons/%s/add" % (
task["id"],
date,
person["id"],
)
return raw.post(path, {"duration": duration}, client=client)
[docs]
def create_preview(
task: str | dict,
comment: str | dict,
revision: int | None = None,
client: KitsuClient = default,
) -> dict:
"""
Create a preview into given comment.
Args:
task (str / dict): The task dict or the task ID.
comment (str / dict): The comment or the comment ID.
revision (int): Revision number.
Returns:
dict: Created preview file model.
"""
task = normalize_model_parameter(task)
comment = normalize_model_parameter(comment)
path = "actions/tasks/%s/comments/%s/add-preview" % (
task["id"],
comment["id"],
)
data = {}
if revision is not None:
data["revision"] = revision
return raw.post(path, data, client=client)
[docs]
def upload_preview_file(
preview_file: str | dict,
file_path: str,
normalize_movie: bool = True,
client: KitsuClient = default,
) -> dict:
"""
Create a preview into given comment.
Args:
preview_file (str / dict): The preview_file dict or the preview_file ID.
file_path (str): Path of the file to upload as preview.
"""
path = (
"pictures/preview-files/%s"
% normalize_model_parameter(preview_file)["id"]
)
if not normalize_movie:
path += "?normalize=false"
return raw.upload(path, file_path, client=client)
[docs]
def add_preview(
task: str | dict,
comment: str | dict,
preview_file_path: str | None = None,
preview_file_url: str | None = None,
normalize_movie: bool = True,
revision: int | None = None,
client: KitsuClient = default,
) -> dict:
"""
Add a preview to given comment.
Args:
task (str / dict): The task dict or the task ID.
comment (str / dict): The comment or the comment ID.
preview_file_path (str): Path of the file to upload as preview.
preview_file_url (str): Url to download the preview file if no path is
given.
normalize_movie (bool): Normalize the movie or not.
revision (int): Revision number.
Returns:
dict: Created preview file model.
"""
if preview_file_url is not None:
preview_file_path = download_file(
preview_file_url,
)
preview_file = create_preview(
task, comment, revision=revision, client=client
)
return upload_preview_file(
preview_file,
preview_file_path,
normalize_movie=normalize_movie,
client=client,
)
[docs]
def publish_preview(
task: str | dict,
task_status: str | dict,
comment: str = "",
person: str | dict | None = None,
checklist: list[dict] = [],
attachments: list[str] = [],
created_at: str | None = None,
preview_file_path: str | None = None,
preview_file_url: str | None = None,
normalize_movie: bool = True,
revision: int | None = None,
set_thumbnail: bool = False,
links: list[str] = [],
client: KitsuClient = default,
) -> tuple[dict, dict]:
"""
Publish a comment and include given preview for given task and set given
task status.
Args:
task (str / dict): The task dict or the task ID.
task_status (str / dict): The task status dict or ID.
comment (str): Comment text
person (str / dict): Comment author
checklist (list): Comment checklist
attachments (list[file_path]): Attachments file paths
created_at (str): Comment date
preview_file_path (str): Path of the file to upload as preview.
preview_file_url (str): Url to download the preview file if no path is
given.
normalize_movie (bool): Set to false to not do operations on it on the
server side.
revision (int): Revision number.
set_thumbnail (bool): Set the preview as thumbnail of the entity.
links (list): List of links to add to the comment
Returns:
tuple(dict, dict): Created comment model and created preview file
model.
"""
new_comment = add_comment(
task,
task_status,
comment=comment,
person=person,
checklist=checklist,
attachments=attachments,
created_at=created_at,
links=links,
client=client,
)
preview_file = add_preview(
task,
new_comment,
preview_file_path=preview_file_path,
preview_file_url=preview_file_url,
normalize_movie=normalize_movie,
revision=revision,
client=client,
)
if set_thumbnail:
set_main_preview(preview_file, client=client)
return new_comment, preview_file
[docs]
def set_main_preview(
preview_file: str | dict,
frame_number: int | None = None,
client: KitsuClient = default,
) -> dict:
"""
Set given preview as thumbnail of given entity.
Args:
preview_file (str / dict): The preview file dict or ID.
frame_number (int): Frame of preview video to set as main preview
Returns:
dict: Created preview file model.
"""
data = {}
if frame_number is not None:
data["frame_number"] = frame_number
preview_file = normalize_model_parameter(preview_file)
path = "actions/preview-files/%s/set-main-preview" % preview_file["id"]
return raw.put(path, data, client=client)
[docs]
@cache
def assign_task(
task: str | dict, person: str | dict, client: KitsuClient = default
) -> dict:
"""
Assign one Person to a Task.
Args:
task (str / dict): The task dict or the task ID.
person (str / dict): The person dict or the person ID.
Returns:
(dict) the affected Task
"""
person = normalize_model_parameter(person)
task = normalize_model_parameter(task)
path = "/actions/persons/%s/assign" % person["id"]
return raw.put(path, {"task_ids": task["id"]}, client=client)
[docs]
def clear_assignations(
tasks: str | dict | list[str | dict],
person: str | dict | None = None,
client: KitsuClient = default,
) -> list[str]:
"""
Clear assignations for a single or multiple tasks.
If no person is given, all assignations are cleared.
Args:
tasks (list / str / dict): Task dict or the task ID, single or list.
person (str / dict): The person dict or the person ID.
Returns:
(list[str]) List of affected Task IDs
"""
if not isinstance(tasks, list):
tasks = [tasks]
data = {}
for task in tasks:
task = normalize_model_parameter(task)
data.setdefault("task_ids", []).append(task["id"])
if person is not None:
person = normalize_model_parameter(person)
data["person_id"] = person["id"]
return raw.put(
"/actions/tasks/clear-assignation", data=data, client=client
)
[docs]
def new_task_type(
name: str,
color: str = "#000000",
for_entity: str = "Asset",
client: KitsuClient = default,
) -> dict:
"""
Create a new task type with the given name.
If a Task Type with the given name already exists within Kitsu, it will
be returned.
Args:
name (str): The name of the task type
color (str): The color of the task type as an hexadecimal string
with # as first character. ex : #00FF00
for_entity (str): The entity type for which the task type is created.
Returns:
dict: The created task type
"""
task_type = get_task_type_by_name(name, for_entity)
if task_type is None:
data = {"name": name, "color": color, "for_entity": for_entity}
task_type = raw.post("data/task-types", data, client=client)
return task_type
[docs]
def new_task_status(
name: str, short_name: str, color: str, client: KitsuClient = default
) -> dict:
"""
Create a new task status with the given name, short name and color.
Args:
name (str): The name of the task status
short_name (str): The short name of the task status
color (str): The color of the task status as an hexadecimal string
with # as first character. ex : #00FF00
Returns:
dict: The created task status
"""
assert color[0] == "#"
assert all(c in string.hexdigits for c in color[1:])
data = {"name": name, "short_name": short_name, "color": color}
return raw.post("data/task-status", data, client=client)
[docs]
def update_task(task: dict, client: KitsuClient = default) -> dict:
"""
Update given task into the API. Metadata are fully replaced by the ones
set on given task.
Args:
task (dict): The task dict to update.
Returns:
dict: Updated task.
"""
if "assignees" in task:
task["assignees"] = normalize_list_of_models_for_links(
task["assignees"]
)
return raw.put("data/tasks/%s" % task["id"], task, client=client)
[docs]
def update_task_data(
task: str | dict, data: dict = {}, client: KitsuClient = default
) -> dict:
"""
Update the metadata for the provided task. Keys that are not provided are
not changed.
Args:
task (str / dict): The task dict or ID to save in database.
data (dict): Free field to set metadata of any kind.
Returns:
dict: Updated task.
"""
task = normalize_model_parameter(task)
current_task = get_task(task["id"], client=client)
updated_task = {"id": current_task["id"], "data": current_task["data"]}
if updated_task["data"] is None:
updated_task["data"] = {}
updated_task["data"].update(data)
return update_task(updated_task, client=client)
[docs]
@cache
def get_task_url(task: dict, client: KitsuClient = default) -> str:
"""
Args:
task (dict): The task dict.
Returns:
url (str): Web url associated to the given task
"""
if not isinstance(task, dict):
raise TaskMustBeADictException
path = "{host}/productions/{project_id}/shots/tasks/{task_id}/"
return path.format(
host=raw.get_api_url_from_host(client=client),
project_id=task["project_id"],
task_id=task["id"],
)
[docs]
def all_tasks_for_project(
project: str | dict,
task_type: str | dict | None = None,
episode: str | dict | None = None,
client: KitsuClient = default,
) -> list[dict]:
"""
Args:
project (str / dict): The project (or its ID) to get tasks from.
task_type (str / dict): The task type (or its ID) to filter tasks.
episode (str / dict): The episode (or its ID) to filter tasks.
Returns:
list[dict]: Tasks related to given project.
"""
project = normalize_model_parameter(project)
path = "/data/projects/%s/tasks" % project["id"]
params = {}
if task_type is not None:
task_type = normalize_model_parameter(task_type)
params["task_type_id"] = task_type["id"]
if episode is not None:
episode = normalize_model_parameter(episode)
params["episode_id"] = episode["id"]
return raw.get(path, params=params, client=client)
[docs]
@cache
def all_open_tasks(client: KitsuClient = default) -> list[dict]:
"""
Get all open tasks.
Returns:
list: Open tasks.
"""
return raw.fetch_all("tasks/open", client=client)
[docs]
@cache
def get_open_tasks_stats(client: KitsuClient = default) -> dict:
"""
Get statistics for open tasks.
Returns:
dict: Open tasks statistics.
"""
return raw.get("data/tasks/open/stats", client=client)
[docs]
@cache
def all_previews_for_task(
task: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Get all previews for a task.
Args:
task (str / dict): The task dict or id.
Returns:
list: Previews for the task.
"""
task = normalize_model_parameter(task)
return raw.fetch_all("tasks/%s/previews" % task["id"], client=client)
[docs]
@cache
def all_open_tasks_for_person(
person: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Get all open tasks for a person.
Args:
person (str / dict): The person dict or id.
Returns:
list: Open tasks for the person.
"""
person = normalize_model_parameter(person)
return raw.fetch_all("persons/%s/tasks/open" % person["id"], client=client)
[docs]
@cache
def all_tasks_for_person_and_type(
person: str | dict, task_type: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Get all tasks for a person and task type.
Args:
person (str / dict): The person dict or id.
task_type (str / dict): The task type dict or id.
Returns:
list: Tasks for the person and task type.
"""
person = normalize_model_parameter(person)
task_type = normalize_model_parameter(task_type)
return raw.fetch_all(
"persons/%s/task-types/%s/tasks" % (person["id"], task_type["id"]),
client=client,
)
[docs]
@cache
def all_notifications_for_project(
project: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Get all notifications for a project.
Args:
project (str / dict): The project dict or id.
Returns:
list: Notifications for the project.
"""
project = normalize_model_parameter(project)
return raw.fetch_all(
"projects/%s/notifications" % project["id"], client=client
)
[docs]
@cache
def all_preview_files_for_project(
project: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Get all preview files for a project.
Args:
project (str / dict): The project dict or id.
Returns:
list: Preview files for the project.
"""
project = normalize_model_parameter(project)
return raw.fetch_all(
"projects/%s/preview-files" % project["id"], client=client
)
[docs]
@cache
def all_subscriptions_for_project(
project: str | dict, client: KitsuClient = default
) -> list[dict]:
"""
Get all subscriptions for a project.
Args:
project (str / dict): The project dict or id.
Returns:
list: Subscriptions for the project.
"""
project = normalize_model_parameter(project)
return raw.fetch_all(
"projects/%s/subscriptions" % project["id"], client=client
)
[docs]
@cache
def get_persons_tasks_dates(
project: str | dict, client: KitsuClient = default
) -> dict:
"""
Get tasks dates for persons in a project.
Args:
project (str / dict): The project dict or id.
Returns:
dict: Tasks dates for persons.
"""
project = normalize_model_parameter(project)
return raw.get(
"data/projects/%s/persons/tasks/dates" % project["id"], client=client
)
[docs]
def remove_tasks_for_type(
project: str | dict, task_type: str | dict, client: KitsuClient = default
) -> str:
"""
Delete all tasks for a task type in a project.
Args:
project (str / dict): The project dict or id.
task_type (str / dict): The task type dict or id.
"""
project = normalize_model_parameter(project)
task_type = normalize_model_parameter(task_type)
return raw.delete(
"data/projects/%s/task-types/%s/tasks"
% (project["id"], task_type["id"]),
client=client,
)
[docs]
def remove_tasks_batch(
tasks: list[str | dict], client: KitsuClient = default
) -> requests.Response:
"""
Delete multiple tasks in batch.
Args:
tasks (list): List of task dicts or IDs to delete.
Returns:
Response: Request response object.
"""
task_ids = [normalize_model_parameter(task)["id"] for task in tasks]
return raw.post(
"data/tasks/delete-batch", {"task_ids": task_ids}, client=client
)
[docs]
def assign_tasks_to_person(
tasks: list[str | dict], person: str | dict, client: KitsuClient = default
) -> dict:
"""
Assign multiple tasks to a person.
Args:
tasks (list): List of task dicts or IDs.
person (str / dict): The person dict or id.
Returns:
dict: Response information.
"""
person = normalize_model_parameter(person)
task_ids = [normalize_model_parameter(task)["id"] for task in tasks]
return raw.put(
"data/tasks/assign",
{"person_id": person["id"], "task_ids": task_ids},
client=client,
)
[docs]
@cache
def get_task_time_spent_for_date(
task: str | dict, date: str, client: KitsuClient = default
) -> dict:
"""
Get time spent for a task on a specific date.
Args:
task (str / dict): The task dict or id.
date (str): Date in YYYY-MM-DD format.
Returns:
dict: Time spent information.
"""
task = normalize_model_parameter(task)
return raw.get(
"data/tasks/%s/time-spent/for-date" % task["id"],
params={"date": date},
client=client,
)
[docs]
def remove_time_spent(
time_spent: str | dict, client: KitsuClient = default
) -> str:
"""
Delete a time spent entry.
Args:
time_spent (str / dict): The time spent dict or id.
Returns:
Response: Request response object.
"""
time_spent = normalize_model_parameter(time_spent)
return raw.delete("data/time-spents/%s" % time_spent["id"], client=client)
[docs]
def create_shot_tasks(
shot: str | dict,
task_types: list[str | dict],
client: KitsuClient = default,
) -> list[dict]:
"""
Create tasks for a shot.
Args:
shot (str / dict): The shot dict or id.
task_types (list): List of task type dicts or IDs.
Returns:
list: Created tasks.
"""
shot = normalize_model_parameter(shot)
task_type_ids = [
normalize_model_parameter(task_type)["id"] for task_type in task_types
]
return raw.post(
"data/shots/%s/tasks" % shot["id"],
{"task_type_ids": task_type_ids},
client=client,
)
[docs]
def create_asset_tasks(
asset: str | dict,
task_types: list[str | dict],
client: KitsuClient = default,
) -> list[dict]:
"""
Create tasks for an asset.
Args:
asset (str / dict): The asset dict or id.
task_types (list): List of task type dicts or IDs.
Returns:
list: Created tasks.
"""
asset = normalize_model_parameter(asset)
task_type_ids = [
normalize_model_parameter(task_type)["id"] for task_type in task_types
]
return raw.post(
"data/assets/%s/tasks" % asset["id"],
{"task_type_ids": task_type_ids},
client=client,
)
[docs]
def create_edit_tasks(
edit: str | dict,
task_types: list[str | dict],
client: KitsuClient = default,
) -> list[dict]:
"""
Create tasks for an edit.
Args:
edit (str / dict): The edit dict or id.
task_types (list): List of task type dicts or IDs.
Returns:
list: Created tasks.
"""
edit = normalize_model_parameter(edit)
task_type_ids = [
normalize_model_parameter(task_type)["id"] for task_type in task_types
]
return raw.post(
"data/edits/%s/tasks" % edit["id"],
{"task_type_ids": task_type_ids},
client=client,
)
[docs]
def create_concept_tasks(
concept: str | dict,
task_types: list[str | dict],
client: KitsuClient = default,
) -> list[dict]:
"""
Create tasks for a concept.
Args:
concept (str / dict): The concept dict or id.
task_types (list): List of task type dicts or IDs.
Returns:
list: Created tasks.
"""
concept = normalize_model_parameter(concept)
task_type_ids = [
normalize_model_parameter(task_type)["id"] for task_type in task_types
]
return raw.post(
"data/concepts/%s/tasks" % concept["id"],
{"task_type_ids": task_type_ids},
client=client,
)
[docs]
def create_entity_tasks(
entity: str | dict,
task_types: list[str | dict],
client: KitsuClient = default,
) -> list[dict]:
"""
Create tasks for an entity.
Args:
entity (str / dict): The entity dict or id.
task_types (list): List of task type dicts or IDs.
Returns:
list: Created tasks.
"""
entity = normalize_model_parameter(entity)
task_type_ids = [
normalize_model_parameter(task_type)["id"] for task_type in task_types
]
return raw.post(
"data/entities/%s/tasks" % entity["id"],
{"task_type_ids": task_type_ids},
client=client,
)