From ece086c03f9eede9ce8214afbb8a2261b2fbd8d5 Mon Sep 17 00:00:00 2001 From: Jakub Trllo <43494761+iLLiCiTiT@users.noreply.github.com> Date: Fri, 7 Nov 2025 15:52:35 +0100 Subject: [PATCH] merge tho methods into one --- client/ayon_core/tools/workfiles/abstract.py | 45 +++++---- client/ayon_core/tools/workfiles/control.py | 17 ++-- .../tools/workfiles/models/workfiles.py | 91 +++++++++---------- 3 files changed, 82 insertions(+), 71 deletions(-) diff --git a/client/ayon_core/tools/workfiles/abstract.py b/client/ayon_core/tools/workfiles/abstract.py index e7a038575a..b41553acc4 100644 --- a/client/ayon_core/tools/workfiles/abstract.py +++ b/client/ayon_core/tools/workfiles/abstract.py @@ -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. diff --git a/client/ayon_core/tools/workfiles/control.py b/client/ayon_core/tools/workfiles/control.py index 13a88e325f..c399a1bf33 100644 --- a/client/ayon_core/tools/workfiles/control.py +++ b/client/ayon_core/tools/workfiles/control.py @@ -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): diff --git a/client/ayon_core/tools/workfiles/models/workfiles.py b/client/ayon_core/tools/workfiles/models/workfiles.py index 5055c203be..c15dda2b4f 100644 --- a/client/ayon_core/tools/workfiles/models/workfiles.py +++ b/client/ayon_core/tools/workfiles/models/workfiles.py @@ -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