Merge branch 'release_24.1' into release_24.2

Author: nsoranzo

lib/galaxy/model/__init__.py

@@ -230,9 +230,12 @@
 from galaxy.util.sanitize_html import sanitize_html
 
 if TYPE_CHECKING:
+    from sqlalchemy.sql.expression import BindParameter
+
     from galaxy.objectstore import (
         BaseObjectStore,
         ObjectStorePopulator,
+        QuotaSourceMap,
     )
     from galaxy.schema.invocation import InvocationMessageUnion
 
@@ -674,9 +677,9 @@ def stderr(self, stderr):
 """
 
 
-def calculate_user_disk_usage_statements(user_id, quota_source_map, for_sqlite=False):
+def calculate_user_disk_usage_statements(user_id: int, quota_source_map: "QuotaSourceMap", for_sqlite: bool = False):
     """Standalone function so can be reused for postgres directly in pgcleanup.py."""
-    statements = []
+    statements: List[Tuple[str, Dict[str, Any]]] = []
     default_quota_enabled = quota_source_map.default_quota_enabled
     default_exclude_ids = quota_source_map.default_usage_excluded_ids()
     default_cond = "dataset.object_store_id IS NULL" if default_quota_enabled and default_exclude_ids else ""
@@ -692,7 +695,7 @@ def calculate_user_disk_usage_statements(user_id, quota_source_map, for_sqlite=F
 UPDATE galaxy_user SET disk_usage = ({default_usage})
 WHERE id = :id
 """
-    params = {"id": user_id}
+    params: Dict[str, Any] = {"id": user_id}
     if default_exclude_ids:
         params["exclude_object_store_ids"] = default_exclude_ids
     statements.append((default_usage, params))
@@ -1161,25 +1164,27 @@ def calculate_disk_usage_default_source(self, object_store):
         usage = sa_session.scalar(sql_calc, params)
         return usage
 
-    def calculate_and_set_disk_usage(self, object_store):
+    def calculate_and_set_disk_usage(self, object_store: "BaseObjectStore"):
         """
         Calculates and sets user disk usage.
         """
         self._calculate_or_set_disk_usage(object_store=object_store)
 
-    def _calculate_or_set_disk_usage(self, object_store):
+    def _calculate_or_set_disk_usage(self, object_store: "BaseObjectStore"):
         """
         Utility to calculate and return the disk usage.  If dryrun is False,
         the new value is set immediately.
         """
         assert object_store is not None
         quota_source_map = object_store.get_quota_source_map()
         sa_session = object_session(self)
+        assert sa_session
+        assert sa_session.bind
         for_sqlite = "sqlite" in sa_session.bind.dialect.name
         statements = calculate_user_disk_usage_statements(self.id, quota_source_map, for_sqlite)
         for sql, args in statements:
             statement = text(sql)
-            binds = []
+            binds: List[BindParameter] = []
             for key, _ in args.items():
                 expand_binding = key.endswith("s")
                 binds.append(bindparam(key, expanding=expand_binding))

lib/galaxy/objectstore/__init__.py

@@ -403,7 +403,7 @@ def to_dict(self) -> Dict[str, Any]:
         raise NotImplementedError()
 
     @abc.abstractmethod
-    def get_quota_source_map(self):
+    def get_quota_source_map(self) -> "QuotaSourceMap":
         """Return QuotaSourceMap describing mapping of object store IDs to quota sources."""
 
     @abc.abstractmethod
@@ -715,7 +715,7 @@ def parse_badges_from_config_xml(clazz, badges_xml):
             badges.append({"type": type, "message": message})
         return badges
 
-    def get_quota_source_map(self):
+    def get_quota_source_map(self) -> "QuotaSourceMap":
         # I'd rather keep this abstract... but register_singleton wants it to be instantiable...
         raise NotImplementedError()
 

scripts/cleanup_datasets/pgcleanup.py

@@ -401,7 +401,8 @@ def recalculate_disk_usage(self):
             statements = calculate_user_disk_usage_statements(user_id, quota_source_map)
 
             for sql, args in statements:
-                sql, _ = re.subn(r"\:([\w]+)", r"%(\1)s", sql)
+                sql = sql.replace("%", "%%")
+                sql = re.sub(r"\:([\w]+)", r"%(\1)s", sql)
                 new_args = {}
                 for key, val in args.items():
                     if isinstance(val, list):