Example API for linking CG-Wire with Avalon with Qtazu

avalon_cgwire_qtazu_link.py

"""These are utilities for connecting open-source pipeline Avalon with CG-Wire using Qtazu

See: 
- https://github.com/getavalon/core
- https://github.com/Colorbleed/qtazu

References:
    - https://github.com/getavalon/docker/blob/master/volume/sync.py#L227

# How it works:

Each Avalon document should store `data.cgwire_link`, like:
```python
{"data": {
    "cgwire_link": {
        "id": "cgwire-entity-id",
        "type": "cgwire-entity-type"
        }
    }
}
```

Using this link we can now query the corresponding CG-Wire data
from just the Avalon document.

```python
# Assume `asset` is Avalon asset document
cgwire_link = _get_cgwire_link(asset)
print(cgwire_link)
```

The reverse would similarly be possible if we would store `avalon_link`
in the CG-Wire entities. This data can be stored there using the `gazu`
api. Which is not yet shown in this code.

"""
from qtazu.utils import get_cgwire_data
import gazu
import avalon.io as io
import avalon.api as api


def _get_cgwire_link(document):
    """Get link from data in Avalon database document."""
    cgwire_link = document['data'].get("cgwire_link", None)
    if cgwire_link is None:
        # No link stored in Avalon asset to CG-Wire asset/shot.
        raise RuntimeError("No cgwire link found for: %s" % document["name"])

    # For now it seems some CG-Wire commands fail on the unicode type in
    # Python 2.7
    # so let's ensure it's `str` for now.
    for key, value in cgwire_link.items():
        cgwire_link[key] = str(value)

    return cgwire_link


def avalon_project_to_cgwire():
    """Get cg-wire project id from active project"""

    avalon_project = io.find_one({"type": "project"})
    link = _get_cgwire_link(avalon_project)
    return get_cgwire_data(link)


def avalon_asset_to_cgwire(avalon_asset):
    """Get cg-wire asset id from Avalon asset"""

    link = _get_cgwire_link(avalon_asset)
    return get_cgwire_data(link)


def avalon_asset_task_to_cgwire(avalon_asset, task_name):
    """From asset id with avalon task name get cgwire task id."""

    # Request task type by name
    task_type = gazu.task.get_task_type_by_name(task_name)
    if task_type is None:
        raise ValueError("There is no cgwire task type named: %s" % task_name)

    link = _get_cgwire_link(avalon_asset)
    entity = get_cgwire_data(link)  # Get CG-Wire Asset or Shot

    tasks = gazu.task.all_tasks_for_entity_and_task_type(entity, task_type)

    # todo: support multiple tasks
    if len(tasks) != 1:
        raise RuntimeError(
            "Not exactly a single task found, instead found: "
            "%s (%s)" % (len(tasks), tasks)
        )

    return tasks[0]


if __name__ == "__main__":

    # Example code
    asset_name = api.Session["AVALON_ASSET"]
    avalon_asset = io.find_one({"name": asset_name, "type": "asset"})
    task_name = api.Session["AVALON_TASK"]
    cgwire_task = avalon_asset_task_to_cgwire(avalon_asset, task_name)