Use model classes from galaxy.model instead of app.model object

for proper type annotation of query results.

Also:
- Drop unused code from `lib/tool_shed/webapp/controllers/user.py`
- Drop Galaxy model code from
  `lib/tool_shed/util/admin_util.py:Admin.purge_user()`

lib/galaxy/model/__init__.py

@@ -4239,7 +4239,7 @@ class conversion_messages(str, Enum):
         OK = "ok"
 
     permitted_actions = get_permitted_actions(filter="DATASET")
-    file_path = "/tmp/"
+    file_path: ClassVar[str] = "/tmp/"
     object_store: ClassVar[Optional["BaseObjectStore"]] = (
         None  # This get initialized in mapping.py (method init) by app.py
     )

lib/galaxy/webapps/galaxy/services/library_contents.py

@@ -22,7 +22,9 @@
 )
 from galaxy.managers.hdas import HDAManager
 from galaxy.model import (
+    ImplicitlyConvertedDatasetAssociation,
     Library,
+    LibraryDatasetDatasetAssociation,
     tags,
 )
 from galaxy.schema.fields import DecodedDatabaseIdField
@@ -189,7 +191,7 @@ def update(
             content_conv = self.get_library_dataset(
                 trans, payload.converted_dataset_id, check_ownership=False, check_accessible=False
             )
-            assoc = trans.app.model.ImplicitlyConvertedDatasetAssociation(
+            assoc = ImplicitlyConvertedDatasetAssociation(
                 parent=content.library_dataset_dataset_association,
                 dataset=content_conv.library_dataset_dataset_association,
                 file_type=content_conv.library_dataset_dataset_association.extension,
@@ -285,13 +287,13 @@ def _traverse(self, trans: ProvidesUserContext, folder, current_user_roles):
                 rval.append(ld)
         return rval
 
-    def _create_response(self, trans, payload, output, library_id):
+    def _create_response(self, trans: ProvidesHistoryContext, payload, output, library_id: DecodedDatabaseIdField):
         rval = []
         for v in output.values():
             if payload.extended_metadata is not None:
                 # If there is extended metadata, store it, attach it to the dataset, and index it
                 self.create_extended_metadata(trans, payload.extended_metadata)
-            if isinstance(v, trans.app.model.LibraryDatasetDatasetAssociation):
+            if isinstance(v, LibraryDatasetDatasetAssociation):
                 v = v.library_dataset
             url = self._url_for(trans, library_id, v.id, payload.create_type)
             rval.append(dict(id=v.id, name=v.name, url=url))

lib/tool_shed/managers/categories.py

@@ -36,7 +36,7 @@ def create(self, trans: ProvidesUserContext, category_request: CreateCategoryReq
                 raise exceptions.Conflict("A category with that name already exists.")
             else:
                 # Create the category
-                category = self.app.model.Category(name=name, description=description)
+                category = Category(name=name, description=description)
                 trans.sa_session.add(category)
                 trans.sa_session.commit()
             return category

lib/tool_shed/managers/groups.py

@@ -22,11 +22,11 @@
     ObjectNotFound,
     RequestParameterInvalidException,
 )
+from tool_shed.webapp.model import Group
 
 log = logging.getLogger(__name__)
 
 
-# =============================================================================
 class GroupManager:
     """
     Interface/service object for interacting with TS groups.
@@ -50,9 +50,9 @@ def get(self, trans, decoded_group_id=None, name=None):
 
         try:
             if decoded_group_id:
-                group = trans.sa_session.get(trans.app.model.Group, decoded_group_id)
+                group = trans.sa_session.get(Group, decoded_group_id)
             else:
-                group = get_group_by_name(trans.sa_session, name, trans.app.model.Group)
+                group = get_group_by_name(trans.sa_session, name, Group)
         except MultipleResultsFound:
             raise InconsistentDatabase("Multiple groups found with the same identifier.")
         except NoResultFound:
@@ -71,7 +71,7 @@ def create(self, trans, name, description=""):
             if self.get(trans, name=name):
                 raise Conflict(f"Group with the given name already exists. Name: {str(name)}")
             # TODO add description field to the model
-            group = trans.app.model.Group(name=name)
+            group = Group(name=name)
             trans.sa_session.add(group)
             trans.sa_session.commit()
             return group
@@ -117,7 +117,6 @@ def list(self, trans, deleted=False):
         :returns: query that will emit all groups
         :rtype:   sqlalchemy query
         """
-        Group = trans.app.model.Group
         stmt = select(Group)
         if trans.user_is_admin:
             if deleted is None:

lib/tool_shed/managers/repositories.py

@@ -67,6 +67,7 @@
 from tool_shed.webapp.model import (
     Repository,
     RepositoryMetadata,
+    User,
 )
 from tool_shed.webapp.model.db import get_repository_by_name_and_owner
 from tool_shed.webapp.search.repo_search import RepoSearch
@@ -197,10 +198,10 @@ def check_updates(app: ToolShedApp, request: UpdatesRequest) -> Union[str, Dict[
     return tool_shed_encode({}) if hexlify_this else json.dumps({})
 
 
-def guid_to_repository(app: ToolShedApp, tool_id: str) -> "Repository":
+def guid_to_repository(app: ToolShedApp, tool_id: str) -> Repository:
     # tool_id = remove_protocol_and_user_from_clone_url(tool_id)
     shed, _, owner, name, rest = tool_id.split("/", 5)
-    return _get_repository_by_name_and_owner(app.model.context, name, owner, app.model.User)
+    return _get_repository_by_name_and_owner(app.model.context, name, owner)
 
 
 def index_tool_ids(app: ToolShedApp, tool_ids: List[str]) -> Dict[str, Any]:
@@ -211,7 +212,7 @@ def index_tool_ids(app: ToolShedApp, tool_ids: List[str]) -> Dict[str, Any]:
         owner = repository.user.username
         name = repository.name
         assert name
-        repository = _get_repository_by_name_and_owner(app.model.session, name, owner, app.model.User)
+        repository = _get_repository_by_name_and_owner(app.model.session, name, owner)
         if not repository:
             log.warning(f"Repository {owner}/{name} does not exist, skipping")
             continue
@@ -253,9 +254,7 @@ def index_tool_ids(app: ToolShedApp, tool_ids: List[str]) -> Dict[str, Any]:
 
 
 def index_repositories(app: ToolShedApp, name: Optional[str], owner: Optional[str], deleted: bool):
-    return list(
-        _get_repositories_by_name_and_owner_and_deleted(app.model.context, name, owner, deleted, app.model.User)
-    )
+    return list(_get_repositories_by_name_and_owner_and_deleted(app.model.context, name, owner, deleted))
 
 
 def can_manage_repo(trans: ProvidesUserContext, repository: Repository) -> bool:
@@ -589,26 +588,26 @@ def ensure_can_manage(trans: ProvidesUserContext, repository: Repository, error_
         raise InsufficientPermissionsException(error_message)
 
 
-def _get_repository_by_name_and_owner(session: scoped_session, name: str, owner: str, user_model):
+def _get_repository_by_name_and_owner(session: scoped_session, name: str, owner: str):
     stmt = (
         select(Repository)
         .where(Repository.deprecated == false())
         .where(Repository.deleted == false())
         .where(Repository.name == name)
-        .where(user_model.username == owner)
-        .where(Repository.user_id == user_model.id)
+        .where(User.username == owner)
+        .where(Repository.user_id == User.id)
         .limit(1)
     )
     return session.scalars(stmt).first()
 
 
 def _get_repositories_by_name_and_owner_and_deleted(
-    session: scoped_session, name: Optional[str], owner: Optional[str], deleted: bool, user_model
+    session: scoped_session, name: Optional[str], owner: Optional[str], deleted: bool
 ):
     stmt = select(Repository).where(Repository.deprecated == false()).where(Repository.deleted == deleted)
     if owner is not None:
-        stmt = stmt.where(user_model.username == owner)
-        stmt = stmt.where(Repository.user_id == user_model.id)
+        stmt = stmt.where(User.username == owner)
+        stmt = stmt.where(Repository.user_id == User.id)
     if name is not None:
         stmt = stmt.where(Repository.name == name)
     stmt = stmt.order_by(Repository.name)

lib/tool_shed/managers/users.py

@@ -19,7 +19,7 @@
 
 def index(app: ToolShedApp, deleted: bool) -> List[ApiUser]:
     users: List[ApiUser] = []
-    for user in get_users_by_deleted(app.model.context, app.model.User, deleted):
+    for user in get_users_by_deleted(app.model.context, User, deleted):
         users.append(get_api_user(app, user))
     return users
 

lib/tool_shed/metadata/repository_metadata_manager.py

@@ -504,7 +504,7 @@ def create_or_update_repository_metadata(self, changeset_revision, metadata_dict
             repository_metadata.includes_tool_dependencies = includes_tool_dependencies
             repository_metadata.includes_workflows = False
         else:
-            repository_metadata = self.app.model.RepositoryMetadata(
+            repository_metadata = RepositoryMetadata(
                 repository_id=self.repository.id,
                 changeset_revision=changeset_revision,
                 metadata=metadata_dict,