merge demos, http replay fixes

src/demo_substrate_events/handlers/on_transfer.py

@@ -1,27 +1,54 @@
+from decimal import Decimal
+
 from demo_substrate_events import models as models
 from demo_substrate_events.types.assethub.substrate_events.assets_transferred import AssetsTransferredPayload
 from dipdup.context import HandlerContext
 from dipdup.models.substrate import SubstrateEvent
+from tortoise.exceptions import DoesNotExist
+
+
+async def sql_update(
+    ctx: HandlerContext,
+    address: str,
+    amount: Decimal,
+    level: int,
+) -> None:
+    await ctx.execute_sql_query(
+        'update_balance',
+        address,
+        str(amount),
+        level,
+    )
+
+
+async def orm_update(
+    ctx: HandlerContext,
+    address: str,
+    amount: Decimal,
+    level: int,
+) -> None:
+    try:
+        holder = await models.Holder.cached_get(pk=address)
+    except DoesNotExist:
+        holder = models.Holder(address=address)
+        holder.cache()
+    holder.balance += amount
+    holder.turnover += abs(amount)
+    holder.tx_count += 1
+    holder.last_seen = level
+    await holder.save()
 
 
 async def on_transfer(
     ctx: HandlerContext,
     event: SubstrateEvent[AssetsTransferredPayload],
 ) -> None:
-    amount = event.payload.get('amount') or event.payload['value']
+    amount = Decimal(event.payload.get('amount') or event.payload['value'])
     if not amount:
         return
 
-    await ctx.execute_sql_query(
-        'update_balance',
-        event.payload['from'],
-        '-' + amount,
-        event.data.level,
-    )
+    await sql_update(ctx, event.payload['from'], -amount, event.data.level)
+    await sql_update(ctx, event.payload['to'], amount, event.data.level)
 
-    await ctx.execute_sql_query(
-        'update_balance',
-        event.payload['to'],
-        amount,
-        event.data.level,
-    )
+    # await orm_update(ctx, event.payload['from'], -amount, event.data.level)
+    # await orm_update(ctx, event.payload['to'], amount, event.data.level)

src/demo_substrate_events/models/__init__.py

@@ -1,10 +1,13 @@
 from dipdup import fields
-from dipdup.models import Model
+from dipdup.models import CachedModel
 
 
-class Holder(Model):
+class Holder(CachedModel):
     address = fields.TextField(primary_key=True)
     balance = fields.DecimalField(decimal_places=6, max_digits=40, default=0)
     turnover = fields.DecimalField(decimal_places=6, max_digits=40, default=0)
     tx_count = fields.BigIntField(default=0)
     last_seen = fields.BigIntField(null=True)
+
+    class Meta:
+        maxsize = 2**12

src/demo_substrate_events/sql/update_balance.sql

@@ -1,16 +1,22 @@
-insert into
-    holder (
-        [address],
-        balance,
-        turnover,
-        tx_count,
-        last_seen
-    )
-values
-    (:address, :balance, abs(:balance), 1, :level) on conflict ([address]) do
+insert into holder (
+    address
+    ,balance
+    ,turnover
+    ,tx_count
+    ,last_seen
+)
+values (
+    :address
+    ,:amount
+    ,abs(:amount)
+    ,1
+    ,:level
+)
+on conflict (address) do
 update
 set
-    balance = balance + :balance,
-    turnover = turnover + abs(:balance),
-    tx_count = tx_count + 1,
-    last_seen = :level;
+    balance = balance + :amount
+    ,turnover = turnover + abs(:amount)
+    ,tx_count = tx_count + 1
+    ,last_seen = :level
+;