from . import client
from .sorting import sort_by_name
from .helpers import normalize_model_parameter
from .cache import cache
[docs]@cache
def all_task_statuses():
"""
Returns:
Task statuses stored in database.
"""
task_statuses = client.fetch_all("task-status")
return sort_by_name(task_statuses)
[docs]@cache
def all_task_types():
"""
Returns:
list: Task types stored in database.
"""
task_types = client.fetch_all("task-types")
return sort_by_name(task_types)
[docs]@cache
def all_tasks_for_shot(shot):
"""
Args:
shot (str / dict): The shot dict or the shot ID.
Returns:
list: Tasks linked to given shot.
"""
shot = normalize_model_parameter(shot)
tasks = client.fetch_all("shots/%s/tasks" % shot['id'])
return sort_by_name(tasks)
[docs]@cache
def all_tasks_for_sequence(sequence):
"""
Args:
sequence (str / dict): The sequence dict or the sequence ID.
Returns
list: Tasks linked to given sequence.
"""
sequence = normalize_model_parameter(sequence)
tasks = client.fetch_all("sequences/%s/tasks" % sequence['id'])
return sort_by_name(tasks)
[docs]@cache
def all_tasks_for_scene(scene):
"""
Args:
sequence (str / dict): The scene dict or the scene ID.
Returns:
list: Tasks linked to given scene.
"""
scene = normalize_model_parameter(scene)
tasks = client.fetch_all("scenes/%s/tasks" % scene['id'])
return sort_by_name(tasks)
[docs]@cache
def all_tasks_for_asset(asset):
"""
Args:
asset (str / dict): The asset dict or the asset ID.
Returns:
list: Tasks directly linked to given asset.
"""
asset = normalize_model_parameter(asset)
tasks = client.fetch_all("assets/%s/tasks" % asset["id"])
return sort_by_name(tasks)
[docs]@cache
def all_tasks_for_task_status(project, task_type, task_status):
"""
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 client.fetch_all(
"tasks?project_id=%s&task_type_id=%s&task_status_id=%s" % (
project["id"],
task_type["id"],
task_status["id"]
)
)
[docs]@cache
def all_task_types_for_shot(shot):
"""
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)
task_types = client.fetch_all("shots/%s/task-types" % shot['id'])
return sort_by_name(task_types)
[docs]@cache
def all_task_types_for_asset(asset):
"""
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 = client.fetch_all("assets/%s/task-types" % asset['id'])
return sort_by_name(task_types)
[docs]@cache
def all_task_types_for_scene(scene):
"""
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)
task_types = client.fetch_all("scenes/%s/task-types" % scene['id'])
return sort_by_name(task_types)
[docs]@cache
def all_task_types_for_sequence(sequence):
"""
Args:
sequence (str / dict): The sequence dict or the sequence ID.
Returns:
Task types of tasks linked directly to given sequence.
"""
sequence = normalize_model_parameter(sequence)
task_types = client.fetch_all("sequences/%s/task-types" % sequence['id'])
return sort_by_name(task_types)
[docs]@cache
def all_tasks_for_entity_and_task_type(entity, task_type):
"""
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"]
return client.fetch_all(
"entities/%s/task-types/%s/tasks" % (
entity_id,
task_type_id
)
)
[docs]@cache
def get_task_by_name(entity, task_type, name="main"):
"""
Deprecated.
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 client.fetch_first(
"tasks?name={name}&entity_id={entity_id}&task_type_id={task_type_id}"
.format(
name=name,
task_type_id=task_type["id"],
entity_id=entity["id"]
)
)
[docs]@cache
def get_task_type(task_type_id):
"""
Args:
task_type_id (str): Id of claimed task type.
Returns:
dict: Task type matching given ID.
"""
return client.fetch_one("task-types", task_type_id)
[docs]@cache
def get_task_type_by_name(task_type_name):
"""
Args:
task_type_name (str): Name of claimed task type.
Returns:
dict: Task type object for given name.
"""
return client.fetch_first("task-types?name=%s" % task_type_name)
[docs]@cache
def get_task_by_path(project, file_path, entity_type="shot"):
"""
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 client.post("data/tasks/from-path/", data)
[docs]@cache
def get_task_status(task):
"""
Args:
task (str / dict): The task dict or the task ID.
Returns:
A task status object corresponding to status set on given task.
"""
return client.fetch_first(
"task-status?id={task_status_id}".format(
task_status_id=task['task_status_id']
)
)
[docs]@cache
def get_task_status_by_name(name):
"""
Args:
name (str / dict): The name of claimed task status.
Returns:
dict: Task status matching given name.
"""
return client.fetch_first("task-status?name=%s" % name)
[docs]@cache
def get_task_status_by_short_name(task_status_short_name):
"""
Args:
short_name (str / dict): The short name of claimed task status.
Returns:
dict: Task status matching given short name.
"""
return client.fetch_first(
"task-status?short_name=%s" % task_status_short_name
)
[docs]def remove_task_status(task_status):
"""
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 client.delete("data/task-status/%s?force=true" % task_status["id"])
[docs]@cache
def get_task(task_id):
"""
Args:
task_id (str): Id of claimed task.
Returns:
dict: Task matching given ID.
"""
return client.get('data/tasks/%s/full' % task_id)
[docs]def new_task(
entity,
task_type,
name="main",
task_status=None,
assigner=None,
assignees=None
):
"""
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")
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)
if task is None:
task = client.post("data/tasks", data)
return task
[docs]def remove_task(task):
"""
Remove given task from database.
Args:
task (str / dict): The task dict or the task ID.
"""
task = normalize_model_parameter(task)
client.delete("data/tasks/%s?force=true" % task["id"])
[docs]def start_task(task):
"""
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 client.put(path, {})
[docs]def task_to_review(
task,
person,
comment,
revision=1,
change_status=True
):
"""
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 client.put(path, data)
[docs]@cache
def get_time_spent(task, date):
"""
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 client.get(path)
[docs]def set_time_spent(task, person, date, duration):
"""
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 client.post(path, {"duration": duration})
[docs]def add_time_spent(task, person, date, duration):
"""
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 client.post(path, {"duration": duration})
[docs]def add_preview(task, comment, preview_file_path):
"""
Add a preview to given comment. It it's a movie, it must be mentioned
in the option.
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.
"""
task = normalize_model_parameter(task)
comment = normalize_model_parameter(comment)
path = "actions/tasks/%s/comments/%s/add-preview" % (
task["id"],
comment["id"]
)
preview_file = client.post(path, {})
path = "pictures/preview-files/%s" % preview_file["id"]
client.upload(path, preview_file_path)
return preview_file
[docs]def set_main_preview(entity, preview_file):
"""
Set given preview as thumbnail of given entity.
Args:
task (str / dict): The task dict or the task ID.
preview_file (str / dict): The preview file dict or ID.
Returns:
dict: Created preview file model.
"""
entity = normalize_model_parameter(entity)
preview_file = normalize_model_parameter(preview_file)
path = "actions/entities/%s/set-main-preview/%s" % (
entity["id"],
preview_file["id"]
)
return client.put(path, {})
[docs]@cache
def assign_task(task, person):
"""
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)
route = "/actions/persons/%s/assign" % person['id']
return client.put(route, {'task_ids': task['id']})