fix issue with publishing

Author: matusdrobuliak66

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/credit_transactions.py

@@ -61,9 +61,10 @@ async def create_credit_transaction(
 
         wallet_total_credits = await sum_credit_transactions_and_publish_to_rabbitmq(
             db_engine,
-            rabbitmq_client,
-            credit_transaction_create_body.product_name,
-            credit_transaction_create_body.wallet_id,
+            connection=conn,
+            rabbitmq_client=rabbitmq_client,
+            product_name=credit_transaction_create_body.product_name,
+            wallet_id=credit_transaction_create_body.wallet_id,
         )
         if wallet_total_credits.available_osparc_credits >= 0:
             # Change status from `IN_DEBT` to `BILLED`
@@ -196,19 +197,19 @@ async def pay_project_debt(
     fire_and_forget_task(
         sum_credit_transactions_and_publish_to_rabbitmq(
             db_engine,
-            rabbitmq_client,
-            new_wallet_transaction_create.product_name,
-            new_wallet_transaction_create.wallet_id,  # <-- New wallet
+            rabbitmq_client=rabbitmq_client,
+            product_name=new_wallet_transaction_create.product_name,
+            wallet_id=new_wallet_transaction_create.wallet_id,  # <-- New wallet
         ),
         task_suffix_name=f"sum_and_publish_credits_wallet_id{new_wallet_transaction_create.wallet_id}",
         fire_and_forget_tasks_collection=rut_fire_and_forget_tasks,
     )
     fire_and_forget_task(
         sum_credit_transactions_and_publish_to_rabbitmq(
             db_engine,
-            rabbitmq_client,
-            current_wallet_transaction_create.product_name,
-            current_wallet_transaction_create.wallet_id,  # <-- Current wallet
+            rabbitmq_client=rabbitmq_client,
+            product_name=current_wallet_transaction_create.product_name,
+            wallet_id=current_wallet_transaction_create.wallet_id,  # <-- Current wallet
         ),
         task_suffix_name=f"sum_and_publish_credits_wallet_id{current_wallet_transaction_create.wallet_id}",
         fire_and_forget_tasks_collection=rut_fire_and_forget_tasks,

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/licensed_items_purchases.py

@@ -157,7 +157,10 @@ async def create_licensed_item_purchase(
 
     # Publish wallet total credits to RabbitMQ
     await sum_credit_transactions_and_publish_to_rabbitmq(
-        db_engine, rabbitmq_client, data.product_name, data.wallet_id
+        db_engine,
+        rabbitmq_client=rabbitmq_client,
+        product_name=data.product_name,
+        wallet_id=data.wallet_id,
     )
 
     return LicensedItemPurchaseGet(

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/process_message_running_service.py

@@ -158,7 +158,10 @@ async def _process_start_event(
 
         # Publish wallet total credits to RabbitMQ
         await sum_credit_transactions_and_publish_to_rabbitmq(
-            db_engine, rabbitmq_client, msg.product_name, msg.wallet_id
+            db_engine,
+            rabbitmq_client=rabbitmq_client,
+            product_name=msg.product_name,
+            wallet_id=msg.wallet_id,
         )
 
 
@@ -216,9 +219,9 @@ async def _process_heartbeat_event(
         # Publish wallet total credits to RabbitMQ
         wallet_total_credits = await sum_credit_transactions_and_publish_to_rabbitmq(
             db_engine,
-            rabbitmq_client,
-            running_service.product_name,
-            running_service.wallet_id,
+            rabbitmq_client=rabbitmq_client,
+            product_name=running_service.product_name,
+            wallet_id=running_service.wallet_id,
         )
         if wallet_total_credits.available_osparc_credits < CreditsLimit.OUT_OF_CREDITS:
             await publish_to_rabbitmq_wallet_credits_limit_reached(
@@ -319,9 +322,9 @@ async def _process_stop_event(
         # Publish wallet total credits to RabbitMQ
         await sum_credit_transactions_and_publish_to_rabbitmq(
             db_engine,
-            rabbitmq_client,
-            running_service.product_name,
-            running_service.wallet_id,
+            rabbitmq_client=rabbitmq_client,
+            product_name=running_service.product_name,
+            wallet_id=running_service.wallet_id,
         )
 
 

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/utils.py

@@ -20,7 +20,7 @@
 from models_library.wallets import WalletID
 from pydantic import PositiveInt
 from servicelib.rabbitmq import RabbitMQClient
-from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 from .modules.db import credit_transactions_db, service_runs_db
 
@@ -33,12 +33,15 @@ def make_negative(n):
 
 async def sum_credit_transactions_and_publish_to_rabbitmq(
     db_engine: AsyncEngine,
+    connection: AsyncConnection | None = None,
+    *,
     rabbitmq_client: RabbitMQClient,
     product_name: ProductName,
     wallet_id: WalletID,
 ) -> WalletTotalCredits:
     wallet_total_credits = await credit_transactions_db.sum_wallet_credits(
         db_engine,
+        connection=connection,
         product_name=product_name,
         wallet_id=wallet_id,
     )