Frame13/ayon-core
1
0
Fork 1
mirror of https://github.com/ynput/ayon-core.git synced 2025-12-24 12:54:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

merge tho methods into one

Browse source
This commit is contained in:
Jakub Trllo 2025-11-07 15:52:35 +01:00
parent 3338dbe473
commit ece086c03f
3 changed files with 82 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

45
client/ayon_core/tools/workfiles/abstract.py
View file

@ -1,8 +1,18 @@
from __future__ import annotations
import os
from abc import ABC, abstractmethod
import typing
from typing import Optional
from ayon_core.style import get_default_entity_icon_color
if typing.TYPE_CHECKING:
from ayon_core.host import (
WorkfileInfo,
PublishedWorkfileInfo,
)
class FolderItem:
"""Item representing folder entity on a server.
@ -159,6 +169,17 @@ class WorkareaFilepathResult:
self.filepath = filepath
class PublishedWorkfileWrap:
"""Wrapper for workfile info that also contains version comment."""
def __init__(
self,
info: Optional[PublishedWorkfileInfo] = None,
comment: Optional[str] = None,
) -> None:
self.info = info
self.comment = comment
class AbstractWorkfilesCommon(ABC):
@abstractmethod
def is_host_valid(self):
@ -788,32 +809,24 @@ class AbstractWorkfilesFrontend(AbstractWorkfilesCommon):
pass
@abstractmethod
def get_published_workfile_info(self, representation_id: str):
def get_published_workfile_info(
self,
folder_id: Optional[str],
representation_id: Optional[str],
) -> PublishedWorkfileWrap:
"""Get published workfile info by representation ID.
Args:
representation_id (str): Representation id.
folder_id (Optional[str]): Folder id.
representation_id (Optional[str]): Representation id.
Returns:
Optional[PublishedWorkfileInfo]: Published workfile info or None
PublishedWorkfileWrap: Published workfile info or None
if not found.
"""
pass
@abstractmethod
def get_published_workfile_version_comment(self, representation_id: str):
"""Get version comment for published workfile.
Args:
representation_id (str): Representation id.
Returns:
Optional[str]: Version comment or None.
"""
pass
@abstractmethod
def get_workfile_info(self, folder_id, task_id, rootless_path):
"""Workfile info from database.

17
client/ayon_core/tools/workfiles/control.py
View file

@ -1,4 +1,7 @@
from __future__ import annotations
import os
from typing import Optional
import ayon_api
@ -18,6 +21,7 @@ from ayon_core.tools.common_models import (
from .abstract import (
AbstractWorkfilesBackend,
AbstractWorkfilesFrontend,
PublishedWorkfileWrap,
)
from .models import SelectionModel, WorkfilesModel
@ -432,14 +436,13 @@ class BaseWorkfileController(
folder_id, task_id
)
def get_published_workfile_info(self, representation_id):
def get_published_workfile_info(
self,
folder_id: Optional[str],
representation_id: Optional[str],
) -> PublishedWorkfileWrap:
return self._workfiles_model.get_published_workfile_info(
representation_id
)
def get_published_workfile_version_comment(self, representation_id):
return self._workfiles_model.get_published_workfile_version_comment(
representation_id
folder_id, representation_id
)
def get_workfile_info(self, folder_id, task_id, rootless_path):

91
client/ayon_core/tools/workfiles/models/workfiles.py
View file

@ -39,6 +39,7 @@ from ayon_core.pipeline.workfile import (
from ayon_core.pipeline.version_start import get_versioning_start
from ayon_core.tools.workfiles.abstract import (
WorkareaFilepathResult,
PublishedWorkfileWrap,
AbstractWorkfilesBackend,
)
@ -79,7 +80,7 @@ class WorkfilesModel:
# Published workfiles
self._repre_by_id = {}
self._version_by_repre_id = {}
self._version_comment_by_id = {}
self._published_workfile_items_cache = NestedCacheItem(
levels=1, default_factory=list
)
@ -96,7 +97,7 @@ class WorkfilesModel:
self._workarea_file_items_cache.reset()
self._repre_by_id = {}
self._version_by_repre_id = {}
self._version_comment_by_id = {}
self._published_workfile_items_cache.reset()
self._workfile_entities_by_task_id = {}
@ -554,13 +555,13 @@ class WorkfilesModel:
)
def get_published_file_items(
self, folder_id: str, task_id: str
self, folder_id: Optional[str], task_id: Optional[str]
) -> list[PublishedWorkfileInfo]:
"""Published workfiles for passed context.
Args:
folder_id (str): Folder id.
task_id (str): Task id.
folder_id (Optional[str]): Folder id.
task_id (Optional[str]): Task id.
Returns:
list[PublishedWorkfileInfo]: List of files for published workfiles.
@ -604,17 +605,10 @@ class WorkfilesModel:
})
# Map versions by representation ID for easy lookup
version_by_id = {
version_entity["id"]: version_entity
self._version_comment_by_id.update({
version_entity["id"]: version_entity["attrib"].get("comment")
for version_entity in version_entities
}
for repre_entity in repre_entities:
repre_id = repre_entity["id"]
version_id = repre_entity.get("versionId")
if version_id and version_id in version_by_id:
self._version_by_repre_id[repre_id] = version_by_id[
version_id
]
})
project_entity = self._controller.get_project_entity(project_name)
@ -643,50 +637,32 @@ class WorkfilesModel:
return items
def get_published_workfile_info(
self, representation_id: str
) -> Optional[PublishedWorkfileInfo]:
self,
folder_id: Optional[str],
representation_id: Optional[str],
) -> PublishedWorkfileWrap:
"""Get published workfile info by representation ID.
Args:
representation_id (str): Representation id.
folder_id (Optional[str]): Folder id.
representation_id (Optional[str]): Representation id.
Returns:
Optional[PublishedWorkfileInfo]: Published workfile info or None
PublishedWorkfileWrap: Published workfile info or None
if not found.
"""
if not representation_id:
return None
return PublishedWorkfileWrap()
# Search through all cached published workfile items
cache_items = self._published_workfile_items_cache._data_by_key
for folder_cache in cache_items.values():
if folder_cache.is_valid:
for item in folder_cache.get_data():
if item.representation_id == representation_id:
return item
return None
def get_published_workfile_version_comment(
self, representation_id: str
) -> Optional[str]:
"""Get version comment for published workfile.
Args:
representation_id (str): Representation id.
Returns:
Optional[str]: Version comment or None.
"""
if not representation_id:
return None
version_entity = self._version_by_repre_id.get(representation_id)
if version_entity:
attrib = version_entity.get("attrib") or {}
return attrib.get("comment")
return None
for item in self.get_published_file_items(folder_id, None):
if item.representation_id == representation_id:
comment = self._get_published_workfile_version_comment(
representation_id
)
return PublishedWorkfileWrap(item, comment)
return PublishedWorkfileWrap()
@property
def _project_name(self) -> str:
@ -704,6 +680,25 @@ class WorkfilesModel:
self._current_username = get_ayon_username()
return self._current_username
def _get_published_workfile_version_comment(
self, representation_id: str
) -> Optional[str]:
"""Get version comment for published workfile.
Args:
representation_id (str): Representation id.
Returns:
Optional[str]: Version comment or None.
"""
if not representation_id:
return None
repre = self._repre_by_id.get(representation_id)
if not repre:
return None
return self._version_comment_by_id.get(repre["versionId"])
# --- Host ---
def _open_workfile(self, folder_id: str, task_id: str, filepath: str):
# TODO move to workfiles pipeline