🐛 Fix: rename PaymentsTransactionsDB to PaymentsTransactionsGetDB for consistency and update database connection handling to use asyncpg engine

Author: pcrespov

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

@@ -5,17 +5,16 @@
 from typing import Final, TypeAlias
 
 import sqlalchemy as sa
-from aiopg.sa.connection import SAConnection
-from aiopg.sa.result import ResultProxy, RowProxy
+import sqlalchemy.exc
+from sqlalchemy.ext.asyncio import AsyncConnection
 
-from . import aiopg_errors
 from .models.payments_transactions import PaymentTransactionState, payments_transactions
 
 _logger = logging.getLogger(__name__)
 
 
 PaymentID: TypeAlias = str
-PaymentTransactionRow: TypeAlias = RowProxy
+PaymentTransactionRow: TypeAlias = sa.Row
 
 
 UNSET: Final[str] = "__UNSET__"
@@ -39,7 +38,7 @@ class PaymentAlreadyAcked(PaymentFailure): ...
 
 
 async def insert_init_payment_transaction(
-    connection: SAConnection,
+    connection: AsyncConnection,
     *,
     payment_id: str,
     price_dollars: Decimal,
@@ -66,14 +65,14 @@ async def insert_init_payment_transaction(
                 initiated_at=initiated_at,
             )
         )
