import string
from . import client as raw
from .sorting import sort_by_name
from .helpers import normalize_model_parameter
from .cache import cache
default = raw.default_client
[docs]@cache
def all_task_statuses(client=default):
"""
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=default):
"""
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_tasks_for_shot(shot, relations=False, client=default):
"""
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_sequence(sequence, relations=False, client=default):
"""
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, relations=False, client=default):
"""
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, relations=False, client=default):
"""
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, relations=False, client=default):
"""
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, relations=False, client=default):
"""
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, relations=False, client=default):
"""
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_tasks_for_task_status(project, task_type, task_status, client=default):
"""
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, task_type, client=default):
"""
Args:
project (str / dict): The project dict or the project ID.
task_type (str / dict): The task type dict or ID.
Returns:
list: Tasks for given project and task type.
"""
project = normalize_model_parameter(project)
task_type = normalize_model_parameter(task_type)
return raw.fetch_all(
"tasks",
{
"project_id": project["id"],
"task_type_id": task_type["id"],
},
client=client
)
[docs]@cache
def all_task_types_for_shot(shot, client=default):
"""
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_asset(asset, client=default):
"""
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, client=default):
"""
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, client=default):
"""
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, client=default):
"""
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, task_type, client=default):
"""
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, client=default):
"""
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, client=default):
"""
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)
@cache
def get_task_by_entity(entity, task_type, client=default):
return get_task_by_name(entity, task_type, client=client)
[docs]@cache
def get_task_by_name(entity, task_type, name="main", client=default):
"""
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 name for given entity and task type.
"""
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, client=default):
"""
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, client=default):
"""
Args:
task_type_name (str): Name of claimed task type.
Returns:
dict: Task type object for given name.
"""
return raw.fetch_first(
"task-types", {"name": task_type_name}, client=client
)
[docs]@cache
def get_task_by_path(project, file_path, entity_type="shot", client=default):
"""
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_task_status(task_status_id, client=default):
"""
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, client=default):
"""
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, client=default):
"""
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_status(task_status, client=default):
"""
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]@cache
def get_task(task_id, client=default):
"""
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,
task_type,
name="main",
task_status=None,
assigner=None,
assignees=None,
client=default
):
"""
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_task_status_by_name("Todo", client=client)
data = {
"project_id": entity["project_id"],
"entity_id": entity["id"],
"task_type_id": task_type["id"],
"task_status_id": task_status["id"],
"name": name,
}
if assigner is not None:
data["assigner_id"] = assigner["id"]
if assignees is not None:
data["assignees"] = [person["id"] for person in assignees]
else:
data["assignees"] = []
task = get_task_by_name(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, client=default):
"""
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, client=default):
"""
Change a task status to WIP and set its real start date to now.
Args:
task (str / dict): The task dict or the task ID.
Returns:
dict: Modified task.
"""
task = normalize_model_parameter(task)
path = "actions/tasks/%s/start" % task["id"]
return raw.put(path, {}, client=client)
[docs]def task_to_review(
task, person, comment, revision=1, change_status=True, client=default
):
"""
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, date, client=default):
"""
Get the time spent by CG artists on a task at a given date. A field contains
the total time spent. Durations are given in seconds. 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/%s" % (task["id"], date)
return raw.get(path, client=client)
[docs]def set_time_spent(task, person, date, duration, client=default):
"""
Set the time spent by a CG artist on a given task at a given date. Durations
must be set in seconds. Date format is YYYY-MM-DD.
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 for which time spent must be set.
duration (int): The duration 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, person, date, duration, client=default):
"""
Add given duration to the already logged duration for given task and person
at a given date. Durations must be set in seconds. Date format is
YYYY-MM-DD.
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 for which time spent must be added.
duration (int): The duration to add on 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, comment, client=default):
"""
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.
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"],
)
return raw.post(path, {}, client=client)
[docs]def upload_preview_file(preview, file_path, client=default):
"""
Create a preview into given comment.
Args:
task (str / dict): The task dict or the task ID.
file_path (str): Path of the file to upload as preview.
"""
path = "pictures/preview-files/%s" % preview["id"]
raw.upload(path, file_path, client=client)
[docs]def add_preview(task, comment, preview_file_path, client=default):
"""
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.
Returns:
dict: Created preview file model.
"""
preview_file = create_preview(task, comment, client=client)
upload_preview_file(preview_file, preview_file_path, client=client)
return preview_file
[docs]def set_main_preview(preview_file, client=default):
"""
Set given preview as thumbnail of given entity.
Args:
preview_file (str / dict): The preview file dict or ID.
Returns:
dict: Created preview file model.
"""
preview_file = normalize_model_parameter(preview_file)
path = "actions/preview-files/%s/set-main-preview" % preview_file["id"]
return raw.put(path, {}, client=client)
[docs]@cache
def assign_task(task, person, client=default):
"""
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 new_task_type(name, client=default):
"""
Create a new task type with the given name.
Args:
name (str): The name of the task type
Returns:
dict: The created task type
"""
data = {"name": name}
return raw.post("data/task-types", data, client=client)
[docs]def new_task_status(name, short_name, color, client=default):
"""
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 has 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, client=default):
"""
Save given task data 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.
"""
return raw.put("data/tasks/%s" % task["id"], task, client=client)
[docs]def update_task_data(task, data={}, client=default):
"""
Update the metadata for the provided task. Keys that are not provided are
not changed.
Args:
task (dict / ID): 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)
update_task(updated_task, client=client)
[docs]@cache
def get_task_url(task, client=default):
"""
Args:
task (str / dict): The task dict or the task ID.
Returns:
url (str): Web url associated to the given task
"""
task = normalize_model_parameter(task)
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, client=default):
"""
Args:
project (str / dict): The project
Returns:
dict: Tasks related to given project.
"""
project = normalize_model_parameter(project)
path = "/data/projects/%s/tasks" % project["id"]
return raw.get(path, client=client)