dataclass

Author: pcrespov

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

@@ -9,7 +9,7 @@
 from servicelib.rabbitmq import RPCRouter
 
 from ..products import products_service
-from ..products.models import CreditResultDict
+from ..products.models import CreditResult
 from ..rabbitmq import get_rabbitmq_rpc_server
 from ..users.api import get_user_display_and_id_names, get_user_invoice_address
 
@@ -24,7 +24,7 @@ async def get_invoice_data(
     dollar_amount: Decimal,
     product_name: ProductName,
 ) -> InvoiceDataGet:
-    credit_result: CreditResultDict = await products_service.get_credit_amount(
+    credit_result: CreditResult = await products_service.get_credit_amount(
         app, dollar_amount=dollar_amount, product_name=product_name
     )
     product_stripe_info = await products_service.get_product_stripe_info(
@@ -36,7 +36,7 @@ async def get_invoice_data(
     user_info = await get_user_display_and_id_names(app, user_id=user_id)
 
     return InvoiceDataGet(
-        credit_amount=credit_result["credit_amount"],
+        credit_amount=credit_result.credit_amount,
         stripe_price_id=product_stripe_info.stripe_price_id,
         stripe_tax_rate_id=product_stripe_info.stripe_tax_rate_id,
         user_invoice_address=user_invoice_address,

services/web/server/src/simcore_service_webserver/products/_controller/rpc.py

@@ -9,7 +9,7 @@
 from ...constants import APP_SETTINGS_KEY
 from ...rabbitmq import get_rabbitmq_rpc_server, setup_rabbitmq
 from .. import _service
-from .._models import CreditResultDict
+from .._models import CreditResult
 
 router = RPCRouter()
 
@@ -21,10 +21,10 @@ async def get_credit_amount(
     dollar_amount: Decimal,
     product_name: ProductName,
 ) -> CreditResultRpcGet:
-    credit_result: CreditResultDict = await _service.get_credit_amount(
+    credit_result: CreditResult = await _service.get_credit_amount(
         app, dollar_amount=dollar_amount, product_name=product_name
     )
-    return CreditResultRpcGet.model_validate(credit_result)
+    return CreditResultRpcGet.model_validate(credit_result, from_attributes=True)
 
 
 async def _register_rpc_routes_on_startup(app: web.Application):

services/web/server/src/simcore_service_webserver/products/_models.py

@@ -1,8 +1,9 @@
 import logging
 import re
 import string
+from dataclasses import dataclass
 from decimal import Decimal
-from typing import Annotated, Any, NamedTuple
+from typing import Annotated, Any
 
 from models_library.basic_regex import (
     PUBLIC_VARIABLE_NAME_RE,
@@ -32,24 +33,26 @@
     WebFeedback,
     products,
 )
-from typing_extensions import TypedDict
 
 from ..constants import FRONTEND_APPS_AVAILABLE
 
 _logger = logging.getLogger(__name__)
 
 
-class CreditResultDict(TypedDict):
+@dataclass(frozen=True)
+class CreditResult:
     product_name: ProductName
     credit_amount: Decimal
 
 
-class ProductStripeInfo(NamedTuple):
+@dataclass(frozen=True)
+class ProductStripeInfo:
     stripe_price_id: StripePriceID
     stripe_tax_rate_id: StripeTaxRateID
 
 
-class PaymentFieldsTuple(NamedTuple):
+@dataclass(frozen=True)
+class PaymentFields:
     enabled: bool
     credits_per_usd: Decimal | None
     min_payment_amount_usd: Decimal | None

services/web/server/src/simcore_service_webserver/products/_repository.py

@@ -27,7 +27,7 @@
 
 from ..constants import FRONTEND_APPS_AVAILABLE
 from ..db.base_repository import BaseRepository
-from ._models import PaymentFieldsTuple, Product, ProductStripeInfo
+from ._models import PaymentFields, Product, ProductStripeInfo
 
 _logger = logging.getLogger(__name__)
 
@@ -60,7 +60,7 @@
 )
 
 
-def _to_domain(products_row: Row, payments: PaymentFieldsTuple) -> Product:
+def _to_domain(products_row: Row, payments: PaymentFields) -> Product:
     return Product(
         **products_row._asdict(),
         is_payment_enabled=payments.enabled,
@@ -70,12 +70,12 @@ def _to_domain(products_row: Row, payments: PaymentFieldsTuple) -> Product:
 
 async def _get_product_payment_fields(
     conn: AsyncConnection, product_name: ProductName
-) -> PaymentFieldsTuple:
+) -> PaymentFields:
     price_info = await get_product_latest_price_info_or_none(
         conn, product_name=product_name
     )
     if price_info is None or price_info.usd_per_credit == 0:
-        return PaymentFieldsTuple(
+        return PaymentFields(
             enabled=False,
             credits_per_usd=None,
             min_payment_amount_usd=None,
@@ -84,7 +84,7 @@ async def _get_product_payment_fields(
     assert price_info.usd_per_credit > 0  # nosec
     assert price_info.min_payment_amount_usd > 0  # nosec
 
-    return PaymentFieldsTuple(
+    return PaymentFields(
         enabled=True,
         credits_per_usd=Decimal(1 / price_info.usd_per_credit).quantize(
             QUANTIZE_EXP_ARG

services/web/server/src/simcore_service_webserver/products/_service.py

@@ -9,7 +9,7 @@
 from simcore_postgres_database.utils_products_prices import ProductPriceInfo
 
 from ..constants import APP_PRODUCTS_KEY
-from ._models import CreditResultDict, ProductStripeInfo
+from ._models import CreditResult, ProductStripeInfo
 from ._repository import ProductRepository
 from .errors import (
     BelowMinimumPaymentError,
@@ -77,7 +77,7 @@ async def get_credit_amount(
     *,
     dollar_amount: Decimal,
     product_name: ProductName,
-) -> CreditResultDict:
+) -> CreditResult:
     """For provided dollars and product gets credit amount.
 
     NOTE: Contrary to other product api functions (e.g. get_current_product) this function
@@ -105,7 +105,7 @@ async def get_credit_amount(
         )
 
     credit_amount = dollar_amount / price_info.usd_per_credit
-    return CreditResultDict(product_name=product_name, credit_amount=credit_amount)
+    return CreditResult(product_name=product_name, credit_amount=credit_amount)
 
 
 async def get_product_stripe_info(

services/web/server/src/simcore_service_webserver/products/models.py

@@ -1,9 +1,9 @@
 from models_library.products import ProductName
 
-from ._models import CreditResultDict, Product, ProductStripeInfo
+from ._models import CreditResult, Product, ProductStripeInfo
 
 __all__: tuple[str, ...] = (
-    "CreditResultDict",
+    "CreditResult",
     "Product",
     "ProductName",
     "ProductStripeInfo",

services/web/server/src/simcore_service_webserver/wallets/_payments_handlers.py

@@ -23,7 +23,7 @@
 )
 from servicelib.logging_utils import get_log_record_extra, log_context
 from servicelib.utils import fire_and_forget_task
-from simcore_service_webserver.products._models import CreditResultDict
+from simcore_service_webserver.products._models import CreditResult
 
 from .._meta import API_VTAG as VTAG
 from ..login.decorators import login_required
@@ -79,7 +79,7 @@ async def _create_payment(request: web.Request):
         log_duration=True,
         extra=get_log_record_extra(user_id=req_ctx.user_id),
     ):
-        credit_result: CreditResultDict = await products_service.get_credit_amount(
+        credit_result: CreditResult = await products_service.get_credit_amount(
             request.app,
             dollar_amount=body_params.price_dollars,
             product_name=req_ctx.product_name,
@@ -90,7 +90,7 @@ async def _create_payment(request: web.Request):
             user_id=req_ctx.user_id,
             product_name=req_ctx.product_name,
             wallet_id=wallet_id,
-            osparc_credits=credit_result["credit_amount"],
+            osparc_credits=credit_result.credit_amount,
             comment=body_params.comment,
             price_dollars=body_params.price_dollars,
         )
@@ -351,7 +351,7 @@ async def _pay_with_payment_method(request: web.Request):
         log_duration=True,
         extra=get_log_record_extra(user_id=req_ctx.user_id),
     ):
-        credit_result: CreditResultDict = await products_service.get_credit_amount(
+        credit_result: CreditResult = await products_service.get_credit_amount(
             request.app,
             dollar_amount=body_params.price_dollars,
             product_name=req_ctx.product_name,
@@ -363,7 +363,7 @@ async def _pay_with_payment_method(request: web.Request):
             product_name=req_ctx.product_name,
             wallet_id=wallet_id,
             payment_method_id=path_params.payment_method_id,
-            osparc_credits=credit_result["credit_amount"],
+            osparc_credits=credit_result.credit_amount,
             comment=body_params.comment,
             price_dollars=body_params.price_dollars,
         )