Merge pull request #1574 from BoostryJP/feature/improve-indexer-transfer

Update data schema of lock/unlock event

app/model/db/__init__.py

@@ -42,6 +42,7 @@
 )
 from .idx_token_list import IDXTokenListBlockNumber, IDXTokenListItem
 from .idx_transfer import (
+    DataMessage,
     IDXTransfer,
     IDXTransferBlockNumber,
     IDXTransferSourceEventType,

app/model/db/idx_transfer.py

@@ -19,8 +19,10 @@
 
 from datetime import datetime, timedelta, timezone
 from enum import StrEnum
+from typing import Literal
 from zoneinfo import ZoneInfo
 
+from pydantic import BaseModel
 from sqlalchemy import JSON, BigInteger, String
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -38,6 +40,14 @@ class IDXTransferSourceEventType(StrEnum):
     UNLOCK = "Unlock"
 
 
+class DataMessage(BaseModel):
+    message: Literal[
+        "garnishment",
+        "inheritance",
+        "force_unlock",
+    ]
+
+
 class IDXTransfer(Base):
     """Token Transfer Events (INDEX)"""
 
@@ -58,7 +68,17 @@ class IDXTransfer(Base):
     # Source Event (IDXTransferSourceEventType)
     source_event: Mapped[str] = mapped_column(String(50), nullable=False)
     # Data
+    #   source_event = "Transfer"
+    #     => None
+    #   source_event = "Unlock"
+    #     =>  DataMessage
     data: Mapped[dict | None] = mapped_column(JSON)
+    # Message
+    #   source_event = "Transfer"
+    #     => None
+    #   source_event = "Unlock"
+    #     => "force_unlock", "garnishment" or "inheritance"
+    message: Mapped[str | None] = mapped_column(String(50), index=True)
 
     @staticmethod
     def format_timestamp(_datetime: datetime) -> str:
@@ -87,6 +107,7 @@ def json(self):
             "value": self.value,
             "source_event": self.source_event,
             "data": self.data,
+            "message": self.message,
             "created": self.format_timestamp(self.created),
         }
 
@@ -99,6 +120,7 @@ def json(self):
         "value": int,
         "source_event": str,
         "data": dict,
+        "message": str,
     }
     FIELDS.update(Base.FIELDS)
 

app/model/schema/token.py

@@ -19,7 +19,7 @@
 
 from datetime import datetime
 from enum import StrEnum
-from typing import Optional
+from typing import Literal, Optional
 
 from pydantic import UUID4, BaseModel, Field, StrictStr
 
@@ -365,7 +365,7 @@ class TokenHoldersCollectionResponse(BaseModel):
     )
 
 
-class TransferHistory(BaseModel):
+class TransferHistoryBase(BaseModel):
     transaction_hash: str = Field(description="Transaction hash")
     token_address: EthereumAddress = Field(description="Token address")
     from_address: EthereumAddress = Field(
@@ -375,16 +375,47 @@ class TransferHistory(BaseModel):
         description="Account address of transfer destination"
     )
     value: int = Field(description="Transfer quantity")
-    source_event: TransferSourceEvent = Field(description="Source Event")
     data: dict | None = Field(description="Event data")
+    message: (
+        Literal[
+            "garnishment",
+            "inheritance",
+            "force_unlock",
+        ]
+        | None
+    )
     created: str = Field(
         description="block_timestamp when Transfer log was emitted (local timezone)"
     )
 
 
+class TransferHistory(TransferHistoryBase):
+    source_event: Literal[TransferSourceEvent.Transfer] = Field(
+        description="Source Event"
+    )
+    data: None = Field(description="Event data")
+
+
+class DataMessage(BaseModel):
+    message: Literal[
+        "garnishment",
+        "inheritance",
+        "force_unlock",
+    ]
+
+
+class UnlockTransferHistory(TransferHistoryBase):
+    source_event: Literal[TransferSourceEvent.Unlock] = Field(
+        description="Source Event"
+    )
+    data: DataMessage | dict = Field(description="Event data")
+
+
 class TransferHistoriesResponse(BaseModel):
     result_set: ResultSet
-    transfer_history: list[TransferHistory] = Field(description="Transfer history")
+    transfer_history: list[TransferHistory | UnlockTransferHistory] = Field(
+        description="Transfer history"
+    )
 
 
 class TransferApprovalHistory(BaseModel):

batch/indexer_Transfer.py

@@ -24,6 +24,7 @@
 from typing import List, Optional, Sequence
 
 from eth_utils import to_checksum_address
+from pydantic import ValidationError
 from sqlalchemy import desc, select
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -34,6 +35,7 @@
 from app.database import BatchAsyncSessionLocal
 from app.errors import ServiceUnavailable
 from app.model.db import (
+    DataMessage,
     IDXTransfer,
     IDXTransferBlockNumber,
     IDXTransferSourceEventType,
@@ -174,10 +176,20 @@ def __insert_idx(
         if data_str is not None:
             try:
                 data = json.loads(data_str)
+                validated_data = DataMessage(**data)
+                message = validated_data.message
+            except ValidationError:
+                data = {}
+                message = None
+            except json.JSONDecodeError:
+                data = {}
+                message = None
             except:
                 data = {}
+                message = None
         else:
             data = None
+            message = None
         transfer = IDXTransfer()
         transfer.transaction_hash = transaction_hash
         transfer.token_address = token_address
@@ -188,6 +200,7 @@ def __insert_idx(
         transfer.modified = event_created
         transfer.source_event = source_event.value
         transfer.data = data
+        transfer.message = message
         db_session.add(transfer)
 
     @staticmethod

migrations/versions/603fd86dc179_v24_12_0_feature_1574.py

@@ -0,0 +1,45 @@
+"""v24_12_0_feature_1574
+
+Revision ID: 603fd86dc179
+Revises: 418af51b07b5
+Create Date: 2024-11-01 11:01:24.485369
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+from app.database import get_db_schema
+
+# revision identifiers, used by Alembic.
+revision = "603fd86dc179"
+down_revision = "418af51b07b5"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    connection = op.get_bind()
+
+    op.add_column(
+        "transfer",
+        sa.Column("message", sa.String(length=50), nullable=True),
+        schema=get_db_schema(),
+    )
+    op.create_index(
+        op.f("ix_transfer_message"),
+        "transfer",
+        ["message"],
+        unique=False,
+        schema=get_db_schema(),
+    )
+
+
+def downgrade():
+    connection = op.get_bind()
+
+    op.drop_index(
+        op.f("ix_transfer_message"), table_name="transfer", schema=get_db_schema()
+    )
+    op.drop_column("transfer", "message", schema=get_db_schema())