Typing for various tool shed utilities.

lib/tool_shed/structured_app.py

@@ -3,6 +3,7 @@ from typing import TYPE_CHECKING
 from galaxy.structured_app import BasicSharedApp
 
 if TYPE_CHECKING:
+    from tool_shed.repository_registry import Registry as RepositoryRegistry
     from tool_shed.repository_types.registry import Registry as RepositoryTypesRegistry
     from tool_shed.webapp.model import mapping
 
@@ -10,3 +11,4 @@ if TYPE_CHECKING:
 class ToolShedApp(BasicSharedApp):
     repository_types_registry: "RepositoryTypesRegistry"
     model: "mapping.ToolShedModelMapping"
+    repository_registry: "RepositoryRegistry"

lib/tool_shed/util/commit_util.py

@@ -6,6 +6,13 @@ import os
 import shutil
 import tempfile
 from collections import namedtuple
+from typing import (
+    List,
+    Optional,
+    Tuple,
+    TYPE_CHECKING,
+    Union,
+)
 
 from sqlalchemy.sql.expression import null
 
@@ -19,13 +26,17 @@ from tool_shed.util import (
     shed_util_common as suc,
 )
 
+if TYPE_CHECKING:
+    from tool_shed.structured_app import ToolShedApp
+    from tool_shed.webapp.model import Repository
+
 log = logging.getLogger(__name__)
 
 UNDESIRABLE_DIRS = [".hg", ".svn", ".git", ".cvs", ".idea"]
 UNDESIRABLE_FILES = [".hg_archival.txt", "hgrc", ".DS_Store", "tool_test_output.html", "tool_test_output.json"]
 
 
-def check_archive(repository, archive):
+def check_archive(repository: "Repository", archive):
     valid = []
     invalid = []
     errors = []
@@ -81,7 +92,7 @@ def check_archive(repository, archive):
     return ArchiveCheckResults(valid, invalid, undesirable_files, undesirable_dirs, errors)
 
 
-def check_file_contents_for_email_alerts(app):
+def check_file_contents_for_email_alerts(app: "ToolShedApp"):
     """
     See if any admin users have chosen to receive email alerts when a repository is updated.
     If so, the file contents of the update must be checked for inappropriate content.
@@ -126,7 +137,7 @@ def get_change_lines_in_file_for_tag(tag, change_dict):
     return cleaned_lines
 
 
-def get_upload_point(repository, **kwd):
+def get_upload_point(repository: "Repository", **kwd) -> Optional[str]:
     upload_point = kwd.get("upload_point", None)
     if upload_point is not None:
         # The value of upload_point will be something like: database/community_files/000/repo_12/1.bed
@@ -148,7 +159,7 @@ def get_upload_point(repository, **kwd):
     return upload_point
 
 
-def handle_bz2(repository, uploaded_file_name):
+def handle_bz2(repository: "Repository", uploaded_file_name):
     with tempfile.NamedTemporaryFile(
         mode="wb",
         prefix=f"repo_{repository.id}_upload_bunzip2_",
@@ -168,19 +179,22 @@ def handle_bz2(repository, uploaded_file_name):
     shutil.move(uncompressed.name, uploaded_file_name)
 
 
+ChangeResponseT = Tuple[Union[bool, str], str, List[str], str, int, int]
+
+
 def handle_directory_changes(
     app,
-    host,
-    username,
-    repository,
-    full_path,
+    host: str,
+    username: str,
+    repository: "Repository",
+    full_path: str,
     filenames_in_archive,
     remove_repo_files_not_in_tar,
     new_repo_alert,
-    commit_message,
-    undesirable_dirs_removed,
-    undesirable_files_removed,
-):
+    commit_message: str,
+    undesirable_dirs_removed: int,
+    undesirable_files_removed: int,
+) -> ChangeResponseT:
     repo_path = repository.repo_path(app)
     content_alert_str = ""
     files_to_remove = []

lib/tool_shed/util/hg_util.py

@@ -51,7 +51,7 @@ def archive_repository_revision(app, repository, archive_dir, changeset_revision
         raise Exception(error_message)
 
 
-def commit_changeset(repo_path, full_path_to_changeset, username, message):
+def commit_changeset(repo_path: str, full_path_to_changeset: str, username: str, message: str) -> None:
     try:
         subprocess.check_output(
             ["hg", "commit", "-u", username, "-m", message, full_path_to_changeset],

lib/tool_shed/util/repository_util.py

@@ -2,6 +2,10 @@ import configparser
 import logging
 import os
 import re
+from typing import (
+    Tuple,
+    TYPE_CHECKING,
+)
 
 from markupsafe import escape
 from sqlalchemy import false
@@ -54,13 +58,18 @@ from tool_shed.util.metadata_util import (
     get_repository_metadata_by_changeset_revision,
 )
 
+if TYPE_CHECKING:
+    from tool_shed.structured_app import ToolShedApp
+    from tool_shed.webapp.model import Repository
+
+
 log = logging.getLogger(__name__)
 
 VALID_REPOSITORYNAME_RE = re.compile(r"^[a-z0-9\_]+$")
 
 
 def create_repo_info_dict(
-    app,
+    app: "ToolShedApp",
     repository_clone_url,
     changeset_revision,
     ctx_rev,
@@ -139,7 +148,7 @@ def create_repo_info_dict(
     return repo_info_dict
 
 
-def create_repository_admin_role(app, repository):
+def create_repository_admin_role(app: "ToolShedApp", repository: "Repository"):
     """
     Create a new role with name-spaced name based on the repository name and its owner's public user
     name.  This will ensure that the tole name is unique.
@@ -160,16 +169,16 @@ def create_repository_admin_role(app, repository):
 
 
 def create_repository(
-    app,
-    name,
-    type,
+    app: "ToolShedApp",
+    name: str,
+    type: str,
     description,
     long_description,
     user_id,
     category_ids=None,
     remote_repository_url=None,
     homepage_url=None,
-):
+) -> Tuple["Repository", str]:
     """Create a new ToolShed repository"""
     category_ids = category_ids or []
     sa_session = app.model.session