Use SQLAlchemy 2.0 queries in shop, shop catalog, storefront services

byceps/services/shop/catalog/service.py

@@ -9,7 +9,7 @@
 from __future__ import annotations
 from typing import Optional
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
 
 from ....database import db
 
@@ -97,20 +97,20 @@ def create_collection(catalog_id: CatalogID, title: str) -> Collection:
 
 def delete_collection(collection_id: CollectionID) -> None:
     """Delete the collection."""
-    db.session.query(DbCollection) \
-        .filter_by(id=collection_id) \
-        .delete()
-
+    db.session.execute(
+        delete(DbCollection)
+        .where(DbCollection.id == collection_id)
+    )
     db.session.commit()
 
 
 def get_collections_for_catalog(catalog_id: CatalogID) -> list[Collection]:
     """Return the catalog's collections."""
-    db_collections = db.session \
-        .query(DbCollection) \
-        .filter_by(catalog_id=catalog_id) \
-        .order_by(DbCollection.position) \
-        .all()
+    db_collections = db.session.execute(
+        select(DbCollection)
+        .filter_by(catalog_id=catalog_id)
+        .order_by(DbCollection.position)
+    ).scalars().all()
 
     return [
         _db_entity_to_collection(db_collection)
@@ -151,8 +151,8 @@ def remove_article_from_collection(
     catalog_article_id: CatalogArticleID,
 ) -> None:
     """Remove article from collection."""
-    db.session.query(DbCatalogArticle) \
-        .filter_by(id=catalog_article_id) \
-        .delete()
-
+    db.session.execute(
+        delete(DbCatalogArticle)
+        .where(DbCatalogArticle.id == catalog_article_id)
+    )
     db.session.commit()

byceps/services/shop/shop/service.py

@@ -9,6 +9,8 @@
 from __future__ import annotations
 from typing import Optional
 
+from sqlalchemy import delete, select
+
 from ....database import db
 from ....typing import BrandID
 
@@ -32,19 +34,19 @@ def create_shop(shop_id: ShopID, brand_id: BrandID, title: str) -> Shop:
 
 def delete_shop(shop_id: ShopID) -> None:
     """Delete a shop."""
-    db.session.query(DbShop) \
-        .filter_by(id=shop_id) \
-        .delete()
-
+    db.session.execute(
+        delete(DbShop)
+        .where(DbShop.id == shop_id)
+    )
     db.session.commit()
 
 
 def find_shop_for_brand(brand_id: BrandID) -> Optional[Shop]:
     """Return the shop for that brand, or `None` if not found."""
-    db_shop = db.session \
-        .query(DbShop) \
-        .filter_by(brand_id=brand_id) \
-        .one_or_none()
+    db_shop = db.session.execute(
+        select(DbShop)
+        .filter_by(brand_id=brand_id)
+    ).scalar_one_or_none()
 
     if db_shop is None:
         return None
@@ -97,20 +99,20 @@ def find_shops(shop_ids: set[ShopID]) -> list[Shop]:
     if not shop_ids:
         return []
 
-    db_shops = db.session \
-        .query(DbShop) \
-        .filter(DbShop.id.in_(shop_ids)) \
-        .all()
+    db_shops = db.session.execute(
+        select(DbShop)
+        .filter(DbShop.id.in_(shop_ids))
+    ).scalars().all()
 
     return [_db_entity_to_shop(db_shop) for db_shop in db_shops]
 
 
 def get_active_shops() -> list[Shop]:
     """Return all shops that are not archived."""
-    db_shops = db.session \
-        .query(DbShop) \
-        .filter_by(archived=False) \
-        .all()
+    db_shops = db.session.execute(
+        select(DbShop)
+        .filter_by(archived=False)
+    ).scalars().all()
 
     return [_db_entity_to_shop(db_shop) for db_shop in db_shops]
 

byceps/services/shop/storefront/service.py

@@ -9,6 +9,8 @@
 from __future__ import annotations
 from typing import Optional
 
+from sqlalchemy import delete, select
+
 from ....database import db
 
 from ..catalog.transfer.models import CatalogID
@@ -66,10 +68,10 @@ def update_storefront(
 
 def delete_storefront(storefront_id: StorefrontID) -> None:
     """Delete a storefront."""
-    db.session.query(DbStorefront) \
-        .filter_by(id=storefront_id) \
-        .delete()
-
+    db.session.execute(
+        delete(DbStorefront)
+        .where(DbStorefront.id == storefront_id)
+    )
     db.session.commit()
 
 
@@ -118,10 +120,10 @@ def find_storefronts(storefront_ids: set[StorefrontID]) -> list[Storefront]:
     if not storefront_ids:
         return []
 
-    db_storefronts = db.session \
-        .query(DbStorefront) \
-        .filter(DbStorefront.id.in_(storefront_ids)) \
-        .all()
+    db_storefronts = db.session.execute(
+        select(DbStorefront)
+        .filter(DbStorefront.id.in_(storefront_ids))
+    ).scalars().all()
 
     return [
         _db_entity_to_storefront(db_storefront)
@@ -131,7 +133,9 @@ def find_storefronts(storefront_ids: set[StorefrontID]) -> list[Storefront]:
 
 def get_all_storefronts() -> list[Storefront]:
     """Return all storefronts."""
-    db_storefronts = db.session.query(DbStorefront).all()
+    db_storefronts = db.session.execute(
+        select(DbStorefront)
+    ).scalars().all()
 
     return [
         _db_entity_to_storefront(db_storefront)
@@ -141,10 +145,10 @@ def get_all_storefronts() -> list[Storefront]:
 
 def get_storefronts_for_shop(shop_id: ShopID) -> set[Storefront]:
     """Return all storefronts for that shop."""
-    rows = db.session \
-        .query(DbStorefront) \
-        .filter_by(shop_id=shop_id) \
-        .all()
+    rows = db.session.execute(
+        select(DbStorefront)
+        .filter_by(shop_id=shop_id)
+    ).scalars().all()
 
     return {_db_entity_to_storefront(row) for row in rows}