Fix addresses parsing

Author: kuyugama

alembic/versions/d2035abc33cf_remove_tx_addresses_index.py

@@ -0,0 +1,30 @@
+"""Remove tx addresses index
+
+Revision ID: d2035abc33cf
+Revises: f14d9f9ad604
+Create Date: 2025-08-04 11:13:34.498234
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'd2035abc33cf'
+down_revision: Union[str, None] = 'f14d9f9ad604'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_service_transactions_addresses'), table_name='service_transactions')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_index(op.f('ix_service_transactions_addresses'), 'service_transactions', ['addresses'], unique=False)
+    # ### end Alembic commands ###

app/address/router.py

@@ -1,10 +1,9 @@
 from app.utils import pagination, paginated_response
 from sqlalchemy.ext.asyncio import AsyncSession
+from fastapi import APIRouter, Depends
 from app.dependencies import get_page
 from app.database import get_session
-from fastapi.params import Depends
 from app.address import service
-from fastapi import APIRouter
 
 from app.schemas import (
     TransactionPaginatedResponse,
@@ -16,9 +15,7 @@
 router = APIRouter(prefix="/address", tags=["Addresses"])
 
 
-@router.get(
-    "/{address}/outputs/{currency}", response_model=OutputPaginatedResponse
-)
+@router.get("/{address}/outputs/{currency}", response_model=OutputPaginatedResponse)
 async def get_unspent_outputs(
     address: str,
     currency: str,
@@ -36,8 +33,26 @@ async def get_unspent_outputs(
 
 
 @router.get(
-    "/{address}/transactions", response_model=TransactionPaginatedResponse
+    "/{address}/utxo/{currency}",
+    summary="Get utxo for specified amount",
+    response_model=OutputPaginatedResponse,
 )
+async def get_utxo(
+    address: str,
+    currency: str,
+    amount: float,
+    session: AsyncSession = Depends(get_session),
+    page: int = Depends(get_page),
+):
+    limit, offset = pagination(page)
+
+    total = await service.count_utxo(session, address, currency, amount)
+    items = await service.list_utxo(session, address, currency, amount, limit, offset)
+
+    return paginated_response(items.all(), total, page, limit)
+
+
+@router.get("/{address}/transactions", response_model=TransactionPaginatedResponse)
 async def get_transactions(
     address: str,
     session: AsyncSession = Depends(get_session),

app/address/service.py

@@ -23,9 +23,7 @@ async def count_unspent_outputs(
     session: AsyncSession, address: str, currency: str
 ) -> int:
     return await session.scalar(
-        unspent_outputs_filters(
-            select(func.count(Output.id)), address, currency
-        )
+        unspent_outputs_filters(select(func.count(Output.id)), address, currency)
     )
 
 
@@ -38,16 +36,55 @@ async def list_unspent_outputs(
 ) -> ScalarResult[Output]:
     return await session.scalars(
         unspent_outputs_filters(
-            select(Output)
-            .order_by(Output.amount.desc())
-            .limit(limit)
-            .offset(offset),
+            select(Output).order_by(Output.amount.desc()).limit(limit).offset(offset),
             address,
             currency,
         )
     )
 
 
+def utxo_cte(address: str, currency: str):
+    return (
+        select(
+            Output,
+            func.sum(Output.amount)  # type: ignore
+            .over(order_by=Output.blockhash)
+            .label("cumulative_amount"),
+        )
+        .filter(Output.address == address, Output.currency == currency, ~Output.spent)
+        .cte("cumulative_outputs")
+    )
+
+
+async def count_utxo(
+    session: AsyncSession, address: str, currency: str, amount: float
+) -> int:
+    cte = utxo_cte(address, currency)
+    query = (
+        select(func.count(1)).select_from(cte).where(cte.c.cumulative_amount < amount)
+    )
+    return await session.scalar(query) or 0
+
+
+async def list_utxo(
+    session: AsyncSession,
+    address: str,
+    currency: str,
+    amount: float,
+    limit: int,
+    offset: int,
+):
+    cte = utxo_cte(address, currency)
+    query = (
+        select(cte)
+        .select_from(cte)
+        .where(cte.c.cumulative_amount < amount)
+        .limit(limit)
+        .offset(offset)
+    )
+    return await session.execute(query)
+
+
 def transactions_filters(query: Select, address: str) -> Select:
     query = query.filter(Transaction.addresses.contains([address]))
 
@@ -74,16 +111,12 @@ async def list_transactions(
             address,
         )
     ):
-        transactions.append(
-            await load_tx_details(session, transaction, latest_block)
-        )
+        transactions.append(await load_tx_details(session, transaction, latest_block))
 
     return transactions
 
 
-async def list_balances(
-    session: AsyncSession, address: str
-) -> list[AddressBalance]:
+async def list_balances(session: AsyncSession, address: str) -> list[AddressBalance]:
     balances = []
     for balance in await session.scalars(
         select(AddressBalance).filter(
@@ -97,18 +130,14 @@ async def list_balances(
     return balances
 
 
-async def list_address_mempool_transactions(
-    session: AsyncSession, address: str
-):
+async def list_address_mempool_transactions(session: AsyncSession, address: str):
     mempool = await session.scalar(select(MemPool).limit(1))
 
     if mempool is None:
         return []
 
     return [
-        await load_mempool_tx_details(
-            session, transaction, mempool.raw["outputs"]
-        )
+        await load_mempool_tx_details(session, transaction, mempool.raw["outputs"])
         for transaction in mempool.raw["transactions"]
         if address in transaction["addresses"]
     ]

app/models/transaction.py

@@ -13,7 +13,7 @@ class Transaction(Base):
     currencies: Mapped[list[str]] = mapped_column(ARRAY(String(64)), index=True)
     txid: Mapped[str] = mapped_column(String(64), index=True, unique=True)
     blockhash: Mapped[str] = mapped_column(String(64), index=True)
-    addresses: Mapped[list[str]] = mapped_column(ARRAY(String), index=True)
+    addresses: Mapped[list[str]] = mapped_column(ARRAY(String))
     created: Mapped[datetime]
     timestamp: Mapped[int]
     size: Mapped[int]

app/parser.py

@@ -164,9 +164,9 @@ async def parse_transactions(txids: list[str]):
 
         addresses = list(
             set(
-                address
+                vout["scriptPubKey"]["address"]
                 for vout in transaction_data["vout"]
-                for address in vout["scriptPubKey"].get("addresses", [])
+                if vout["scriptPubKey"]["type"] != "nulldata"
             )
         )
         timestamp = transaction_data.get("time", None)

scripts/add_addresses_to_txs.py

@@ -0,0 +1,85 @@
+import asyncio
+import typing
+
+from sqlalchemy import func, select, update
+
+from app.database import sessionmanager
+from app.settings import get_settings
+from app.parser import make_request
+from app.models import Transaction
+
+settings: typing.Any = get_settings()
+
+
+async def main():
+    sessionmanager.init(settings.database.endpoint)
+
+    async with sessionmanager.session() as session:
+
+        total = (
+            await session.scalar(
+                select(func.count(Transaction.id)).filter(Transaction.addresses == ())
+            )
+            or 0
+        )
+        limit = 100
+
+        offset = 0
+        query = (
+            select(Transaction.id, Transaction.txid)
+            .limit(limit)
+            .filter(Transaction.addresses == ())
+            .limit(limit)
+        )
+
+        total /= limit
+
+        while txs := (await session.execute(query.offset(offset))).all():
+            offset += limit
+
+            transactions_result = await make_request(
+                settings.blockchain.endpoint,
+                [
+                    {
+                        "id": str(dbid),
+                        "method": "getrawtransaction",
+                        "params": [txid, True],
+                    }
+                    for dbid, txid in txs
+                ],
+            )
+
+            addresses: dict[str, list[str]] = {
+                tx["id"]: (
+                    list(
+                        set(
+                            vout["scriptPubKey"]["address"]
+                            for vout in tx["result"]["vout"]
+                            if vout["scriptPubKey"]["type"] != "nulldata"
+                        )
+                    )
+                )
+                for tx in transactions_result
+            }
+
+            await session.execute(
+                update(Transaction),
+                [
+                    {"id": dbid, "addresses": addresses}
+                    for dbid, addresses in addresses.items()
+                ],
+                execution_options={"synchronize_session": False},
+            )
+            await session.commit()
+            print(
+                f"Progress: {offset // limit}/{total} ({(offset // limit/total)*100:.2f})",
+                end="\r",
+                flush=True,
+            )
+            break
+
+        print("\nDone")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())