diff --git a/python/tk_multi_loader/medm/__init__.py b/python/tk_multi_loader/medm/__init__.py
new file mode 100644
index 0000000..4ebbf58
--- /dev/null
+++ b/python/tk_multi_loader/medm/__init__.py
@@ -0,0 +1,32 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""MEDM (Flow Asset Management) integration models for the Loader app.
+
+This package provides Qt models that back the loader when ``use_medm_data``
+is enabled in the app configuration. All models share a single
+:class:`~medm.shared_cache.MedmSharedCache` and
+:class:`~medm.thumbnail_service.MedmThumbnailService` instance injected by
+the dialog at construction time.
+"""
+
+from .entity_model import MedmEntityModel
+from .latestpublish_model import MedmLatestPublishModel
+from .publishhistory_model import MedmPublishHistoryModel
+from .shared_cache import MedmSharedCache
+from .thumbnail_service import MedmThumbnailService
+
+__all__ = [
+ "MedmEntityModel",
+ "MedmLatestPublishModel",
+ "MedmPublishHistoryModel",
+ "MedmSharedCache",
+ "MedmThumbnailService",
+]
diff --git a/python/tk_multi_loader/medm/entity_model.py b/python/tk_multi_loader/medm/entity_model.py
new file mode 100644
index 0000000..93c9ac5
--- /dev/null
+++ b/python/tk_multi_loader/medm/entity_model.py
@@ -0,0 +1,448 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""MEDM Asset Model - Tree model for MEDM asset hierarchy
+
+This module provides a tree model that displays MEDM assets
+in the left-hand tree view of the loader.
+
+Children are loaded lazily: only the immediate children of the current project
+are fetched on startup. Deeper levels are fetched on demand when the user
+expands a tree node (via Qt's ``canFetchMore`` / ``fetchMore`` protocol).
+The shared :class:`~medm.shared_cache.MedmSharedCache` ``children`` dict
+prevents duplicate API round-trips when the same asset's children are
+requested by both the tree and the center-panel publish model.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, List, Optional
+
+import sgtk
+from sgtk.platform.qt import QtCore, QtGui
+
+from .shared_cache import MedmSharedCache
+from .utils import is_structural_asset as _is_structural_asset_util
+
+# Import types for type hints only - actual objects come from framework at runtime
+# Framework is loaded dynamically via sgtk.platform.import_framework()
+if TYPE_CHECKING:
+ from adsk.flow.data import Asset
+else:
+ Asset = Any
+
+
+class MedmEntityModel(QtGui.QStandardItemModel):
+ """
+ Tree model that displays MEDM assets in a hierarchical structure.
+ This replaces SgEntityModel for MEDM data sources.
+
+ Uses lazy loading: only the project's immediate children are fetched at
+ startup. Deeper levels are fetched when the user expands a node.
+ """
+
+ # Custom roles - matching ShotgunModel interface
+ SG_DATA_ROLE = QtCore.Qt.UserRole + 1
+ SG_ASSOCIATED_FIELD_ROLE = QtCore.Qt.UserRole + 2
+ ASSET_ROLE = (
+ QtCore.Qt.UserRole + 200
+ ) # Stores MEDM Asset object (shared with all MEDM models)
+
+ # Lazy-loading bookkeeping role: True once children have been fetched for a node.
+ CHILDREN_LOADED_ROLE = QtCore.Qt.UserRole + 201
+
+ # Signals - required for ShotgunModelOverlayWidget compatibility
+ cache_loaded = QtCore.Signal()
+ data_refreshed = QtCore.Signal(bool) # Argument: data_changed
+ query_changed = QtCore.Signal()
+ data_refreshing = QtCore.Signal()
+ data_refresh_fail = QtCore.Signal(str)
+
+ def __init__(
+ self,
+ parent,
+ entity_type,
+ filters,
+ hierarchy,
+ bg_task_manager,
+ cache: Optional[MedmSharedCache] = None,
+ ):
+ """
+ Constructor
+
+ :param parent: Parent QObject
+ :param entity_type: Entity type (kept for API compatibility, not used)
+ :param filters: Filters (kept for API compatibility, not used)
+ :param hierarchy: Hierarchy fields (kept for API compatibility, not used)
+ :param bg_task_manager: Background task manager (kept for API compatibility)
+ :param cache: Shared :class:`MedmSharedCache`. When provided the
+ ``children`` dict is used so that expanding a tree node and
+ selecting it in the centre panel never duplicate an API call.
+ When *None* a private fallback cache is used (test / standalone use).
+ """
+ super().__init__(parent)
+
+ self._app = sgtk.platform.current_bundle()
+ self._flow_module = sgtk.platform.import_framework(
+ "tk-framework-flowam", "flow"
+ )
+
+ self._cache = cache if cache is not None else MedmSharedCache()
+ self._folder_icon = QtGui.QIcon(QtGui.QPixmap(":/res/icon_Folder.png"))
+ self._binary_icon = QtGui.QIcon(QtGui.QPixmap(":/res/icon_Asset_dark.png"))
+
+ # Lazily resolved set of structural type IDs (folder, pipeline step).
+ # Assets matching any of these are always shown in the tree;
+ # others are only shown when they have structural descendants.
+ self._structural_type_ids: Optional[set] = None
+
+ self._project = None
+ self._initialize_project()
+
+ # Defer loading to allow UI to set up first (shows spinner)
+ self.data_refreshing.emit()
+ QtCore.QTimer.singleShot(100, self._load_medm_assets)
+
+ # -------------------------------------------------------------------------
+ # Qt virtual overrides - lazy loading protocol
+ # -------------------------------------------------------------------------
+
+ def hasChildren(self, parent: QtCore.QModelIndex = QtCore.QModelIndex()) -> bool:
+ """
+ Return ``True`` when *parent* might have children.
+
+ For nodes whose children have not been fetched yet we optimistically
+ return ``True`` so that Qt draws an expansion arrow. Once children
+ have been loaded the answer is based on the actual row count.
+ """
+ if not parent.isValid():
+ return self.rowCount() > 0
+ item = self.itemFromIndex(parent)
+ if item is None:
+ return False
+ if item.data(self.CHILDREN_LOADED_ROLE):
+ return item.rowCount() > 0
+ # Not yet loaded -> assume children exist (shows the expand arrow)
+ return True
+
+ def canFetchMore(self, parent: QtCore.QModelIndex) -> bool:
+ """Return ``True`` if *parent*'s children have not been loaded yet."""
+ if not parent.isValid():
+ return False
+ item = self.itemFromIndex(parent)
+ if item is None:
+ return False
+ return not item.data(self.CHILDREN_LOADED_ROLE)
+
+ def fetchMore(self, parent: QtCore.QModelIndex) -> None:
+ """Load the immediate children of *parent* from the MEDM API (or cache)."""
+ if not parent.isValid():
+ return
+ item = self.itemFromIndex(parent)
+ if item is None or item.data(self.CHILDREN_LOADED_ROLE):
+ return
+ self._load_children_for_item(item)
+
+ # -------------------------------------------------------------------------
+ # Public API - Called by dialog.py and other external code
+ # -------------------------------------------------------------------------
+
+ def destroy(self) -> None:
+ """Clean up model resources."""
+ self._cache.children.clear()
+
+ def async_refresh(self) -> None:
+ """Refresh the model data."""
+ self.clear()
+ self._cache.clear_on_hard_refresh()
+ self.data_refreshing.emit()
+ QtCore.QTimer.singleShot(100, self._load_medm_assets)
+
+ def hard_refresh(self) -> None:
+ """Hard refresh (same as async_refresh for this simple model)."""
+ self.async_refresh()
+
+ def item_from_entity(
+ self, entity_type: str, entity_id: int
+ ) -> Optional[QtGui.QStandardItem]:
+ """
+ Returns a QStandardItem based on entity type and entity id.
+
+ **OVERRIDE:** This method overrides the ShotgunModel.item_from_entity() interface
+ to provide MEDM-compatible implementation. The original ShotgunModel version uses
+ an internal data handler (_data_handler.get_uid_from_entity_id), but MEDM models
+ store data in a tree structure requiring recursive search.
+
+ **Implementation Differences from ShotgunModel.item_from_entity:**
+ - Original: Uses flat data handler lookup (uid-based)
+ - This override: Performs recursive tree search through QStandardItem hierarchy
+ - Original: Validates entity_type matches model's __entity_type
+ - This override: Ignores entity_type (MEDM uses unified Asset model)
+
+ **Note:** Method name preserved for API compatibility with dialog.py which expects
+ all entity models (SgEntityModel, SgHierarchyModel, MedmEntityModel) to implement
+ this interface. Called by dialog._get_item_from_entity() for navigation/selection.
+
+ :param entity_type: Shotgun entity type (ignored in MEDM implementation)
+ :param entity_id: Entity ID to search for (compared against SG_DATA_ROLE["id"])
+ :returns: :class:`~PySide.QtGui.QStandardItem` or None if not found
+ """
+
+ def search_item(parent):
+ for row in range(parent.rowCount() if parent else self.rowCount()):
+ item = parent.child(row) if parent else self.item(row)
+ if item:
+ sg_data = item.data(self.SG_DATA_ROLE)
+ if sg_data and sg_data.get("id") == entity_id:
+ return item
+ found = search_item(item)
+ if found:
+ return found
+ return None
+
+ return search_item(None)
+
+ def get_cached_children(self, asset: Asset) -> List[Asset]:
+ """
+ Return child :class:`Asset` objects for *asset*.
+
+ Uses the internal cache when available; otherwise fetches from the MEDM
+ API and stores the result. This is the single entry-point that both
+ the tree's ``fetchMore`` and :class:`MedmLatestPublishModel` use, so
+ that a drill-down never fetches the same level twice.
+
+ :param asset: Parent MEDM Asset whose children are needed.
+ :returns: List of child Asset objects (may be empty).
+ """
+ return self._fetch_and_cache_children(asset)
+
+ # -------------------------------------------------------------------------
+ # Private utility methods - Internal implementation details
+ # -------------------------------------------------------------------------
+
+ def _initialize_project(self) -> None:
+ """
+ Initialize and cache the MEDM Project object.
+ Called during __init__ to fail fast if project is unavailable.
+ """
+ try:
+ session_project = self._flow_module.data.get_session_project()
+ self._project = self._flow_module.data.Project(session_project.id)
+ self._app.log_debug(
+ f"MEDM Entity: Initialized project '{self._project.name}'"
+ )
+ except Exception as e:
+ self._app.log_error(
+ f"MEDM Entity: Failed to initialize project: {type(e).__name__}: {e}. "
+ "Entity tree will not be loaded."
+ )
+ self._project = None
+
+ def _get_structural_type_ids(self) -> set:
+ """
+ Return the set of type-ID strings that are always shown in the tree
+ regardless of whether they have children.
+
+ The set is resolved once and cached on the instance. It contains:
+ - ``FOLDER_TYPE_ID`` - Autodesk built-in type, available as a constant.
+ - pipeline-step - schema-registered type whose ID varies per collection
+ and is resolved via ``flow_module.schema.get_schema_id``.
+
+ Template and generic-workfile types are intentionally excluded: they
+ are publishable leaf assets that belong in the centre panel, not in
+ the tree.
+
+ On any error (e.g. framework not ready) an empty set is returned so
+ that the tree still loads without crashing.
+ """
+ if self._structural_type_ids is not None:
+ return self._structural_type_ids
+
+ try:
+ folder_id = self._flow_module.data.FOLDER_TYPE_ID
+ pipeline_step_id = self._flow_module.schema.get_schema_id(
+ self._flow_module.asset_management.PIPELINE_STEP_TYPE
+ )
+
+ self._structural_type_ids = {folder_id, pipeline_step_id}
+ self._app.log_debug(
+ f"MEDM Entity: structural type IDs = {self._structural_type_ids}"
+ )
+ except self._flow_module.FlowError as e:
+ self._app.log_warning(
+ f"MEDM Entity: could not resolve structural type IDs ({e}); "
+ "non-structural assets without structural descendants will be hidden."
+ )
+ self._structural_type_ids = set()
+
+ return self._structural_type_ids
+
+ def _is_tree_node(self, asset: Asset) -> bool:
+ """
+ Return ``True`` when *asset* should appear as a node in the left-hand
+ tree view.
+
+ An asset qualifies if **any** of the following conditions hold:
+
+ * it is a structural container (folder, container type, or pipeline
+ step) - these are always visible regardless of whether they have
+ children; **or**
+ * it has at least one direct child asset - workfiles can themselves
+ parent child workfiles, making them container nodes in the tree
+ even though they are not structural types.
+
+ Assets that satisfy none of the above are pure leaf items that belong
+ only in the centre-panel publish list, not in the tree.
+
+ :param asset: MEDM ``Asset`` to test.
+ :returns: ``True`` if the asset should appear in the tree.
+ """
+ if _is_structural_asset_util(asset, self._flow_module):
+ return True
+
+ # Non-structural: show in the tree only when the asset has direct
+ # children. Results are cached so each asset is fetched from the
+ # API at most once.
+ children = self._fetch_and_cache_children(asset)
+ return len(children) > 0
+
+ def _icon_for_asset(self, asset: Asset) -> QtGui.QIcon:
+ """
+ Return the appropriate tree icon for *asset* based on its type.
+
+ * **Structural container** (folder, container type, or pipeline-step)
+ -> folder icon.
+ * **Everything else** (workfiles, generic assets, ...) -> binary/data
+ icon, reflecting that the item holds or organises file data.
+
+ :param asset: MEDM ``Asset`` to pick an icon for.
+ :returns: A :class:`QtGui.QIcon` instance.
+ """
+ return (
+ self._folder_icon
+ if _is_structural_asset_util(asset, self._flow_module)
+ else self._binary_icon
+ )
+
+ def _load_medm_assets(self) -> None:
+ """
+ Load the first level of MEDM assets (project's immediate children).
+ Called asynchronously after a short delay to show the loading spinner.
+ """
+ if self._project is None:
+ self._app.log_warning(
+ "MEDM Entity: Cannot load assets - project not initialized"
+ )
+ self.data_refresh_fail.emit("Project not initialized")
+ return
+
+ try:
+ self._app.log_debug("MEDM: Loading entity tree (project children only)...")
+
+ count = 0
+ for asset in self._project.iterate_children():
+ if self._is_tree_node(asset):
+ self._add_asset_item(asset, None)
+ count += 1
+
+ self._app.log_debug(
+ f"MEDM: Entity tree loaded successfully. Loaded {count} root assets"
+ )
+ self.cache_loaded.emit()
+ self.data_refreshed.emit(True)
+
+ except Exception as e:
+ self._app.log_error(f"Failed to load MEDM data: {e}")
+ import traceback
+
+ self._app.log_debug(traceback.format_exc())
+ self.data_refresh_fail.emit(str(e))
+
+ def _add_asset_item(
+ self, asset: Asset, parent_item: Optional[QtGui.QStandardItem]
+ ) -> QtGui.QStandardItem:
+ """
+ Create a single ``QStandardItem`` for *asset* and append it to the tree.
+
+ The item is marked as *not* children-loaded so that ``hasChildren``
+ reports ``True`` and Qt draws an expand arrow until the user actually
+ drills in.
+
+ :param asset: The MEDM Asset to add.
+ :param parent_item: Parent item, or ``None`` for root level.
+ :returns: The newly created item.
+ """
+ asset_item = QtGui.QStandardItem(asset.name)
+ asset_item.setEditable(False)
+
+ asset_item.setData(asset, self.ASSET_ROLE)
+
+ sg_data = {
+ "type": asset.__class__.__name__,
+ "id": None,
+ "name": asset.name,
+ "code": asset.name,
+ }
+ asset_item.setData(sg_data, self.SG_DATA_ROLE)
+
+ # Mark children as not-yet-loaded so canFetchMore/hasChildren work.
+ asset_item.setData(False, self.CHILDREN_LOADED_ROLE)
+
+ asset_item.setIcon(self._icon_for_asset(asset))
+
+ if parent_item is None:
+ self.appendRow(asset_item)
+ else:
+ parent_item.appendRow(asset_item)
+
+ return asset_item
+
+ def _load_children_for_item(self, item: QtGui.QStandardItem) -> None:
+ """
+ Fetch the immediate children of *item* and add them to the tree.
+
+ ``CHILDREN_LOADED_ROLE`` is set to ``True`` **before** any rows are
+ inserted so that the ``rowsInserted`` signal (emitted by ``appendRow``)
+ cannot re-enter ``fetchMore`` for the same item.
+
+ :param item: The tree item whose children should be loaded.
+ """
+ # Mark loaded FIRST to prevent re-entrant fetchMore calls triggered by
+ # appendRow -> rowsInserted -> canFetchMore check on the same parent.
+ item.setData(True, self.CHILDREN_LOADED_ROLE)
+
+ asset = item.data(self.ASSET_ROLE)
+ if asset is None:
+ return
+
+ try:
+ children = self._fetch_and_cache_children(asset)
+ tree_children = [c for c in children if self._is_tree_node(c)]
+ for child_asset in tree_children:
+ self._add_asset_item(child_asset, item)
+ self._app.log_debug(
+ f"MEDM: Loaded {len(tree_children)}/{len(children)} children for '{asset.name}' "
+ f"(non-structural leaf children hidden from tree)"
+ )
+ except Exception as e:
+ self._app.log_debug(f"MEDM: Could not get children for '{asset.name}': {e}")
+
+ def _fetch_and_cache_children(self, asset: Asset) -> List[Asset]:
+ """
+ Return child assets for *asset*, fetching from the API only on the
+ first call and caching the result in the shared cache for subsequent
+ lookups by any MEDM model.
+ """
+ if asset.id in self._cache.children:
+ return self._cache.children[asset.id]
+
+ children = list(asset.iterate_children())
+ self._cache.children[asset.id] = children
+ return children
diff --git a/python/tk_multi_loader/medm/latestpublish_model.py b/python/tk_multi_loader/medm/latestpublish_model.py
new file mode 100644
index 0000000..2fd2357
--- /dev/null
+++ b/python/tk_multi_loader/medm/latestpublish_model.py
@@ -0,0 +1,730 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""MEDM Latest Publish Model - Replacement for SgLatestPublishModel
+
+This module provides a drop-in replacement for SgLatestPublishModel that uses
+Flow Asset Management (MEDM) data instead of Shotgun data.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+import sgtk
+from sgtk.platform.qt import QtCore, QtGui
+
+from .shared_cache import MedmSharedCache
+from .thumbnail_service import MedmThumbnailService
+from .utils import build_draft_sg_dict
+from .utils import is_structural_asset as _is_structural_asset_util
+
+if TYPE_CHECKING:
+ from flow.data import Asset
+ from flow.sandbox import DraftInfo
+else:
+ Asset = Any
+ DraftInfo = Any
+
+
+class MedmLatestPublishModel(QtGui.QStandardItemModel):
+ """
+ Model which handles the main spreadsheet view which displays the latest version of all
+ publishes from Flow Asset Management.
+
+ This is a drop-in replacement for SgLatestPublishModel that uses MEDM data.
+ """
+
+ # Matches the V1 FlowActions hook constant. Draft rows carry this as
+ # version_number so action hooks route them to asset_management.open_draft().
+ DRAFT_VERSION_IDENTIFIER = -1
+
+ # Sentinel key used inside cache.drafts to store the list returned by
+ # get_drafts(draft_type="new"). A real asset.id is always a UUID/storage
+ # key string; this double-underscore key cannot collide with a real ID.
+ _NEW_DRAFTS_CACHE_KEY = "__new_asset_drafts__"
+
+ # Custom roles - matching the original model's interface
+ TYPE_ID_ROLE = QtCore.Qt.UserRole + 101
+ IS_FOLDER_ROLE = QtCore.Qt.UserRole + 102
+ ASSOCIATED_TREE_VIEW_ITEM_ROLE = QtCore.Qt.UserRole + 103
+ PUBLISH_TYPE_NAME_ROLE = QtCore.Qt.UserRole + 104
+ SEARCHABLE_NAME = QtCore.Qt.UserRole + 105
+
+ # Additional MEDM-specific roles
+ SG_DATA_ROLE = QtCore.Qt.UserRole + 1 # To maintain compatibility with ShotgunModel
+ SG_ASSOCIATED_FIELD_ROLE = QtCore.Qt.UserRole + 2
+ ASSET_ROLE = (
+ QtCore.Qt.UserRole + 200
+ ) # Stores MEDM Asset object (shared with all MEDM models)
+ DRAFT_ROLE = (
+ QtCore.Qt.UserRole + 202
+ ) # Stores DraftInfo for draft rows (shared with history model)
+
+ # Signals
+ loadingStarted = QtCore.Signal()
+ loadingFinished = QtCore.Signal()
+ loadingError = QtCore.Signal(str)
+ cache_loaded = QtCore.Signal()
+ data_refreshed = QtCore.Signal(bool) # Argument: data_changed
+ query_changed = QtCore.Signal() # For overlay widget compatibility
+ data_refreshing = QtCore.Signal() # For overlay widget compatibility
+ data_refresh_fail = QtCore.Signal(str) # For overlay widget compatibility
+
+ def __init__(
+ self,
+ parent: Optional[QtCore.QObject],
+ publish_type_model,
+ bg_task_manager,
+ cache: Optional[MedmSharedCache] = None,
+ thumbnail_service: Optional[MedmThumbnailService] = None,
+ ):
+ """
+ Model which represents the latest publishes for an entity from MEDM.
+
+ :param parent: Parent QObject
+ :param publish_type_model: Model for tracking publish types
+ :param bg_task_manager: Background task manager (kept for API compatibility)
+ :param cache: Shared :class:`MedmSharedCache`. When provided all data
+ caches (children, drafts, publish types) are shared with other MEDM
+ models so no duplicate API calls are made. When *None* a private
+ cache is created for standalone use.
+ :param thumbnail_service: Shared :class:`MedmThumbnailService` instance.
+ When provided (the normal case) thumbnail downloads are shared with
+ other models. When *None* a private service is created.
+ """
+ super().__init__(parent)
+
+ self._app = sgtk.platform.current_bundle()
+ self._flow_module = sgtk.platform.import_framework(
+ "tk-framework-flowam", "flow"
+ )
+
+ self._publish_type_model = publish_type_model
+ self._bg_task_manager = bg_task_manager
+ self._cache = cache if cache is not None else MedmSharedCache()
+ self._thumbnail_service = thumbnail_service or MedmThumbnailService(
+ self._cache, self
+ )
+ self._owns_thumbnail_service = thumbnail_service is None
+
+ self._loading_icon = QtGui.QIcon.fromTheme("view-refresh")
+ self._publish_icon = QtGui.QIcon.fromTheme("document")
+
+ self._current_entity = None
+
+ self._project_id = 0
+ self._project_name = "MEDM Project"
+ self._initialize_project_info()
+
+ # -------------------------------------------------------------------------
+ # Public API - Called by dialog.py and other external code
+ # -------------------------------------------------------------------------
+
+ def destroy(self) -> None:
+ """Clean up model resources."""
+ self._cache.drafts.clear()
+ if self._owns_thumbnail_service:
+ self._thumbnail_service.destroy()
+
+ def load_data(self, item: Optional[QtGui.QStandardItem]) -> None:
+ """
+ Clears the model and sets it up for the selected asset from left treeview panel.
+ Loads data from MEDM instead of Shotgun.
+
+ :param item: Selected item in the treeview, None if nothing is selected.
+ """
+ self.loadingStarted.emit()
+
+ try:
+ self._current_entity = item
+
+ self.layoutAboutToBeChanged.emit()
+ self.removeRows(0, self.rowCount())
+ self._populate_model_from_selected_item(item)
+ self.layoutChanged.emit()
+
+ self.loadingFinished.emit()
+ self.cache_loaded.emit()
+
+ except Exception as e:
+ self.loadingError.emit(str(e))
+
+ def async_refresh(self) -> None:
+ """
+ Refresh the current data set.
+
+ Clears volatile shared caches (drafts) so fresh state is fetched from
+ the server. Thumbnail and publish-type caches are intentionally kept:
+ revision IDs are immutable so those will always return the same data.
+ """
+ self._cache.clear_on_refresh()
+ if self._current_entity is not None:
+ self.load_data(self._current_entity)
+ self.data_refreshed.emit(True)
+
+ # -------------------------------------------------------------------------
+ # Private utility methods - Internal implementation details
+ # -------------------------------------------------------------------------
+
+ def _initialize_project_info(self) -> None:
+ """Initialize project ID and name from the session project."""
+ try:
+ self._project_id = self._app.context.project["id"]
+ self._project_name = self._app.context.project["name"]
+ except Exception as e:
+ self._app.log_warning(
+ f"MEDM LatestPublish: Failed to initialize project info: {type(e).__name__}: {e}. "
+ "Using default project values. This may indicate the session project was not "
+ "initialized or the project ID is invalid."
+ )
+
+ def _populate_model_from_selected_item(
+ self, selected_item: Optional[QtGui.QStandardItem]
+ ) -> None:
+ """
+ Populate the model with latest versions of child assets from the selected tree item.
+
+ Extracts the Asset from the selected tree view item, fetches latest versions
+ of all its child assets, and populates the model with them as Qt items.
+
+ :param selected_item: The selected asset item from the tree view (or None)
+ """
+ if selected_item is None:
+ return
+
+ asset = self._extract_asset_from_tree_item(selected_item)
+ if asset is None:
+ self._app.log_warning("MEDM: Could not extract asset from selected item")
+ return
+
+ self._app.log_debug(f"MEDM: Asset extracted: {asset.name}")
+
+ children_asset_sg_dicts = self._fetch_asset_children(asset)
+
+ # If the selected asset has no (non-structural) children it may itself
+ # be the publishable leaf. Only fall back to showing it directly when
+ # the asset is NOT a structural container.
+ leaf_asset_fallback = None
+ if not children_asset_sg_dicts and not _is_structural_asset_util(
+ asset, self._flow_module
+ ):
+ try:
+ children_asset_sg_dicts = [self._asset_to_sg_dict(asset)]
+ except Exception as e:
+ self._app.log_warning(
+ f"MEDM: Could not convert leaf asset '{asset.name}' to sg_dict: {e}"
+ )
+ # Keep a reference to the raw asset so we can still fetch its
+ # drafts below.
+ leaf_asset_fallback = asset
+
+ self._app.log_debug(
+ f"MEDM: Fetched {len(children_asset_sg_dicts)} latest version dicts from children"
+ )
+
+ assets_for_draft_lookup = [
+ sg_dict.get("_medm_asset") for sg_dict in children_asset_sg_dicts
+ ]
+ assets_for_draft_lookup = [a for a in assets_for_draft_lookup if a is not None]
+ if not assets_for_draft_lookup and leaf_asset_fallback is not None:
+ assets_for_draft_lookup = [leaf_asset_fallback]
+
+ # --- Pass 1: collect draft cards per asset ----------------------------
+ drafts_by_asset_id: Dict[str, list] = {}
+ for child_asset in assets_for_draft_lookup:
+ try:
+ if child_asset.id in self._cache.drafts:
+ raw_drafts = self._cache.drafts[child_asset.id]
+ else:
+ raw_drafts = self._flow_module.asset_management.get_asset_drafts(
+ child_asset.id
+ )
+ self._cache.drafts[child_asset.id] = raw_drafts
+ except Exception as e:
+ self._app.log_debug(
+ f"MEDM: Could not fetch drafts for '{child_asset.name}': {e}"
+ )
+ continue
+
+ draft_dicts = []
+ for draft_info in raw_drafts:
+ try:
+ draft_dicts.append(self._draft_to_sg_dict(draft_info, child_asset))
+ self._app.log_debug(
+ f"MEDM: Found draft '{getattr(draft_info, 'name', '?')}' "
+ f"for asset '{child_asset.name}'"
+ )
+ except Exception as e:
+ self._app.log_warning(
+ f"MEDM: Could not convert draft '{getattr(draft_info, 'name', '?')}' "
+ f"for '{child_asset.name}': {e}"
+ )
+
+ if draft_dicts:
+ drafts_by_asset_id[child_asset.id] = draft_dicts
+
+ # --- Pass 2: populate the center panel --------------------------------
+ # - Asset has a draft -> show only the draft card(s).
+ # - Asset has no draft -> show its latest published version card.
+ draft_count = 0
+ published_count = 0
+
+ for sg_dict in children_asset_sg_dicts:
+ medm_asset = sg_dict.get("_medm_asset")
+ asset_id = medm_asset.id if medm_asset is not None else None
+
+ if asset_id is not None and asset_id in drafts_by_asset_id:
+ for draft_sg_dict in drafts_by_asset_id[asset_id]:
+ self._add_sg_dict_as_qt_item(draft_sg_dict)
+ draft_count += 1
+ else:
+ self._add_sg_dict_as_qt_item(sg_dict)
+ published_count += 1
+
+ # Surface drafts for the leaf-fallback asset.
+ if (
+ leaf_asset_fallback is not None
+ and leaf_asset_fallback.id in drafts_by_asset_id
+ ):
+ for draft_sg_dict in drafts_by_asset_id[leaf_asset_fallback.id]:
+ self._add_sg_dict_as_qt_item(draft_sg_dict)
+ draft_count += 1
+
+ # --- Pass 3: surface NewDraftInfo entries for unpublished child assets
+ for new_draft_sg_dict in self._fetch_new_draft_items_for_parent(asset.id):
+ self._add_sg_dict_as_qt_item(new_draft_sg_dict)
+ draft_count += 1
+
+ self._app.log_debug(
+ f"MEDM: center panel now has {self.rowCount()} items "
+ f"({draft_count} draft(s), {published_count} published)"
+ )
+
+ sg_publish_type_counts = self._calculate_sg_publish_type_counts()
+ self._publish_type_model.set_active_types(sg_publish_type_counts)
+
+ def _extract_asset_from_tree_item(
+ self, item: QtGui.QStandardItem
+ ) -> Optional[Asset]:
+ """
+ Extract the MEDM Asset object from a tree view QStandardItem.
+
+ :param item: The QStandardItem from the entity tree (left panel)
+ :returns: MEDM Asset object or None if not found
+ """
+ # Both MedmEntityModel and MedmLatestPublishModel use ASSET_ROLE = Qt.UserRole + 200
+ asset = item.data(self.ASSET_ROLE)
+ if asset:
+ return asset
+
+ asset_data = item.data(QtCore.Qt.UserRole + 1)
+ return asset_data
+
+ def _fetch_asset_children(self, asset: Asset) -> List[Dict[str, Any]]:
+ """
+ Fetch all non-structural child assets and convert to sg_data dicts.
+
+ Structural containers (folders, pipeline steps, container types) are
+ filtered out here because they belong only in the left-hand tree, not
+ in the center panel publish list.
+
+ The shared ``cache.children`` dict is consulted first so that selecting
+ a tree node never duplicates an API call that was already made when the
+ node was expanded by :class:`MedmEntityModel` (or vice-versa).
+
+ :param asset: The selected MEDM Asset in MEDM treeview
+ :returns: List of sg_data dictionaries representing each non-structural child asset
+ """
+ children_asset_sg_dicts = []
+
+ try:
+ if asset.id in self._cache.children:
+ child_assets = self._cache.children[asset.id]
+ else:
+ child_assets = list(asset.iterate_children())
+ self._cache.children[asset.id] = child_assets
+
+ for child_asset in child_assets:
+ if _is_structural_asset_util(child_asset, self._flow_module):
+ self._app.log_debug(
+ f"MEDM: Skipping structural asset '{child_asset.name}' from center panel"
+ )
+ continue
+ try:
+ asset_dict = self._asset_to_sg_dict(child_asset)
+ children_asset_sg_dicts.append(asset_dict)
+ self._app.log_debug(
+ f"MEDM: Added asset '{child_asset.name}' with latest version "
+ f"v{child_asset.version_number}"
+ )
+ except Exception as e:
+ self._app.log_warning(
+ f"MEDM: Error processing child asset '{child_asset.name}': {e}"
+ )
+ continue
+
+ except Exception as e:
+ self._app.log_warning(f"MEDM: Error fetching asset children: {e}")
+
+ self._app.log_debug(
+ f"MEDM: Loaded {len(children_asset_sg_dicts)} children assets for asset '{asset.name}'"
+ )
+ return children_asset_sg_dicts
+
+ def _asset_to_sg_dict(self, asset: Asset) -> Dict[str, Any]:
+ """
+ Convert an MEDM Asset to a Shotgun-compatible dictionary.
+
+ :param asset: The MEDM Asset
+ :returns: Dictionary with Shotgun-compatible fields
+ """
+ sg_publish_type_id = None
+ sg_publish_type_code = "MEDM Asset"
+
+ medm_type_ids = asset.type_ids
+ if len(medm_type_ids) > 0:
+ sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(
+ medm_type_ids[0]
+ )
+
+ sg_dict = {
+ "id": None,
+ "type": "PublishedFile",
+ "code": asset.name,
+ "name": asset.name,
+ "version_number": asset.version_number,
+ "description": asset.description or "",
+ "created_at": asset.created_at,
+ "created_by": {
+ "type": "HumanUser",
+ "id": 1,
+ "name": asset.created_by or "MEDM User",
+ },
+ "entity": {"type": "Asset", "id": None, "name": asset.name},
+ "project": {
+ "type": "Project",
+ "id": self._project_id,
+ "name": self._project_name,
+ },
+ "task": None,
+ "task_uniqueness": True,
+ "published_file_type": {
+ "type": "PublishedFileType",
+ "id": sg_publish_type_id,
+ "name": sg_publish_type_code,
+ },
+ "tank_type": {
+ "type": "TankType",
+ "id": sg_publish_type_id,
+ "name": sg_publish_type_code,
+ },
+ "path": {"local_path": ""},
+ "image": None,
+ # Thumbnail is resolved in a background thread via this revision ID.
+ "_thumbnail_revision_id": asset.revision_id,
+ "sg_flow_revision_id": asset.revision_id,
+ "_medm_asset": asset,
+ }
+
+ return sg_dict
+
+ def _draft_to_sg_dict(
+ self, draft_info: DraftInfo, asset: Optional[Asset] = None
+ ) -> Dict[str, Any]:
+ """
+ Convert a local DraftInfo into a Shotgun-compatible dictionary suitable
+ for display as a center-panel card.
+
+ Key conventions that V1 action hooks rely on:
+ - ``version_number == DRAFT_VERSION_IDENTIFIER (-1)`` - identifies a local draft
+ - ``sg_flow_revision_id`` - the draft's sandbox ID (draft_info.draft_id), used
+ by asset_management.open_draft() and sandbox.is_local_draft()
+
+ :param draft_info: DraftInfo returned by asset_management.get_asset_drafts()
+ (CheckoutDraftInfo) or get_drafts() (NewDraftInfo).
+ :param asset: The MEDM Asset the draft belongs to. May be ``None`` for
+ ``NewDraftInfo`` entries whose parent asset has not been published yet.
+ :returns: sg_data dictionary compatible with action hooks and center panel UI.
+ """
+ sg_publish_type_id = None
+ sg_publish_type_code = "MEDM Asset"
+
+ # Prefer the published asset's type_ids; fall back to the draft's own
+ # type_ids for NewDraftInfo where no published asset exists yet.
+ type_ids = []
+ if asset is not None:
+ type_ids = getattr(asset, "type_ids", None) or []
+ if not type_ids:
+ type_ids = getattr(draft_info, "type_ids", None) or []
+ if type_ids:
+ sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(
+ type_ids[0]
+ )
+
+ # For checkout drafts the parent asset's revision_id provides the
+ # thumbnail; for new drafts there is no published revision.
+ draft_type = getattr(draft_info, "draft_type", "unknown")
+ thumb_revision_id = (
+ asset.revision_id
+ if (draft_type == "checkout" and asset is not None)
+ else None
+ )
+
+ sg_dict = build_draft_sg_dict(
+ draft_info,
+ asset,
+ self._app.context,
+ self._project_id,
+ self._project_name,
+ sg_publish_type_id,
+ sg_publish_type_code,
+ )
+ sg_dict.update(
+ {
+ "task_uniqueness": True,
+ "tank_type": {
+ "type": "TankType",
+ "id": sg_publish_type_id,
+ "name": sg_publish_type_code,
+ },
+ "_thumbnail_revision_id": thumb_revision_id,
+ }
+ )
+ return sg_dict
+
+ def _fetch_new_draft_items_for_parent(
+ self, parent_asset_id: str
+ ) -> List[Dict[str, Any]]:
+ """
+ Return sg_dict items for ``NewDraftInfo`` entries whose ``parent_id``
+ matches *parent_asset_id*.
+
+ These represent brand-new assets that exist only on disk and have never
+ been published to MEDM. Because they have no published asset record
+ they are invisible to ``asset.iterate_children()`` and must be surfaced
+ via :func:`~flow.asset_management.get_drafts`.
+
+ The full list returned by ``get_drafts(draft_type="new")`` is cached
+ under :attr:`_NEW_DRAFTS_CACHE_KEY` in ``cache.drafts`` for the
+ lifetime of a single load cycle (cleared by
+ :meth:`~MedmSharedCache.clear_on_refresh`).
+
+ :param parent_asset_id: ``asset.id`` of the tree item currently selected.
+ :returns: List of sg_data dicts ready for :meth:`_add_sg_dict_as_qt_item`.
+ """
+ if self._NEW_DRAFTS_CACHE_KEY not in self._cache.drafts:
+ try:
+ all_new_drafts = self._flow_module.asset_management.get_drafts(
+ draft_type="new"
+ )
+ except Exception as e:
+ self._app.log_warning(f"MEDM: Could not fetch new-asset drafts: {e}")
+ all_new_drafts = []
+ self._cache.drafts[self._NEW_DRAFTS_CACHE_KEY] = all_new_drafts
+ else:
+ all_new_drafts = self._cache.drafts[self._NEW_DRAFTS_CACHE_KEY]
+
+ result = []
+ for draft_info in all_new_drafts:
+ if getattr(draft_info, "parent_id", None) != parent_asset_id:
+ continue
+ try:
+ draft_sg_dict = self._draft_to_sg_dict(draft_info, asset=None)
+ self._app.log_debug(
+ f"MEDM: Found new-asset draft '{getattr(draft_info, 'name', '?')}' "
+ f"under parent '{parent_asset_id}'"
+ )
+ result.append(draft_sg_dict)
+ except Exception as e:
+ self._app.log_warning(
+ f"MEDM: Could not convert new-asset draft "
+ f"'{getattr(draft_info, 'name', '?')}': {e}"
+ )
+ return result
+
+ def _resolve_publish_type(self, medm_type_id_str: str) -> tuple:
+ """
+ Resolve a MEDM schema type ID to a ``(sg_publish_type_id, display_name)`` pair.
+
+ Resolution order:
+ 1. In-process cache (avoids redundant SG round-trips within the same load).
+ 2. ShotGrid ``PublishedFileType`` lookup by display name (real ID).
+ 3. Hash-based integer fallback when no SG record exists.
+
+ :param medm_type_id_str: MEDM schema type ID string.
+ :returns: Tuple of (integer publish type id, human-readable display name).
+ """
+ if medm_type_id_str in self._cache.publish_types:
+ return self._cache.publish_types[medm_type_id_str]
+
+ display_name = medm_type_id_str
+ try:
+ schema_name = self._flow_module.schema.get_schema_display_name(
+ medm_type_id_str
+ )
+ if schema_name:
+ display_name = schema_name
+ except Exception as e:
+ self._app.log_debug(
+ f"MEDM: Could not get schema display name for '{medm_type_id_str}': {e}"
+ )
+
+ sg_publish_type_id = None
+ try:
+ pft = self._app.shotgun.find_one(
+ "PublishedFileType",
+ [["code", "is", display_name]],
+ ["id", "code"],
+ )
+ if pft:
+ sg_publish_type_id = pft["id"]
+ self._app.log_debug(
+ f"MEDM: Resolved PublishedFileType '{display_name}' "
+ f"-> SG id={sg_publish_type_id}"
+ )
+ else:
+ self._app.log_debug(
+ f"MEDM: No SG PublishedFileType found for '{display_name}', "
+ f"item will bypass type filter"
+ )
+ except Exception as e:
+ self._app.log_debug(
+ f"MEDM: Could not look up PublishedFileType for '{display_name}': {e}"
+ )
+
+ result = (sg_publish_type_id, display_name)
+ self._cache.publish_types[medm_type_id_str] = result
+ return result
+
+ def _add_sg_dict_as_qt_item(self, sg_item: Dict[str, Any]) -> None:
+ """
+ Create a QStandardItem from a Shotgun-compatible dict and add it to the model.
+
+ The QStandardItem stores multiple pieces of data in custom Qt roles:
+ - SG_DATA_ROLE: Full Shotgun-compatible dict for backwards compatibility
+ - ASSET_ROLE: Original MEDM Asset object (from "_medm_asset" key)
+ - TYPE_ID_ROLE: Publish type ID for filtering
+ - PUBLISH_TYPE_NAME_ROLE: Publish type name for display
+ - SEARCHABLE_NAME: Name for search/filter operations
+
+ The item is then appended to the model as a new row, making it visible in the
+ center panel's publish view.
+
+ :param sg_item: Shotgun-compatible dictionary created by _asset_to_sg_dict().
+ Must contain "_medm_asset" key with the original MEDM Asset object.
+ """
+ qt_item = QtGui.QStandardItem(sg_item.get("code", "Unnamed"))
+
+ qt_item.setData(sg_item, self.SG_DATA_ROLE)
+ qt_item.setData(sg_item.get("code", ""), self.SEARCHABLE_NAME)
+ qt_item.setData(False, self.IS_FOLDER_ROLE)
+ pft = sg_item.get("published_file_type") or {}
+ qt_item.setData(pft.get("id"), self.TYPE_ID_ROLE)
+ qt_item.setData(pft.get("name"), self.PUBLISH_TYPE_NAME_ROLE)
+
+ if "_medm_asset" in sg_item:
+ qt_item.setData(sg_item["_medm_asset"], self.ASSET_ROLE)
+ if "_medm_draft" in sg_item:
+ qt_item.setData(sg_item["_medm_draft"], self.DRAFT_ROLE)
+
+ qt_item.setEditable(False)
+ qt_item.setIcon(self._publish_icon)
+
+ revision_id = sg_item.get("_thumbnail_revision_id")
+ if revision_id:
+ self._resolve_and_download_thumbnail(qt_item, revision_id)
+
+ self._set_tooltip(qt_item, sg_item)
+
+ def get_sg_data():
+ return qt_item.data(self.SG_DATA_ROLE)
+
+ qt_item.get_sg_data = get_sg_data
+
+ self.appendRow(qt_item)
+
+ self._app.log_debug(
+ f"MEDM: Added item '{qt_item.text()}' to model (row count: {self.rowCount()})"
+ )
+
+ def _set_tooltip(self, item: QtGui.QStandardItem, sg_item: Dict[str, Any]) -> None:
+ """
+ Sets a tooltip for a publish item.
+
+ :param item: QStandardItem associated with the publish
+ :param sg_item: Publish information dictionary
+ """
+ tooltip = f"Name: {sg_item.get('code', 'No name given.')}"
+
+ # Version info - drafts use DRAFT_VERSION_IDENTIFIER (-1) internally;
+ # show a human-readable label instead of the raw sentinel value.
+ version = sg_item.get("version_number")
+ if version == self.DRAFT_VERSION_IDENTIFIER:
+ draft_type = sg_item.get("_medm_draft_type", "")
+ vers_str = f"Draft ({draft_type})" if draft_type else "Draft"
+ elif version is not None and version >= 0:
+ vers_str = f"{version:03d}"
+ else:
+ vers_str = "N/A"
+
+ author_str = sg_item.get("created_by", {}).get("name", "Unknown")
+ date_str = str(sg_item.get("created_at", "Unknown"))
+
+ tooltip += f"
Version: {vers_str} by {author_str} at {date_str}"
+ tooltip += f"
Description: {sg_item.get('description', 'No description given.')}"
+
+ item.setToolTip(tooltip)
+
+ def _resolve_and_download_thumbnail(
+ self, qt_item: QtGui.QStandardItem, revision_id: str
+ ) -> None:
+ """
+ Delegate thumbnail resolution and download to the shared
+ :class:`MedmThumbnailService`. The service calls :meth:`_apply_thumbnail`
+ on the main thread once the image bytes are available.
+
+ :param qt_item: The QStandardItem to set the thumbnail on.
+ :param revision_id: MEDM AssetRevision ID whose thumbnail is needed.
+ """
+ self._thumbnail_service.request(qt_item, revision_id, self._apply_thumbnail)
+
+ def _apply_thumbnail(self, qt_item: QtGui.QStandardItem, image_data: bytes) -> None:
+ """
+ Apply downloaded image bytes as a scaled icon on *qt_item*.
+ Called on the main thread by :class:`MedmThumbnailService`.
+
+ :param qt_item: The QStandardItem to update.
+ :param image_data: Raw image bytes.
+ """
+ pixmap = QtGui.QPixmap()
+ if pixmap.loadFromData(image_data):
+ scaled = pixmap.scaled(
+ 512, 400, QtCore.Qt.KeepAspectRatio, QtCore.Qt.SmoothTransformation
+ )
+ qt_item.setIcon(QtGui.QIcon(scaled))
+
+ def _calculate_sg_publish_type_counts(self) -> Dict[int, int]:
+ """
+ Count how many items in the model have each Shotgun PublishedFileType.
+
+ :returns: Dictionary mapping sg_publish_type_id to item count
+ """
+ sg_publish_type_aggregates = defaultdict(int)
+
+ for row in range(self.rowCount()):
+ item = self.item(row, 0)
+ if item and not item.data(self.IS_FOLDER_ROLE):
+ sg_publish_type_id = item.data(self.TYPE_ID_ROLE)
+ if sg_publish_type_id is not None:
+ sg_publish_type_aggregates[sg_publish_type_id] += 1
+
+ return sg_publish_type_aggregates
diff --git a/python/tk_multi_loader/medm/publishhistory_model.py b/python/tk_multi_loader/medm/publishhistory_model.py
new file mode 100644
index 0000000..ee735ee
--- /dev/null
+++ b/python/tk_multi_loader/medm/publishhistory_model.py
@@ -0,0 +1,507 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""
+MEDM Publish History Model - Shows all revisions for a selected MEDM entity.
+
+This module provides a model that displays all revisions (version history) for
+a selected MEDM entity, similar to SgPublishHistoryModel for Shotgun data.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+import sgtk
+from sgtk.platform.qt import QtCore, QtGui
+
+from .. import utils
+from .shared_cache import MedmSharedCache
+from .thumbnail_service import MedmThumbnailService
+from .utils import build_draft_sg_dict
+
+if TYPE_CHECKING:
+ from adsk.flow.am import (
+ Asset,
+ AssetRevision,
+ AssetVersion,
+ )
+ from flow.sandbox import CheckoutDraftInfo, DraftInfo, NewDraftInfo
+else:
+ Asset = Any
+ AssetRevision = Any
+ AssetVersion = Any
+ CheckoutDraftInfo = Any
+ DraftInfo = Any
+ NewDraftInfo = Any
+
+
+class MedmPublishHistoryModel(QtGui.QStandardItemModel):
+ """
+ Model that displays the version history (all revisions) for an MEDM entity.
+
+ This is the MEDM equivalent of SgPublishHistoryModel.
+ """
+
+ # Matches the V1 FlowActions hook constant so action hooks correctly identify
+ # draft rows (version_number == -1 -> local draft; version_number > -1 -> published).
+ DRAFT_VERSION_IDENTIFIER = -1
+
+ # Custom roles - matching SgPublishHistoryModel interface
+ USER_THUMB_ROLE = QtCore.Qt.UserRole + 101
+ PUBLISH_THUMB_ROLE = QtCore.Qt.UserRole + 102
+ FULL_IMAGE_PATH_ROLE = QtCore.Qt.UserRole + 103
+
+ # MEDM-specific roles
+ SG_DATA_ROLE = QtCore.Qt.UserRole + 1 # To maintain compatibility with ShotgunModel
+ ASSET_ROLE = (
+ QtCore.Qt.UserRole + 200
+ ) # Stores MEDM Asset object (shared with all MEDM models)
+ VERSION_ROLE = QtCore.Qt.UserRole + 201 # Stores MEDM AssetVersion object
+ DRAFT_ROLE = QtCore.Qt.UserRole + 202 # Stores DraftInfo for draft rows
+
+ # Signals for compatibility with ShotgunModelOverlayWidget
+ cache_loaded = QtCore.Signal()
+ data_refreshed = QtCore.Signal(bool) # Argument: data_changed
+ query_changed = QtCore.Signal()
+ data_refreshing = QtCore.Signal()
+ data_refresh_fail = QtCore.Signal(str)
+
+ def __init__(
+ self,
+ parent,
+ bg_task_manager,
+ cache: Optional[MedmSharedCache] = None,
+ thumbnail_service: Optional[MedmThumbnailService] = None,
+ ):
+ """
+ Constructor
+
+ :param parent: Parent QObject
+ :param bg_task_manager: Background task manager (kept for API compatibility)
+ :param cache: Shared :class:`MedmSharedCache`. When provided all data
+ caches (drafts, versions, publish types) are shared with other MEDM
+ models. When *None* a private cache is created for standalone use.
+ :param thumbnail_service: Shared :class:`MedmThumbnailService` instance.
+ When provided thumbnail downloads are shared with other models.
+ When *None* a private service is created.
+ """
+ super().__init__(parent)
+
+ self._app = sgtk.platform.current_bundle()
+ self._flow_module = sgtk.platform.import_framework(
+ "tk-framework-flowam", "flow"
+ )
+
+ self._bg_task_manager = bg_task_manager
+ self._cache = cache if cache is not None else MedmSharedCache()
+ self._thumbnail_service = thumbnail_service or MedmThumbnailService(
+ self._cache, self
+ )
+ self._owns_thumbnail_service = thumbnail_service is None
+ self._loading_icon = QtGui.QPixmap(":/res/loading_100x100.png")
+ self._current_sg_data = None
+
+ self._project_id = 0
+ self._project_name = "MEDM Project"
+ self._initialize_project_info()
+
+ # -------------------------------------------------------------------------
+ # Public API - Called by dialog.py and other external code
+ # -------------------------------------------------------------------------
+
+ def destroy(self) -> None:
+ """Clean up model resources."""
+ self._cache.drafts.clear()
+ self._cache.versions.clear()
+ if self._owns_thumbnail_service:
+ self._thumbnail_service.destroy()
+
+ def load_data(self, sg_data: Dict[str, Any]) -> None:
+ """
+ Load and display all versions (version history) for the selected asset.
+
+ This populates the history panel with all versions of the asset
+ selected in the center panel, ordered by version number (newest first).
+
+ :param sg_data: Shotgun-compatible data dict from center panel selection.
+ Must contain "_medm_asset" field with the MEDM Asset.
+ """
+ self.clear()
+ self._current_sg_data = sg_data
+
+ medm_asset = sg_data.get("_medm_asset")
+ if medm_asset is None:
+ # A NewDraftInfo card has no published asset yet. Show the draft
+ # itself as the sole history entry so the user can still open it.
+ draft_info = sg_data.get("_medm_draft")
+ if (
+ draft_info is not None
+ and getattr(draft_info, "draft_type", None) == "new"
+ ):
+ self._add_draft_as_qt_item(draft_info, asset=None)
+ self.cache_loaded.emit()
+ self.data_refreshed.emit(True)
+ else:
+ self._app.log_warning(
+ "MEDM History: No asset found in selected publish asset sg_data dict"
+ )
+ self.data_refresh_fail.emit("No asset found in selection")
+ return
+
+ try:
+ asset_id = medm_asset.id
+ if asset_id in self._cache.versions:
+ versions = self._cache.versions[asset_id]
+ else:
+ versions = list(medm_asset.iterate_versions())
+ self._cache.versions[asset_id] = versions
+
+ for asset_version in versions:
+ self._add_version_as_qt_item(asset_version, medm_asset)
+
+ self._app.log_debug(
+ f"MEDM History: Loaded {len(versions)} published versions"
+ )
+
+ try:
+ if asset_id in self._cache.drafts:
+ drafts = self._cache.drafts[asset_id]
+ else:
+ drafts = self._flow_module.asset_management.get_asset_drafts(
+ asset_id
+ )
+ self._cache.drafts[asset_id] = drafts
+ for draft_info in drafts:
+ self._add_draft_as_qt_item(draft_info, medm_asset)
+ if drafts:
+ self._app.log_debug(
+ f"MEDM History: Added {len(drafts)} draft(s) for asset '{medm_asset.name}'"
+ )
+ except Exception as e:
+ # Drafts are optional - a failure here should not prevent published
+ # versions from being shown.
+ self._app.log_warning(f"MEDM History: Could not fetch drafts: {e}")
+
+ self.cache_loaded.emit()
+ self.data_refreshed.emit(True)
+
+ except Exception as e:
+ self._app.log_error(f"MEDM History: Error loading versions: {e}")
+ self.data_refresh_fail.emit(str(e))
+
+ def async_refresh(self) -> None:
+ """
+ Refresh the current data set.
+
+ Clears volatile shared caches (versions, drafts) so that fresh data is
+ fetched. Thumbnail and publish-type caches are intentionally kept:
+ revision IDs are immutable so those will always return the same data.
+ """
+ self._cache.clear_on_refresh()
+ if self._current_sg_data is not None:
+ self.load_data(self._current_sg_data)
+ self.data_refreshed.emit(True)
+
+ def hard_refresh(self) -> None:
+ """Force refresh of data (same as async_refresh for MEDM)."""
+ self.async_refresh()
+
+ # -------------------------------------------------------------------------
+ # Private utility methods - Internal implementation details
+ # -------------------------------------------------------------------------
+
+ def _initialize_project_info(self) -> None:
+ """Initialize project ID and name from the session project."""
+ try:
+ self._project_id = self._app.context.project["id"]
+ self._project_name = self._app.context.project["name"]
+ except Exception as e:
+ self._app.log_warning(
+ f"MEDM History: Failed to initialize project info: {type(e).__name__}: {e}. "
+ "Using default project values. This may indicate the session project was not "
+ "initialized or the project ID is invalid."
+ )
+
+ def _add_version_as_qt_item(
+ self, asset_version: AssetVersion, asset: Asset
+ ) -> None:
+ """
+ Convert an AssetVersion to a QStandardItem and add it to the history model.
+
+ :param asset_version: The MEDM AssetVersion to add
+ :param asset: The parent MEDM Asset
+ """
+ version_number = asset_version.version_number
+
+ qt_item = QtGui.QStandardItem(f"{version_number:03d}")
+ qt_item.setEditable(False)
+
+ sg_data = self._version_to_sg_dict(asset_version, asset)
+ qt_item.setData(sg_data, self.SG_DATA_ROLE)
+
+ qt_item.setData(asset, self.ASSET_ROLE)
+ qt_item.setData(asset_version, self.VERSION_ROLE)
+
+ qt_item.setData(self._loading_icon, self.PUBLISH_THUMB_ROLE)
+ thumb = utils.create_overlayed_user_publish_thumbnail(
+ qt_item.data(self.PUBLISH_THUMB_ROLE), None
+ )
+ qt_item.setIcon(QtGui.QIcon(thumb))
+
+ revision_id = sg_data.get("_thumbnail_revision_id")
+ if revision_id:
+ self._resolve_and_download_thumbnail(qt_item, revision_id)
+
+ def get_sg_data():
+ return qt_item.data(self.SG_DATA_ROLE)
+
+ qt_item.get_sg_data = get_sg_data
+
+ self.appendRow(qt_item)
+ self._app.log_debug(f"MEDM History: Added version v{version_number}")
+
+ def _version_to_sg_dict(
+ self, version: AssetVersion, asset: Asset
+ ) -> Dict[str, Any]:
+ """
+ Convert a MEDM AssetVersion to Shotgun-compatible dictionary.
+
+ :param version: The MEDM AssetVersion
+ :param asset: The parent MEDM Asset
+ :returns: sg_data dictionary compatible with Shotgun UI
+ """
+ sg_publish_type_id = None
+ sg_publish_type_code = "MEDM Asset"
+
+ medm_type_ids = asset.type_ids
+ if medm_type_ids:
+ sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(
+ medm_type_ids[0]
+ )
+
+ asset_revision = version.revision
+ description = asset_revision.comment or ""
+
+ sg_dict = {
+ "id": None,
+ "type": "PublishedFile",
+ "code": asset_revision.name,
+ "name": asset_revision.name,
+ "version_number": version.version_number,
+ "description": description,
+ "created_at": version.created_at,
+ "created_by": {
+ "type": "HumanUser",
+ "id": 1,
+ "name": version.created_by or "MEDM User",
+ },
+ "entity": {
+ "type": "Asset",
+ "id": None,
+ "name": asset.name,
+ },
+ "project": {
+ "type": "Project",
+ "id": self._project_id,
+ "name": self._project_name,
+ },
+ "task": None,
+ "published_file_type": {
+ "type": "PublishedFileType",
+ "id": sg_publish_type_id,
+ "name": sg_publish_type_code,
+ },
+ "image": None,
+ # Thumbnail is resolved in a background thread via this revision ID.
+ "_thumbnail_revision_id": asset_revision.id,
+ "sg_flow_revision_id": asset_revision.id,
+ "_medm_asset": asset,
+ "_medm_version": version,
+ }
+
+ return sg_dict
+
+ def _draft_to_sg_dict(
+ self, draft_info: DraftInfo, asset: Optional[Asset]
+ ) -> Dict[str, Any]:
+ """
+ Convert a DraftInfo (CheckoutDraftInfo or NewDraftInfo) to a Shotgun-compatible
+ dictionary for display in the version history list as a local draft entry.
+
+ The key conventions that V1 action hooks (flowam_actions.py) rely on:
+ - ``version_number == DRAFT_VERSION_IDENTIFIER (-1)`` -> row is a local draft
+ - ``sg_flow_revision_id`` -> the draft's unique sandbox ID (draft_info.draft_id),
+ used by asset_management.open_draft() and sandbox.is_local_draft()
+
+ :param draft_info: DraftInfo object returned by asset_management.get_asset_drafts()
+ or get_drafts(). May be CheckoutDraftInfo or NewDraftInfo.
+ :param asset: The parent MEDM Asset (may be None for NewDraftInfo)
+ :returns: sg_data dictionary compatible with action hooks and Shotgun UI
+ """
+ sg_publish_type_id = None
+ sg_publish_type_code = "MEDM Asset"
+
+ # Prefer the published asset's type_ids; fall back to the draft's own
+ # type_ids for NewDraftInfo where no published asset exists yet.
+ type_ids = []
+ if asset is not None:
+ type_ids = getattr(asset, "type_ids", None) or []
+ if not type_ids:
+ type_ids = getattr(draft_info, "type_ids", None) or []
+ if type_ids:
+ sg_publish_type_id, sg_publish_type_code = self._resolve_publish_type(
+ type_ids[0]
+ )
+
+ sg_dict = build_draft_sg_dict(
+ draft_info,
+ asset,
+ self._app.context,
+ self._project_id,
+ self._project_name,
+ sg_publish_type_id,
+ sg_publish_type_code,
+ )
+ sg_dict["_medm_version"] = None
+ return sg_dict
+
+ def _add_draft_as_qt_item(
+ self, draft_info: DraftInfo, asset: Optional[Asset]
+ ) -> None:
+ """
+ Convert a DraftInfo to a QStandardItem and insert it at the top of
+ the history model so drafts always appear above published versions.
+
+ Supports both CheckoutDraftInfo (checkout of a published revision) and
+ NewDraftInfo (new asset not yet published).
+
+ :param draft_info: DraftInfo object (CheckoutDraftInfo or NewDraftInfo)
+ :param asset: The parent MEDM Asset (may be None for NewDraftInfo)
+ """
+ draft_type = getattr(draft_info, "draft_type", "unknown")
+
+ if draft_type == "checkout":
+ source_version = getattr(draft_info, "version", "?")
+ label = f"Draft (v{source_version} checkout)"
+ elif draft_type == "new":
+ label = "Draft (new)"
+ else:
+ label = f"Draft ({draft_type})"
+
+ qt_item = QtGui.QStandardItem(label)
+ qt_item.setEditable(False)
+
+ sg_data = self._draft_to_sg_dict(draft_info, asset)
+ qt_item.setData(sg_data, self.SG_DATA_ROLE)
+ qt_item.setData(asset, self.ASSET_ROLE)
+ qt_item.setData(draft_info, self.DRAFT_ROLE)
+
+ qt_item.setData(self._loading_icon, self.PUBLISH_THUMB_ROLE)
+ thumb = utils.create_overlayed_user_publish_thumbnail(
+ qt_item.data(self.PUBLISH_THUMB_ROLE), None
+ )
+ qt_item.setIcon(QtGui.QIcon(thumb))
+
+ def get_sg_data():
+ return qt_item.data(self.SG_DATA_ROLE)
+
+ qt_item.get_sg_data = get_sg_data
+
+ # Prepend so drafts always appear above all published versions.
+ self.insertRow(0, qt_item)
+ self._app.log_debug(
+ f"MEDM History: Added {draft_type} draft '{draft_info.name}' "
+ f"(draft_id={draft_info.draft_id})"
+ )
+
+ def _resolve_publish_type(self, medm_type_id_str: str) -> tuple:
+ """
+ Resolve a MEDM schema type ID to a ``(sg_publish_type_id, display_name)`` pair.
+
+ Resolution order:
+ 1. In-process cache (avoids redundant SG round-trips within the same load).
+ 2. ShotGrid ``PublishedFileType`` lookup by display name (real ID).
+ 3. Hash-based integer fallback when no SG record exists.
+
+ :param medm_type_id_str: MEDM schema type ID string.
+ :returns: Tuple of (integer publish type id, human-readable display name).
+ """
+ if medm_type_id_str in self._cache.publish_types:
+ return self._cache.publish_types[medm_type_id_str]
+
+ display_name = medm_type_id_str
+ try:
+ schema_name = self._flow_module.schema.get_schema_display_name(
+ medm_type_id_str
+ )
+ if schema_name:
+ display_name = schema_name
+ except Exception as e:
+ self._app.log_debug(
+ f"MEDM History: Could not get schema display name for '{medm_type_id_str}': {e}"
+ )
+
+ sg_publish_type_id = None
+ try:
+ pft = self._app.shotgun.find_one(
+ "PublishedFileType",
+ [["code", "is", display_name]],
+ ["id", "code"],
+ )
+ if pft:
+ sg_publish_type_id = pft["id"]
+ self._app.log_debug(
+ f"MEDM History: Resolved PublishedFileType '{display_name}' "
+ f"-> SG id={sg_publish_type_id}"
+ )
+ else:
+ self._app.log_debug(
+ f"MEDM History: No SG PublishedFileType found for '{display_name}', "
+ f"item will bypass type filter"
+ )
+ except Exception as e:
+ self._app.log_debug(
+ f"MEDM History: Could not look up PublishedFileType for '{display_name}': {e}"
+ )
+
+ result = (sg_publish_type_id, display_name)
+ self._cache.publish_types[medm_type_id_str] = result
+ return result
+
+ def _resolve_and_download_thumbnail(
+ self, qt_item: QtGui.QStandardItem, revision_id: str
+ ) -> None:
+ """
+ Delegate thumbnail resolution and download to the shared
+ :class:`MedmThumbnailService`. The service calls :meth:`_apply_thumbnail`
+ on the main thread once the image bytes are available.
+
+ :param qt_item: The QStandardItem to set the thumbnail on.
+ :param revision_id: MEDM AssetRevision ID whose thumbnail is needed.
+ """
+ self._thumbnail_service.request(qt_item, revision_id, self._apply_thumbnail)
+
+ def _apply_thumbnail(self, qt_item: QtGui.QStandardItem, image_data: bytes) -> None:
+ """
+ Apply downloaded image bytes to *qt_item* as a composite publish thumbnail.
+ Called on the main thread by :class:`MedmThumbnailService`.
+
+ :param qt_item: The QStandardItem to update.
+ :param image_data: Raw image bytes.
+ """
+ pixmap = QtGui.QPixmap()
+ if pixmap.loadFromData(image_data):
+ qt_item.setData(pixmap, self.PUBLISH_THUMB_ROLE)
+ thumb = utils.create_overlayed_user_publish_thumbnail(
+ qt_item.data(self.PUBLISH_THUMB_ROLE),
+ qt_item.data(self.USER_THUMB_ROLE),
+ )
+ qt_item.setIcon(QtGui.QIcon(thumb))
diff --git a/python/tk_multi_loader/medm/shared_cache.py b/python/tk_multi_loader/medm/shared_cache.py
new file mode 100644
index 0000000..ea9831f
--- /dev/null
+++ b/python/tk_multi_loader/medm/shared_cache.py
@@ -0,0 +1,107 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""
+Shared cache container for all MEDM model data.
+
+A single :class:`MedmSharedCache` instance is created by the dialog and
+injected into every MEDM model. Centralising the dictionaries here means:
+
+* No duplicate API calls when the same data is needed by more than one model
+ (e.g. drafts used by both the latest-publish and history panels).
+* A single place to inspect or clear all MEDM state.
+* Refresh semantics are explicit: :meth:`clear_on_refresh` drops only the
+ data that can change between user-triggered refreshes; immutable data
+ (thumbnail images, publish-type mappings) is preserved.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import Any, Dict, List, Optional
+
+
+@dataclasses.dataclass
+class MedmSharedCache:
+ """
+ Central store for all dictionaries shared across MEDM models.
+
+ All values default to empty dicts so the object can be created with
+ ``MedmSharedCache()`` and immediately passed to model constructors.
+
+ Cache semantics
+ ---------------
+ ``children``
+ ``asset.id → list[Asset]``. Populated by both :class:`MedmEntityModel`
+ (when the user expands a tree node) and :class:`MedmLatestPublishModel`
+ (when the user selects a node that hasn't been expanded yet). Cleared
+ on *hard* refresh only, since a new tree load invalidates the hierarchy.
+
+ ``drafts``
+ ``asset.id → list[DraftInfo]``. Volatile — cleared on every refresh
+ so local-draft state is always up-to-date.
+
+ ``versions``
+ ``asset.id → list[AssetVersion]``. Cleared on refresh; a publish
+ action could add a new version.
+
+ ``publish_types``
+ ``medm_type_id_str → (sg_id, display_name)``. Schema-derived and
+ stable for a session; never cleared.
+
+ ``thumbnail_urls``
+ ``revision_id → URL | None``. Immutable (a revision's thumbnail URL
+ never changes); never cleared.
+
+ ``thumbnail_data``
+ ``URL → bytes``. Immutable pixel data; never cleared.
+ """
+
+ # asset.id → list[Asset]
+ children: Dict[str, List[Any]] = dataclasses.field(default_factory=dict)
+ # asset.id → list[DraftInfo]
+ drafts: Dict[str, List[Any]] = dataclasses.field(default_factory=dict)
+ # asset.id → list[AssetVersion]
+ versions: Dict[str, List[Any]] = dataclasses.field(default_factory=dict)
+ # medm_type_id_str → (sg_publish_type_id, display_name)
+ publish_types: Dict[str, tuple] = dataclasses.field(default_factory=dict)
+ # revision_id → thumbnail URL (or None when the API returned nothing)
+ thumbnail_urls: Dict[str, Optional[str]] = dataclasses.field(default_factory=dict)
+ # URL → raw image bytes
+ thumbnail_data: Dict[str, bytes] = dataclasses.field(default_factory=dict)
+
+ # -------------------------------------------------------------------------
+ # Convenience clear helpers
+ # -------------------------------------------------------------------------
+
+ def clear_on_refresh(self) -> None:
+ """
+ Clear caches that may be stale after a user-triggered refresh.
+
+ Deliberately preserves:
+ * ``children`` — tree structure is still valid.
+ * ``publish_types`` — ShotGrid types don't change during a session.
+ * ``thumbnail_urls`` — revision IDs are immutable.
+ * ``thumbnail_data`` — pixel data for a URL never changes.
+ """
+ self.drafts.clear()
+ self.versions.clear()
+
+ def clear_on_hard_refresh(self) -> None:
+ """
+ Full reset used when the tree is reloaded from scratch.
+
+ Still preserves thumbnail caches because those are purely content-
+ addressed (URL → bytes) and revision IDs are immutable.
+ """
+ self.children.clear()
+ self.drafts.clear()
+ self.versions.clear()
+ self.publish_types.clear()
diff --git a/python/tk_multi_loader/medm/thumbnail_service.py b/python/tk_multi_loader/medm/thumbnail_service.py
new file mode 100644
index 0000000..40a43e7
--- /dev/null
+++ b/python/tk_multi_loader/medm/thumbnail_service.py
@@ -0,0 +1,209 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""
+MEDM Thumbnail Service - shared thumbnail URL resolver and image downloader.
+
+A single instance is created per dialog session and injected into every MEDM
+model that needs thumbnails. Because MEDM revision IDs are immutable (a given
+ID always resolves to the same URL and pixel data), both caches are kept for
+the entire session and are never evicted on refresh.
+
+The URL and data caches are owned by the :class:`~medm.shared_cache.MedmSharedCache`
+passed at construction time, so all MEDM state lives in one place.
+"""
+
+from __future__ import annotations
+
+import socket
+import threading
+import urllib.request
+from typing import TYPE_CHECKING, Callable, Dict, Optional
+
+import sgtk
+from sgtk.platform.qt import QtCore, QtGui
+
+if TYPE_CHECKING:
+ from .shared_cache import MedmSharedCache
+
+
+class MedmThumbnailService(QtCore.QObject):
+ """
+ Session-scoped thumbnail URL resolver and image downloader shared across
+ all MEDM model instances (MedmLatestPublishModel, MedmPublishHistoryModel).
+
+ The URL and image-data caches are provided externally via
+ :class:`~medm.shared_cache.MedmSharedCache` so all MEDM caches live in a
+ single, inspectable location. The service itself owns only the operational
+ state: a timer, a pending-work queue, and the background threads.
+
+ Usage::
+
+ cache = MedmSharedCache()
+ service = MedmThumbnailService(cache, parent=dialog)
+ service.request(qt_item, revision_id, my_apply_fn)
+ # my_apply_fn(qt_item, image_data) is called on the main thread.
+ """
+
+ _CONNECTION_TIMEOUT = 5 # seconds - waiting for the server to respond
+ _READ_TIMEOUT = 10 # seconds - total socket timeout
+ _MAX_IMAGE_BYTES = 10 * 1024 * 1024 # 10 MB hard cap per thumbnail
+
+ def __init__(
+ self,
+ cache: "MedmSharedCache",
+ parent: Optional[QtCore.QObject] = None,
+ ) -> None:
+ """
+ :param cache: Shared cache whose ``thumbnail_urls`` and
+ ``thumbnail_data`` dicts are used for storage. The service never
+ clears these dicts; that responsibility belongs to the owner of the
+ cache (they are immutable by design).
+ :param parent: Optional Qt parent object.
+ """
+ super().__init__(parent)
+
+ self._app = sgtk.platform.current_bundle()
+ self._flow_module = sgtk.platform.import_framework(
+ "tk-framework-flowam", "flow"
+ )
+
+ # Both dicts are references into the shared cache - not owned here.
+ self._url_cache: Dict[str, Optional[str]] = cache.thumbnail_urls
+ self._data_cache: Dict[str, bytes] = cache.thumbnail_data
+
+ # item_id(int) → (qt_item, image_data, callback)
+ self._pending: Dict[int, tuple] = {}
+ self._timer: Optional[QtCore.QTimer] = None
+
+ # -------------------------------------------------------------------------
+ # Public API
+ # -------------------------------------------------------------------------
+
+ def request(
+ self,
+ qt_item: QtGui.QStandardItem,
+ revision_id: str,
+ callback: Callable,
+ ) -> None:
+ """
+ Schedule a thumbnail download for *qt_item*.
+
+ If both the URL and the pixel bytes are already cached the item is
+ queued for the next timer tick (no background thread is started).
+ Otherwise a daemon thread resolves the URL and downloads the image.
+
+ :param qt_item: ``QStandardItem`` whose icon should be updated.
+ :param revision_id: MEDM ``AssetRevision`` ID to look up.
+ :param callback: ``callable(qt_item, image_data: bytes)`` that will be
+ invoked on the **main thread** once the image bytes are available.
+ The callback is responsible for converting bytes to a ``QPixmap``
+ and applying it to *qt_item*.
+ """
+ self._ensure_timer()
+
+ # Fast path: both URL and image bytes already cached.
+ cached_url = self._url_cache.get(revision_id)
+ if cached_url and cached_url in self._data_cache:
+ self._pending[id(qt_item)] = (
+ qt_item,
+ self._data_cache[cached_url],
+ callback,
+ )
+ return
+
+ threading.Thread(
+ target=self._resolve_and_fetch,
+ args=(qt_item, revision_id, callback),
+ daemon=True,
+ ).start()
+
+ def destroy(self) -> None:
+ """Stop the processing timer and discard all pending work."""
+ if self._timer is not None:
+ self._timer.stop()
+ self._timer = None
+ self._pending.clear()
+
+ # -------------------------------------------------------------------------
+ # Private helpers
+ # -------------------------------------------------------------------------
+
+ def _ensure_timer(self) -> None:
+ """Start the main-thread processing timer on first use."""
+ if self._timer is None:
+ self._timer = QtCore.QTimer(self)
+ self._timer.timeout.connect(self._process_pending)
+ self._timer.start(100)
+
+ def _resolve_and_fetch(
+ self, qt_item: QtGui.QStandardItem, revision_id: str, callback: Callable
+ ) -> None:
+ """Background-thread worker: resolve URL then download bytes."""
+ # 1. Resolve the thumbnail URL (hits the API at most once per revision).
+ url = self._url_cache.get(revision_id)
+ if url is None and revision_id not in self._url_cache:
+ try:
+ url = self._flow_module.asset_management.get_thumbnail_url(revision_id)
+ except Exception as exc:
+ self._app.log_debug(
+ f"MEDM ThumbnailService: URL resolve failed for {revision_id}: {exc}"
+ )
+ self._url_cache[revision_id] = url
+
+ if not url:
+ return
+
+ # 2. Download image bytes (cached after the first download).
+ if url in self._data_cache:
+ self._pending[id(qt_item)] = (qt_item, self._data_cache[url], callback)
+ return
+
+ try:
+ req = urllib.request.Request(url)
+ req.add_header("User-Agent", "ShotgunToolkit/1.0")
+
+ old_timeout = socket.getdefaulttimeout()
+ try:
+ socket.setdefaulttimeout(self._CONNECTION_TIMEOUT + self._READ_TIMEOUT)
+ with urllib.request.urlopen(
+ req, timeout=self._CONNECTION_TIMEOUT
+ ) as response:
+ if response.status != 200:
+ self._app.log_debug(
+ f"MEDM ThumbnailService: HTTP {response.status} for {revision_id}"
+ )
+ return
+ image_data = response.read(self._MAX_IMAGE_BYTES)
+ if len(image_data) >= self._MAX_IMAGE_BYTES:
+ # Suspiciously large - skip rather than store garbage.
+ return
+ self._data_cache[url] = image_data
+ self._pending[id(qt_item)] = (qt_item, image_data, callback)
+ finally:
+ socket.setdefaulttimeout(old_timeout)
+
+ except Exception as exc:
+ self._app.log_debug(
+ f"MEDM ThumbnailService: Download failed: {type(exc).__name__}: {exc}"
+ )
+
+ def _process_pending(self) -> None:
+ """Main-thread timer callback: fire stored callbacks for ready items."""
+ pending = list(self._pending.items())
+ self._pending.clear()
+
+ for _item_id, (qt_item, image_data, callback) in pending:
+ try:
+ callback(qt_item, image_data)
+ except Exception as exc:
+ self._app.log_debug(
+ f"MEDM ThumbnailService: Callback error: {type(exc).__name__}: {exc}"
+ )
diff --git a/python/tk_multi_loader/medm/utils.py b/python/tk_multi_loader/medm/utils.py
new file mode 100644
index 0000000..ba4212a
--- /dev/null
+++ b/python/tk_multi_loader/medm/utils.py
@@ -0,0 +1,201 @@
+# Copyright (c) 2026 Shotgun Software Inc.
+#
+# CONFIDENTIAL AND PROPRIETARY
+#
+# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
+# Source Code License included in this distribution package. See LICENSE.
+# By accessing, using, copying or modifying this work you indicate your
+# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
+# not expressly granted therein are reserved by Shotgun Software Inc.
+
+"""Shared utility helpers for MEDM models.
+
+Functions here are intentionally free of Qt and SGTK framework imports so
+they stay lightweight and easy to unit-test.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Dict, Optional, Tuple
+
+
+def is_structural_asset(asset: Any, flow_module: Any) -> bool:
+ """Return ``True`` when *asset* is a structural container in the MEDM hierarchy.
+
+ An asset is considered structural — and therefore belongs only in the
+ left-hand tree, never in the centre-panel publish list — when **any** of
+ the following hold:
+
+ * its type IDs include the built-in ``FOLDER_TYPE_ID``; **or**
+ * it carries a ``CONTAINER_TYPE`` or ``PIPELINE_STEP_TYPE`` component
+ (schema-registered organisational node types).
+
+ All framework calls are wrapped in a broad ``except`` so that a transient
+ schema error never surfaces as a visible crash; the safe default is
+ ``False`` (treat the asset as publishable).
+
+ :param asset: MEDM ``Asset`` object to test.
+ :param flow_module: The ``flow`` framework module imported via
+ ``sgtk.platform.import_framework("tk-framework-flowam", "flow")``.
+ :returns: ``True`` if the asset is a structural container.
+ """
+ try:
+ type_ids = set(getattr(asset, "type_ids", None) or [])
+ if flow_module.data.FOLDER_TYPE_ID in type_ids:
+ return True
+
+ _am = flow_module.asset_management
+ structural_types = (_am.CONTAINER_TYPE, _am.PIPELINE_STEP_TYPE)
+ return any(
+ asset.find_component(type_id=flow_module.schema.get_schema_id(ct))
+ for ct in structural_types
+ )
+ except Exception:
+ return False
+
+
+# Sentinel version number that action hooks use to detect a local draft row
+# and route it to asset_management.open_draft() instead of checkout_revision().
+# Both MEDM model classes expose this as a class constant with the same value.
+DRAFT_VERSION_IDENTIFIER: int = -1
+
+
+def get_draft_created_at(draft_info: Any) -> Optional[float]:
+ """Return the creation timestamp for a local draft as a Unix float.
+
+ Uses the mtime of the draft's ``source_path`` file on disk. Falls back
+ to ``None`` when the path is absent or the filesystem query fails.
+
+ :param draft_info: A ``DraftInfo`` instance (``CheckoutDraftInfo`` or
+ ``NewDraftInfo``).
+ :returns: Unix timestamp float, or ``None`` if unavailable.
+ """
+ source_path = getattr(draft_info, "source_path", None)
+ if source_path:
+ try:
+ return os.path.getmtime(source_path)
+ except OSError:
+ pass
+ return None
+
+
+def get_draft_created_by(context: Any) -> Dict[str, Any]:
+ """Return a Shotgun-compatible ``created_by`` dict for the current user.
+
+ Local drafts are always owned by the currently logged-in user, so the
+ SGTK context is the authoritative source.
+
+ :param context: A ``sgtk.Context`` object (``self._app.context``).
+ :returns: Dict with ``type``, ``id``, and ``name`` keys.
+ """
+ ctx_user = getattr(context, "user", None) or {}
+ return {
+ "type": "HumanUser",
+ "id": ctx_user.get("id", 0),
+ "name": ctx_user.get("name", ""),
+ }
+
+
+def get_draft_metadata(
+ draft_info: Any, context: Any
+) -> Tuple[Optional[float], Dict[str, Any]]:
+ """Convenience wrapper returning ``(created_at, created_by)`` for a draft.
+
+ Combines :func:`get_draft_created_at` and :func:`get_draft_created_by`
+ into a single call for the common case where both values are needed
+ together.
+
+ :param draft_info: A ``DraftInfo`` instance.
+ :param context: A ``sgtk.Context`` object.
+ :returns: Tuple of ``(created_at_float_or_none, created_by_dict)``.
+ """
+ return get_draft_created_at(draft_info), get_draft_created_by(context)
+
+
+def get_draft_description(draft_info: Any) -> str:
+ """Build a human-readable description string for a local draft.
+
+ :param draft_info: A ``DraftInfo`` instance.
+ :returns: Non-empty description string.
+ """
+ draft_type = getattr(draft_info, "draft_type", "unknown")
+ if draft_type == "checkout":
+ source_version = getattr(draft_info, "version", "?")
+ return f"Local checkout of v{source_version}"
+ if draft_type == "new":
+ return getattr(draft_info, "description", "") or "New unpublished asset"
+ return f"Local draft ({draft_type})"
+
+
+def build_draft_sg_dict(
+ draft_info: Any,
+ asset: Any,
+ context: Any,
+ project_id: int,
+ project_name: str,
+ sg_publish_type_id: Optional[int],
+ sg_publish_type_code: str,
+) -> Dict[str, Any]:
+ """Build the common Shotgun-compatible dictionary for a local draft.
+
+ Returns the intersection of fields used by both the latest-publish model
+ and the publish-history model. Callers should merge in any additional
+ model-specific keys after calling this function::
+
+ sg_dict = build_draft_sg_dict(...)
+ sg_dict.update({"_medm_version": None}) # history model
+ sg_dict.update({ # latest-publish model
+ "task_uniqueness": True,
+ "tank_type": {...},
+ "_thumbnail_revision_id": ...,
+ })
+
+ :param draft_info: ``DraftInfo`` instance (``CheckoutDraftInfo`` or
+ ``NewDraftInfo``).
+ :param asset: MEDM ``Asset`` the draft belongs to. May be ``None`` for
+ ``NewDraftInfo`` entries surfaced by the history model.
+ :param context: ``sgtk.Context`` used to resolve the current user.
+ :param project_id: ShotGrid project ID.
+ :param project_name: ShotGrid project name.
+ :param sg_publish_type_id: Resolved SG ``PublishedFileType`` ID (may be
+ ``None`` when no matching SG record exists).
+ :param sg_publish_type_code: Human-readable type display name.
+ :returns: sg_data dict ready for Qt model consumption.
+ """
+ draft_name = getattr(draft_info, "name", getattr(asset, "name", ""))
+ draft_type = getattr(draft_info, "draft_type", "unknown")
+ created_at, created_by = get_draft_metadata(draft_info, context)
+
+ return {
+ "id": None,
+ "type": "PublishedFile",
+ "code": f"{draft_name}.DRAFT",
+ "name": draft_name,
+ # DRAFT_VERSION_IDENTIFIER (-1) signals action hooks to call
+ # open_draft() instead of checkout_revision().
+ "version_number": DRAFT_VERSION_IDENTIFIER,
+ "description": get_draft_description(draft_info),
+ "created_at": created_at,
+ "created_by": created_by,
+ "entity": (
+ {"type": "Asset", "id": None, "name": asset.name}
+ if asset is not None
+ else None
+ ),
+ "project": {"type": "Project", "id": project_id, "name": project_name},
+ "task": None,
+ "published_file_type": {
+ "type": "PublishedFileType",
+ "id": sg_publish_type_id,
+ "name": sg_publish_type_code,
+ },
+ "image": None,
+ # draft_id is the sandbox ID consumed by asset_management.open_draft()
+ # and sandbox.is_local_draft().
+ "sg_flow_revision_id": getattr(draft_info, "draft_id", None),
+ "path": {"local_path": getattr(draft_info, "source_path", None)},
+ "_medm_asset": asset,
+ "_medm_draft": draft_info,
+ "_medm_draft_type": draft_type,
+ }