Add quota_repo; add get_deleted method; use in services

lib/galaxy/model/repositories/quota.py

@@ -0,0 +1,36 @@
+from typing import (
+    cast,
+    List,
+    Optional,
+)
+
+from sqlalchemy import (
+    false,
+    select,
+    true,
+)
+
+from galaxy.model import Quota
+from galaxy.model.repositories import (
+    BaseRepository,
+    MappedType,
+    SessionType,
+)
+
+
+class QuotaRepository(BaseRepository):
+    def __init__(self, session: SessionType, model_class: Optional[MappedType] = None):
+        model_class = model_class or Quota
+        super().__init__(session, model_class)
+
+    def get(self, primary_key: int) -> Quota:
+        return cast(Quota, super().get(primary_key))
+
+    def get_deleted(self, deleted: bool = True) -> List:
+        # type-ignore/SessionlessContext
+        if deleted:
+            is_deleted = true()
+        else:
+            is_deleted = false()
+        stmt = select(self.model_class).filter(self.model_class.deleted == is_deleted)  # type:ignore[attr-defined]
+        return self.session.scalars(stmt).all()  # type:ignore[union-attr]

lib/galaxy/webapps/galaxy/services/quotas.py

@@ -1,18 +1,15 @@
 import logging
 from typing import Optional
 
-from sqlalchemy import (
-    false,
-    select,
-    true,
-)
+from sqlalchemy import select
 
 from galaxy import (
     model,
     util,
 )
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.quotas import QuotaManager
+from galaxy.model.repositories.quota import QuotaRepository
 from galaxy.quota._schema import (
     CreateQuotaParams,
     CreateQuotaResult,
@@ -40,14 +37,14 @@ def __init__(self, security: IdEncodingHelper, quota_manager: QuotaManager):
     def index(self, trans: ProvidesUserContext, deleted: bool = False) -> QuotaSummaryList:
         """Displays a list of quotas."""
         rval = []
-        stmt = select(model.Quota)
+        quota_repo = QuotaRepository(trans.sa_session)
         if deleted:
             route = "deleted_quota"
-            stmt = stmt.filter(model.Quota.deleted == true())
+            quotas = quota_repo.get_deleted()
         else:
             route = "quota"
-            stmt = stmt.filter(model.Quota.deleted == false())
-        for quota in trans.sa_session.scalars(stmt):
+            quotas = quota_repo.get_deleted(False)
+        for quota in quotas:
             item = quota.to_dict(value_mapper={"id": DecodedDatabaseIdField.encode})
             encoded_id = DecodedDatabaseIdField.encode(quota.id)
             item["url"] = url_for(route, id=encoded_id)