use asyncio lock for updating PrismaClient txs

litellm/proxy/db/redis_update_buffer.py

@@ -80,6 +80,7 @@ class RedisUpdateBuffer:
                 "redis_cache is None, skipping store_in_memory_spend_updates_in_redis"
             )
             return
+        async with prisma_client.in_memory_transaction_lock:
             db_spend_update_transactions: DBSpendUpdateTransactions = (
                 DBSpendUpdateTransactions(
                     user_list_transactions=prisma_client.user_list_transactions,
@@ -93,7 +94,9 @@ class RedisUpdateBuffer:
 
             # only store in redis if there are any updates to commit
             if (
-            self._number_of_transactions_to_store_in_redis(db_spend_update_transactions)
+                self._number_of_transactions_to_store_in_redis(
+                    db_spend_update_transactions
+                )
                 == 0
             ):
                 return
@@ -103,7 +106,9 @@ class RedisUpdateBuffer:
                 key=REDIS_UPDATE_BUFFER_KEY,
                 values=list_of_transactions,
             )
-        self._clear_all_in_memory_spend_updates(prisma_client)
+
+            # clear the in-memory spend updates
+            RedisUpdateBuffer._clear_all_in_memory_spend_updates(prisma_client)
 
     @staticmethod
     def _number_of_transactions_to_store_in_redis(
@@ -201,12 +206,15 @@ class RedisUpdateBuffer:
 
     @staticmethod
     def _parse_list_of_transactions(
-        list_of_transactions: List[str],
+        list_of_transactions: Union[Any, List[Any]],
     ) -> List[DBSpendUpdateTransactions]:
         """
         Parses the list of transactions from Redis
         """
+        if isinstance(list_of_transactions, list):
             return [json.loads(transaction) for transaction in list_of_transactions]
+        else:
+            return [json.loads(list_of_transactions)]
 
     @staticmethod
     def _combine_list_of_transactions(

litellm/proxy/utils.py

@@ -1121,6 +1121,7 @@ class PrismaClient:
         self.iam_token_db_auth: Optional[bool] = str_to_bool(
             os.getenv("IAM_TOKEN_DB_AUTH")
         )
+        self.in_memory_transaction_lock = asyncio.Lock()
         verbose_proxy_logger.debug("Creating Prisma Client..")
         try:
             from prisma import Prisma  # type: ignore