♻️Storage: use transactions helpers (#7343)

Co-authored-by: Pedro Crespo-Valero <[email protected]>

Author: sanderegg

packages/postgres-database/src/simcore_postgres_database/utils_repos.py

@@ -42,7 +42,7 @@ async def pass_or_acquire_connection(
 @asynccontextmanager
 async def transaction_context(
     engine: AsyncEngine, connection: AsyncConnection | None = None
-):
+) -> AsyncIterator[AsyncConnection]:
     """
     When to use: For WRITE operations!
     This function manages the database connection and ensures that a transaction context is established for write operations.

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/modules/db/credit_transactions_db.py

@@ -17,6 +17,7 @@
     resource_tracker_service_runs,
 )
 from simcore_postgres_database.utils_repos import transaction_context
+from sqlalchemy.engine import CursorResult
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 from ....exceptions.errors import CreditTransactionNotCreatedDBError
@@ -165,6 +166,8 @@ async def batch_update_credit_transaction_status_for_in_debt_transactions(
         )
     async with transaction_context(engine, connection) as conn:
         result = await conn.execute(update_stmt)
+        # NOTE: see https://docs.sqlalchemy.org/en/20/tutorial/data_update.html#getting-affected-row-count-from-update-delete
+        assert isinstance(result, CursorResult)  # nosec
         if result.rowcount:
             _logger.info(
                 "Wallet %s and project %s transactions in DEBT were changed to BILLED. Num. of transaction %s",

services/storage/src/simcore_service_storage/datcore_dsm.py

@@ -33,7 +33,8 @@
 from .modules.datcore_adapter.datcore_adapter_exceptions import (
     DatcoreAdapterMultipleFilesError,
 )
-from .modules.db.tokens import get_api_token_and_secret
+from .modules.db import get_db_engine
+from .modules.db.tokens import TokenRepository
 
 
 def _check_api_credentials(
@@ -60,7 +61,9 @@ class DatCoreDataManager(BaseDataManager):
     async def _get_datcore_tokens(
         self, user_id: UserID
     ) -> tuple[str | None, str | None]:
-        return await get_api_token_and_secret(self.app, user_id)
+        return await TokenRepository.instance(
+            get_db_engine(self.app)
+        ).get_api_token_and_secret(user_id=user_id)
 
     @classmethod
     def get_location_id(cls) -> LocationID:

services/storage/src/simcore_service_storage/modules/db/_base.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+from typing import TypeVar
+
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+RepositoryType = TypeVar("RepositoryType", bound="BaseRepository")
+
+
+@dataclass
+class BaseRepository:
+    db_engine: AsyncEngine
+
+    @classmethod
+    def instance(cls: type[RepositoryType], db_engine: AsyncEngine) -> RepositoryType:
+        return cls(db_engine=db_engine)