class ModelAccessManager(ABC):
@abstractmethod
def on_model_access_forbidden(self, model_id: str, api_key: Optional[str]) -> None:
pass
@abstractmethod
def on_model_package_access_granted(
self, access_identifiers: AccessIdentifiers
) -> None:
pass
@abstractmethod
def on_file_created(
self, file_path: str, access_identifiers: AccessIdentifiers
) -> None:
pass
@abstractmethod
def on_file_renamed(
self, old_path: str, new_path: str, access_identifiers: AccessIdentifiers
) -> None:
pass
@abstractmethod
def on_symlink_created(
self, target_path: str, link_name: str, access_identifiers: AccessIdentifiers
) -> None:
pass
@abstractmethod
def on_symlink_deleted(self, link_name: str) -> None:
pass
@abstractmethod
def on_file_deleted(self, file_path: str) -> None:
pass
@abstractmethod
def on_directory_deleted(self, dir_path: str) -> None:
pass
@abstractmethod
def is_model_access_forbidden(self, model_id: str, api_key: Optional[str]) -> bool:
pass
@abstractmethod
def is_model_package_access_granted(
self, model_id: str, package_id: str, api_key: Optional[str]
) -> bool:
pass
@abstractmethod
def retrieve_model_instance(
self,
model_id: str,
package_id: Optional[str],
api_key: Optional[str],
loading_parameter_digest: Optional[str],
) -> Optional[AnyModel]:
pass
def retrieve_model_storage_path(
self,
model: AnyModel,
model_id: str,
package_id: Optional[str],
api_key: Optional[str],
loading_parameter_digest: Optional[str],
) -> Optional[str]:
"""Return the verified package path associated with a cached instance.
Managers that keep their own instance registry may override this.
Auto-loaded models also carry this private attribution directly.
"""
model_storage_path = getattr(
model,
"_inference_models_package_path",
None,
)
return model_storage_path if isinstance(model_storage_path, str) else None
@abstractmethod
def on_model_loaded(
self,
model: AnyModel,
access_identifiers: AccessIdentifiers,
model_storage_path: str,
) -> None:
pass
@abstractmethod
def on_model_alias_discovered(self, alias: str, model_id: str) -> None:
pass
@abstractmethod
def on_model_dependency_discovered(
self,
base_model_id: str,
base_model_package_id: Optional[str],
dependent_model_id: str,
) -> None:
pass