✨ payments service enables notifications via sio (#5109)

Author: pcrespov

packages/models-library/src/models_library/invitations.py

@@ -1,8 +1,11 @@
 from datetime import datetime, timezone
+from typing import Final
 
 from models_library.products import ProductName
 from pydantic import BaseModel, EmailStr, Field, PositiveInt, validator
 
+_MAX_LEN: Final = 40
+
 
 class InvitationInputs(BaseModel):
     """Input data necessary to create an invitation"""
@@ -11,7 +14,7 @@ class InvitationInputs(BaseModel):
         ...,
         description="Identifies who issued the invitation. E.g. an email, a service name etc. NOTE: it will be trimmed if exceeds maximum",
         min_length=1,
-        max_length=30,
+        max_length=_MAX_LEN,
     )
     guest: EmailStr = Field(
         ...,
@@ -35,7 +38,7 @@ class InvitationInputs(BaseModel):
     @classmethod
     def trim_long_issuers_to_max_length(cls, v):
         if v and isinstance(v, str):
-            return v[:29]
+            return v[: _MAX_LEN - 1]
         return v
 
 

packages/pytest-simcore/src/pytest_simcore/helpers/rawdata_fakers.py

@@ -23,6 +23,7 @@
 from faker import Faker
 from simcore_postgres_database.models.api_keys import api_keys
 from simcore_postgres_database.models.comp_pipeline import StateType
+from simcore_postgres_database.models.groups import groups
 from simcore_postgres_database.models.payments_methods import InitPromptAckFlowState
 from simcore_postgres_database.models.payments_transactions import (
     PaymentTransactionState,
@@ -101,11 +102,15 @@ def random_project(**overrides) -> dict[str, Any]:
 
 
 def random_group(**overrides) -> dict[str, Any]:
+
     data = {
         "name": FAKE.company(),
         "description": FAKE.text(),
         "type": GroupType.STANDARD.name,
     }
+
+    assert set(data.keys()).issubset({c.name for c in groups.columns})  # nosec
+
     data.update(overrides)
     return data
 

packages/service-library/src/servicelib/fastapi/http_client.py

@@ -82,7 +82,11 @@ def set_to_app_state(self, app: FastAPI):
 
     @classmethod
     def pop_from_app_state(cls, app: FastAPI):
-        old = getattr(app.state, cls.app_state_name, None)
+        """
+        Raises:
+            AttributeError: if instance is not in app.state
+        """
+        old = getattr(app.state, cls.app_state_name)
         delattr(app.state, cls.app_state_name)
         return old
 

services/payments/src/simcore_service_payments/api/rest/_acknowledgements.py

@@ -16,8 +16,14 @@
     PaymentMethodID,
 )
 from ...services import payments, payments_methods
+from ...services.notifier import Notifier
 from ...services.resource_usage_tracker import ResourceUsageTrackerApi
-from ._dependencies import get_current_session, get_repository, get_rut_api
+from ._dependencies import (
+    create_repository,
+    get_current_session,
+    get_from_app_state,
+    get_rut_api,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -31,12 +37,13 @@ async def acknowledge_payment(
     ack: AckPayment,
     _session: Annotated[SessionData, Depends(get_current_session)],
     repo_pay: Annotated[
-        PaymentsTransactionsRepo, Depends(get_repository(PaymentsTransactionsRepo))
+        PaymentsTransactionsRepo, Depends(create_repository(PaymentsTransactionsRepo))
     ],
     repo_methods: Annotated[
-        PaymentsMethodsRepo, Depends(get_repository(PaymentsMethodsRepo))
+        PaymentsMethodsRepo, Depends(create_repository(PaymentsMethodsRepo))
     ],
     rut_api: Annotated[ResourceUsageTrackerApi, Depends(get_rut_api)],
+    notifier: Annotated[Notifier, Depends(get_from_app_state(Notifier))],
     background_tasks: BackgroundTasks,
 ):
     """completes (ie. ack) request initated by `/init` on the payments-gateway API"""
@@ -60,26 +67,33 @@ async def acknowledge_payment(
 
     assert f"{payment_id}" == f"{transaction.payment_id}"  # nosec
     background_tasks.add_task(
-        payments.on_payment_completed, transaction, rut_api, notify_enabled=True
+        payments.on_payment_completed, transaction, rut_api, notifier=notifier
     )
 
     if ack.saved:
-        created = await payments_methods.insert_payment_method(
+        inserted = await payments_methods.insert_payment_method(
             repo=repo_methods,
             payment_method_id=ack.saved.payment_method_id,
             user_id=transaction.user_id,
             wallet_id=transaction.wallet_id,
             ack=ack.saved,
         )
-        background_tasks.add_task(payments_methods.on_payment_method_completed, created)
+        background_tasks.add_task(
+            payments_methods.on_payment_method_completed,
+            payment_method=inserted,
+            notifier=notifier,
+        )
 
 
 @router.post("/payments-methods/{payment_method_id}:ack")
 async def acknowledge_payment_method(
     payment_method_id: PaymentMethodID,
     ack: AckPaymentMethod,
     _session: Annotated[SessionData, Depends(get_current_session)],
-    repo: Annotated[PaymentsMethodsRepo, Depends(get_repository(PaymentsMethodsRepo))],
+    repo: Annotated[
+        PaymentsMethodsRepo, Depends(create_repository(PaymentsMethodsRepo))
+    ],
+    notifier: Annotated[Notifier, Depends(get_from_app_state(Notifier))],
     background_tasks: BackgroundTasks,
 ):
     """completes (ie. ack) request initated by `/payments-methods:init` on the payments-gateway API"""
@@ -101,4 +115,8 @@ async def acknowledge_payment_method(
             ) from err
 
         assert f"{payment_method_id}" == f"{acked.payment_method_id}"  # nosec
-        background_tasks.add_task(payments_methods.on_payment_method_completed, acked)
+        background_tasks.add_task(
+            payments_methods.on_payment_method_completed,
+            payment_method=acked,
+            notifier=notifier,
+        )

services/payments/src/simcore_service_payments/api/rest/_dependencies.py

@@ -2,9 +2,10 @@
 from collections.abc import AsyncGenerator, Callable
 from typing import Annotated, cast
 
-from fastapi import Depends, Request
+from fastapi import Depends, FastAPI, Request
 from fastapi.security import OAuth2PasswordBearer
 from servicelib.fastapi.dependencies import get_app, get_reverse_url_mapper
+from servicelib.fastapi.http_client import AppStateMixin
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ..._meta import API_VTAG
@@ -43,19 +44,30 @@ def get_rut_api(request: Request) -> ResourceUsageTrackerApi:
     )
 
 
+def get_from_app_state(app_state_mixin_subclass: type[AppStateMixin]) -> Callable:
+    """Generic getter of app.state objects"""
+
+    def _(app: Annotated[FastAPI, Depends(get_app)]):
+        return app_state_mixin_subclass.get_from_app_state(app)
+
+    return _
+
+
 def get_db_engine(request: Request) -> AsyncEngine:
     engine: AsyncEngine = get_engine(request.app)
     assert engine  # nosec
     return engine
 
 
-def get_repository(repo_type: type[BaseRepository]) -> Callable:
-    async def _get_repo(
+def create_repository(repo_cls: type[BaseRepository]) -> Callable:
+    """Generic object factory of BaseRepository instances"""
+
+    async def _(
         engine: Annotated[AsyncEngine, Depends(get_db_engine)],
     ) -> AsyncGenerator[BaseRepository, None]:
-        yield repo_type(db_engine=engine)
+        yield repo_cls(db_engine=engine)
 
-    return _get_repo
+    return _
 
 
 # Implements `password` flow defined in OAuth2

services/payments/src/simcore_service_payments/core/application.py

@@ -1,5 +1,6 @@
 from fastapi import FastAPI
 from servicelib.fastapi.openapi import override_fastapi_openapi_method
+from simcore_service_payments.services.notifier import setup_notifier
 from simcore_service_payments.services.socketio import setup_socketio
 
 from .._meta import (
@@ -56,6 +57,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
     # Listening to Rabbitmq
     setup_auto_recharge_listener(app)
     setup_socketio(app)
+    setup_notifier(app)
 
     # ERROR HANDLERS
     # ... add here ...

services/payments/src/simcore_service_payments/db/payment_users_repo.py

@@ -0,0 +1,22 @@
+import sqlalchemy as sa
+from models_library.users import GroupID, UserID
+from simcore_postgres_database.models.users import users
+
+from .base import BaseRepository
+
+
+class PaymentsUsersRepo(BaseRepository):
+    # NOTE:
+    # Currently linked to `users` but expected to be linked to `payments_users`
+    # when databases are separated. The latter will be a subset copy of the former.
+    #
+    async def get_primary_group_id(self, user_id: UserID) -> GroupID:
+        async with self.db_engine.begin() as conn:
+            result = await conn.execute(
+                sa.select(users.c.primary_gid).where(users.c.id == user_id)
+            )
+            row = result.first()
+            if row is None:
+                msg = f"{user_id=} not found"
+                raise ValueError(msg)
+            return GroupID(row.primary_gid)

services/payments/src/simcore_service_payments/models/db.py

@@ -2,11 +2,7 @@
 from decimal import Decimal
 from typing import Any, ClassVar
 
-from models_library.api_schemas_webserver.wallets import (
-    PaymentID,
-    PaymentMethodID,
-    PaymentTransaction,
-)
+from models_library.api_schemas_webserver.wallets import PaymentID, PaymentMethodID
 from models_library.emails import LowerCaseEmailStr
 from models_library.products import ProductName
 from models_library.users import UserID
@@ -63,28 +59,6 @@ class Config:
             ]
         }
 
-    def to_api_model(self) -> PaymentTransaction:
-        data: dict[str, Any] = {
-            "payment_id": self.payment_id,
-            "price_dollars": self.price_dollars,
-            "osparc_credits": self.osparc_credits,
-            "wallet_id": self.wallet_id,
-            "created_at": self.initiated_at,
-            "state": self.state,
-            "completed_at": self.completed_at,
-        }
-
-        if self.comment:
-            data["comment"] = self.comment
-
-        if self.state_message:
-            data["state_message"] = self.state_message
-
-        if self.invoice_url:
-            data["invoice_url"] = self.invoice_url
-
-        return PaymentTransaction.parse_obj(data)
-
 
 _EXAMPLE_AFTER_INIT_PAYMENT_METHOD = {
     "payment_method_id": "12345",

services/payments/src/simcore_service_payments/models/db_to_api.py

@@ -0,0 +1,41 @@
+from typing import Any
+
+from models_library.api_schemas_webserver.wallets import (
+    PaymentMethodTransaction,
+    PaymentTransaction,
+)
+
+from .db import PaymentsMethodsDB, PaymentsTransactionsDB
+
+
+def to_payments_api_model(transaction: PaymentsTransactionsDB) -> PaymentTransaction:
+    data: dict[str, Any] = {
+        "payment_id": transaction.payment_id,
+        "price_dollars": transaction.price_dollars,
+        "osparc_credits": transaction.osparc_credits,
+        "wallet_id": transaction.wallet_id,
+        "created_at": transaction.initiated_at,
+        "state": transaction.state,
+        "completed_at": transaction.completed_at,
+    }
+
+    if transaction.comment:
+        data["comment"] = transaction.comment
+
+    if transaction.state_message:
+        data["state_message"] = transaction.state_message
+
+    if transaction.invoice_url:
+        data["invoice_url"] = transaction.invoice_url
+
+    return PaymentTransaction(**data)
+
+
+def to_payment_method_api_model(
+    payment_method: PaymentsMethodsDB,
+) -> PaymentMethodTransaction:
+    return PaymentMethodTransaction(
+        wallet_id=payment_method.wallet_id,
+        payment_method_id=payment_method.payment_method_id,
+        state=payment_method.state.value,
+    )