-    except aiopg_errors.UniqueViolation:
+    except sqlalchemy.exc.IntegrityError:
         return PaymentAlreadyExists(payment_id)
 
     return payment_id
 
 
 async def update_payment_transaction_state(
-    connection: SAConnection,
+    connection: AsyncConnection,
     *,
     payment_id: str,
     completion_state: PaymentTransactionState,
@@ -101,16 +100,15 @@ async def update_payment_transaction_state(
         optional["invoice_url"] = invoice_url
 
     async with connection.begin():
-        row = await (
-            await connection.execute(
-                sa.select(
-                    payments_transactions.c.initiated_at,
-                    payments_transactions.c.completed_at,
-                )
-                .where(payments_transactions.c.payment_id == payment_id)
-                .with_for_update()
+        result = await connection.execute(
+            sa.select(
+                payments_transactions.c.initiated_at,
+                payments_transactions.c.completed_at,
             )
-        ).fetchone()
+            .where(payments_transactions.c.payment_id == payment_id)
+            .with_for_update()
+        )
+        row = result.one_or_none()
 
         if row is None:
             return PaymentNotFound(payment_id=payment_id)
@@ -125,16 +123,14 @@ async def update_payment_transaction_state(
             payments_transactions.update()
             .values(completed_at=sa.func.now(), state=completion_state, **optional)
             .where(payments_transactions.c.payment_id == payment_id)
-            .returning(sa.literal_column("*"))
+            .returning(payments_transactions)
         )
-        row = await result.first()
-        assert row, "execute above should have caught this"  # nosec
-        assert isinstance(row, RowProxy)  # nosec
+        row = result.one()
         return row
 
 
 async def get_user_payments_transactions(
-    connection: SAConnection,
+    connection: AsyncConnection,
     *,
     user_id: int,
     offset: int | None = None,
@@ -149,7 +145,7 @@ async def get_user_payments_transactions(
 
     # NOTE: what if between these two calls there are new rows? can we get this in an atomic call?å
     stmt = (
-        payments_transactions.select()
+        sa.select(payments_transactions)
         .where(payments_transactions.c.user_id == user_id)
         .order_by(payments_transactions.c.created.desc())
     )  # newest first
@@ -162,6 +158,6 @@ async def get_user_payments_transactions(
         # InvalidRowCountInLimitClause: LIMIT must not be negative
         stmt = stmt.limit(limit)
 
-    result: ResultProxy = await connection.execute(stmt)
-    rows = await result.fetchall() or []
+    result = await connection.execute(stmt)
+    rows = result.fetchall()
     return total_number_of_items, rows

packages/postgres-database/tests/test_models_payments_transactions.py

@@ -10,8 +10,6 @@
 
 import pytest
 import sqlalchemy as sa
-from aiopg.sa.connection import SAConnection
-from aiopg.sa.result import RowProxy
 from faker import Faker
 from pytest_simcore.helpers.faker_factories import random_payment_transaction, utcnow
 from simcore_postgres_database.models.payments_transactions import (
@@ -26,9 +24,10 @@
     insert_init_payment_transaction,
     update_payment_transaction_state,
 )
+from sqlalchemy.ext.asyncio import AsyncConnection
 
 
-async def test_numerics_precission_and_scale(connection: SAConnection):
+async def test_numerics_precission_and_scale(connection: AsyncConnection):
     # https://docs.sqlalchemy.org/en/20/core/type_basics.html#sqlalchemy.types.Numeric
     # precision: This parameter specifies the total number of digits that can be stored, both before and after the decimal point.
     # scale: This parameter specifies the number of digits that can be stored to the right of the decimal point.
@@ -58,7 +57,7 @@ def _remove_not_required(data: dict[str, Any]) -> dict[str, Any]:
 
 
 @pytest.fixture
-def init_transaction(connection: SAConnection):
+def init_transaction(connection: AsyncConnection):
     async def _init(payment_id: str):
         # get payment_id from payment-gateway
         values = _remove_not_required(random_payment_transaction(payment_id=payment_id))
@@ -81,7 +80,7 @@ def payment_id() -> str:
 
 
 async def test_init_transaction_sets_it_as_pending(
-    connection: SAConnection, init_transaction: Callable, payment_id: str
+    connection: AsyncConnection, init_transaction: Callable, payment_id: str
 ):
     values = await init_transaction(payment_id)
     assert values["payment_id"] == payment_id
@@ -94,11 +93,11 @@ async def test_init_transaction_sets_it_as_pending(
             payments_transactions.c.state_message,
         ).where(payments_transactions.c.payment_id == payment_id)
     )
-    row: RowProxy | None = await result.fetchone()
+    row = result.one_or_none()
     assert row is not None
 
     # tests that defaults are right?
-    assert dict(row.items()) == {
+    assert dict(row._mapping.items()) == {
         "completed_at": None,
         "state": PaymentTransactionState.PENDING,
         "state_message": None,
@@ -127,7 +126,7 @@ def invoice_url(faker: Faker, expected_state: PaymentTransactionState) -> str |
     ],
 )
 async def test_complete_transaction(
-    connection: SAConnection,
+    connection: AsyncConnection,
     init_transaction: Callable,
     payment_id: str,
     expected_state: PaymentTransactionState,
@@ -152,7 +151,7 @@ async def test_complete_transaction(
 
 
 async def test_update_transaction_failures_and_exceptions(
-    connection: SAConnection,
+    connection: AsyncConnection,
     init_transaction: Callable,
     payment_id: str,
 ):
@@ -188,7 +187,9 @@ def user_id() -> int:
 
 
 @pytest.fixture
-def create_fake_user_transactions(connection: SAConnection, user_id: int) -> Callable:
+def create_fake_user_transactions(
+    connection: AsyncConnection, user_id: int
+) -> Callable:
     async def _go(expected_total=5):
         payment_ids = []
         for _ in range(expected_total):
@@ -204,7 +205,7 @@ async def _go(expected_total=5):
 
 
 async def test_get_user_payments_transactions(
-    connection: SAConnection, create_fake_user_transactions: Callable, user_id: int
+    connection: AsyncConnection, create_fake_user_transactions: Callable, user_id: int
 ):
     expected_payments_ids = await create_fake_user_transactions()
     expected_total = len(expected_payments_ids)
@@ -216,7 +217,7 @@ async def test_get_user_payments_transactions(
 
 
 async def test_get_user_payments_transactions_with_pagination_options(
-    connection: SAConnection, create_fake_user_transactions: Callable, user_id: int
+    connection: AsyncConnection, create_fake_user_transactions: Callable, user_id: int
 ):
     expected_payments_ids = await create_fake_user_transactions()
     expected_total = len(expected_payments_ids)
@@ -244,3 +245,4 @@ async def test_get_user_payments_transactions_with_pagination_options(
         connection, user_id=user_id, limit=0
     )
     assert not rows
+    assert not rows

services/web/server/src/simcore_service_webserver/payments/_onetime_api.py

@@ -24,7 +24,7 @@
 from simcore_postgres_database.utils_payments import insert_init_payment_transaction
 from yarl import URL
 
-from ..db.plugin import get_database_engine_legacy
+from ..db.plugin import get_asyncpg_engine
 from ..products import products_service
 from ..resource_usage.service import add_credits_to_wallet
 from ..users import users_service
@@ -46,7 +46,7 @@
 
 
 def _to_api_model(
-    transaction: _onetime_db.PaymentsTransactionsDB,
+    transaction: _onetime_db.PaymentsTransactionsGetDB,
 ) -> PaymentTransaction:
     data: dict[str, Any] = {
         "payment_id": transaction.payment_id,
@@ -90,7 +90,7 @@ async def _fake_init_payment(
         .with_query(id=payment_id)
     )
     # (2) Annotate INIT transaction
-    async with get_database_engine_legacy(app).acquire() as conn:
+    async with get_asyncpg_engine(app).begin() as conn:
         await insert_init_payment_transaction(
             conn,
             payment_id=payment_id,

services/web/server/src/simcore_service_webserver/payments/_onetime_db.py

@@ -20,8 +20,13 @@
     get_user_payments_transactions,
     update_payment_transaction_state,
 )
+from simcore_postgres_database.utils_repos import (
+    pass_or_acquire_connection,
+    transaction_context,
+)
+from sqlalchemy.ext.asyncio import AsyncConnection
 
-from ..db.plugin import get_database_engine_legacy
+from ..db.plugin import get_asyncpg_engine
 from .errors import PaymentCompletedError, PaymentNotFoundError
 
 _logger = logging.getLogger(__name__)
@@ -30,7 +35,7 @@
 #
 # NOTE: this will be moved to the payments service
 # NOTE: with https://sqlmodel.tiangolo.com/ we would only define this once!
-class PaymentsTransactionsDB(BaseModel):
+class PaymentsTransactionsGetDB(BaseModel):
     payment_id: PaymentID
     price_dollars: Decimal  # accepts negatives
     osparc_credits: Decimal  # accepts negatives
@@ -48,43 +53,47 @@ class PaymentsTransactionsDB(BaseModel):
 
 
 async def list_user_payment_transactions(
-    app,
+    app: web.Application,
+    connection: AsyncConnection | None = None,
     *,
     user_id: UserID,
     offset: PositiveInt,
     limit: PositiveInt,
-) -> tuple[int, list[PaymentsTransactionsDB]]:
+) -> tuple[int, list[PaymentsTransactionsGetDB]]:
     """List payments done by a give user (any wallet)
 
     Sorted by newest-first
     """
-    async with get_database_engine_legacy(app).acquire() as conn:
+    async with pass_or_acquire_connection(get_asyncpg_engine(app), connection) as conn:
         total_number_of_items, rows = await get_user_payments_transactions(
             conn, user_id=user_id, offset=offset, limit=limit
         )
-        page = TypeAdapter(list[PaymentsTransactionsDB]).validate_python(rows)
+        page = TypeAdapter(list[PaymentsTransactionsGetDB]).validate_python(rows)
         return total_number_of_items, page
 
 
-async def get_pending_payment_transactions_ids(app: web.Application) -> list[PaymentID]:
-    async with get_database_engine_legacy(app).acquire() as conn:
+async def get_pending_payment_transactions_ids(
+    app: web.Application, connection: AsyncConnection | None = None
+) -> list[PaymentID]:
+    async with pass_or_acquire_connection(get_asyncpg_engine(app), connection) as conn:
         result = await conn.execute(
             sa.select(payments_transactions.c.payment_id)
             .where(payments_transactions.c.completed_at == None)  # noqa: E711
             .order_by(payments_transactions.c.initiated_at.asc())  # oldest first
         )
-        rows = await result.fetchall() or []
+        rows = result.fetchall()
         return [TypeAdapter(PaymentID).validate_python(row.payment_id) for row in rows]
 
 
 async def complete_payment_transaction(
     app: web.Application,
+    connection: AsyncConnection | None = None,
     *,
     payment_id: PaymentID,
     completion_state: PaymentTransactionState,
     state_message: str | None,
     invoice_url: HttpUrl | None = None,
-) -> PaymentsTransactionsDB:
+) -> PaymentsTransactionsGetDB:
     """
 
     Raises:
@@ -95,7 +104,7 @@ async def complete_payment_transaction(
     if invoice_url:
         optional_kwargs["invoice_url"] = invoice_url
 
-    async with get_database_engine_legacy(app).acquire() as conn:
+    async with transaction_context(get_asyncpg_engine(app), connection) as conn:
         row = await update_payment_transaction_state(
             conn,
             payment_id=payment_id,
@@ -111,4 +120,4 @@ async def complete_payment_transaction(
             raise PaymentCompletedError(payment_id=row.payment_id)
 
         assert row  # nosec
-        return PaymentsTransactionsDB.model_validate(row)
+        return PaymentsTransactionsGetDB.model_validate(row)