PyBook 09: Embedding

from pathlib import Path


def repo_root() -> Path:
    """The checkout this book lives in, wherever it is executed from."""
    here = Path.cwd().resolve()
    for candidate in (here, *here.parents):
        if (candidate / "tests" / "cassettes").is_dir():
            return candidate
    raise RuntimeError("run this book from inside a checkout of auradefi")


CASSETTES = repo_root() / "tests" / "cassettes"
print("cassettes:", CASSETTES.relative_to(repo_root()))

cassettes: tests/cassettes

from sqlalchemy import create_engine
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, select

from auradefi.ledger.backends.models import LedgerTransactionRow, metadata
from auradefi.ledger.backends.sqlmodel import SqlModelLedger

engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)


def session_factory() -> Session:
    "The HOST's session factory: the library only ever calls this."
    return Session(engine)


metadata.create_all(engine)          # the host's DDL, on the host's schedule
ledger = SqlModelLedger(session_factory=session_factory)

assert LedgerTransactionRow.__tablename__ == "auradefi_ledger_transactions"
print("tables the host created:", sorted(metadata.tables))

tables the host created: ['auradefi_ledger_seqs', 'auradefi_ledger_transactions']

import httpx

from auradefi.chains.evm import chain_id_from_caip2
from auradefi.errors import SourceError, ValidationError
from auradefi.sources.evm.etherscan import EtherscanV2
from auradefi.testing.cassettes import load

BASE_URL = "https://api.etherscan.io/v2/api"
PAGE_SIZE = 2


class CountingTransport(httpx.BaseTransport):
    "Counts every request that leaves the client, then replays it."

    def __init__(self, inner):
        self._inner, self.calls = inner, 0

    def handle_request(self, request):
        self.calls += 1
        return self._inner.handle_request(request)


class HostSource:
    "The host's adapter: balances via EtherscanV2, raw txlist pages by hand."

    def __init__(self, client):
        self._client = client
        self._balances = EtherscanV2(client, api_key=None, page_size=PAGE_SIZE)

    def balances(self, chain_id, address):
        return self._balances.balances(chain_id, address)

    def fetch_txlist(self, chain_id, address, *, start_block, end_block, page, offset, sort):
        response = self._client.get(BASE_URL, params={
            "chainid": str(chain_id_from_caip2(chain_id)), "module": "account",
            "action": "txlist", "address": address, "startblock": str(start_block),
            "endblock": str(end_block), "page": str(page), "offset": str(offset), "sort": sort,
        })
        envelope = response.json()
        if envelope.get("status") == "0" and envelope.get("message") == "No transactions found":
            return []
        if envelope.get("status") != "1":
            raise SourceError(f"etherscan txlist error: {envelope.get('message')!r}")
        return list(envelope["result"])


transport = CountingTransport(load(CASSETTES / "embed_gate.json").transport())
client = httpx.Client(transport=transport)
source = HostSource(client)
print("seams:", [name for name in ("balances", "fetch_txlist") if hasattr(source, name)])

seams: ['balances', 'fetch_txlist']

from auradefi import Auradefi
from auradefi.clock import FrozenClock
from auradefi.embed.state import MemorySyncState
from auradefi.config import Settings
from auradefi.prices.inquirer import Inquirer
from auradefi.prices.oracles.defillama import DefiLlamaOracle

T0, INTERVAL_MS = 1_754_000_000_000, 60_000
clock = FrozenClock(T0)
state = MemorySyncState()   # the HOST's cursor store; it outlives the facade

auradefi = Auradefi(
    ledger,
    source,
    Inquirer([DefiLlamaOracle(client)]),
    clock,
    Settings(sync_min_interval_s=60),
    sync_state=state,
    sync_page_size=PAGE_SIZE,
)
assert transport.calls == 0, "binding is ZERO I/O"


class HalfASource:
    def balances(self, chain_id, address):
        return []


try:
    Auradefi(ledger, HalfASource(), Inquirer([]), clock)
except ValidationError as exc:
    print("refused at bind time:", exc)
print("bound with", transport.calls, "requests made")

refused at bind time: source must satisfy PageFetcher: no 'fetch_txlist'
bound with 0 requests made

from auradefi.embed.models import derive_connection_id
from auradefi.errors import CaipParseError, ConflictError, UnknownChainError

CHAIN, ADDRESS = "eip155:1", "0x1111111111111111111111111111111111111111"
user = auradefi.user("host-user-1")
assert user.tenant_id == "usr_1e63721d071ea2d9"   # sha256("embed|host-user-1")[:16]

before = transport.calls
connection = user.connect_address(CHAIN, ADDRESS)
assert transport.calls == before + 1              # exactly one probe
assert connection.id == "conn_d0327e21d9b0ea55"
assert connection.created_at_ms == T0

# The id is CHAIN-SCOPED: one address, two chains, two ids, so two
# independent sync cursors. Pure derivation, no I/O, nothing stored.
assert derive_connection_id(user.tenant_id, ADDRESS, "eip155:137") != connection.id

# A bad chain never reaches the network...
calls = transport.calls
try:
    user.connect_address("ethereum", ADDRESS)
except CaipParseError as exc:
    print("refused pre-flight:", exc)
# ...nor does a well-formed chain the registry does not hold: accepting it
# would store a connection every later sync() dies on.
try:
    user.connect_address("eip155:42161", ADDRESS)
except UnknownChainError as exc:
    print("refused, not seeded:", exc)
# ...and neither does a duplicate: the id is DERIVED from the chain and the
# normalised (lowercased) address, so the clash is known before any request.
try:
    user.connect_address(CHAIN, ADDRESS)
except ConflictError as exc:
    print("duplicate:", exc, "| existing_id:", exc.existing_id)
assert transport.calls == calls, "rejections cost zero requests"
print(connection)

refused pre-flight: not a canonical eip155 CAIP-2: 'ethereum'
refused, not seeded: unknown chain 'eip155:42161': CAIP-2 is the only key
duplicate: connection already exists: 'conn_d0327e21d9b0ea55' | existing_id: conn_d0327e21d9b0ea55
ConnectionRecord(id='conn_d0327e21d9b0ea55', chain_id='eip155:1', address='0x1111111111111111111111111111111111111111', created_at_ms=1754000000000)

before = transport.calls
first = auradefi.sync(budget=2)

assert first.no_op is False
assert (first.pages_fetched, first.live_pages, first.backfill_pages) == (2, 1, 1)
# 3, not 4: the first backfill page deliberately OVERLAPS the anchor's
# lowest block, because the anchor page may have cut that block in half
# and nothing can know whether it did. One redelivered row buys never
# losing the remainder of a split block; it adds no event.
assert first.transactions_ingested == 3
assert transport.calls == before + 2               # budget respected exactly
row = first.connections[0]
assert (row.live_cursor, row.backfill_cursor, row.backfill_complete) == (106, 104, False)
print(first)

SyncReport(no_op=False, pages_fetched=2, live_pages=1, backfill_pages=1, transactions_ingested=3, connections=(ConnectionSyncReport(connection_id='conn_d0327e21d9b0ea55', no_op=False, pages_fetched=2, live_pages=1, backfill_pages=1, transactions_ingested=3, live_cursor=106, backfill_cursor=104, backfill_complete=False, failed=False),))

calls = transport.calls
noop = auradefi.sync(budget=2)

assert noop.no_op is True
assert noop.pages_fetched == 0 and noop.transactions_ingested == 0
assert transport.calls == calls          # not one byte left the process
print(f"no-op sync: {noop.pages_fetched} pages, transport calls still {transport.calls}")

clock.advance(INTERVAL_MS)               # the host's clock, the host's tick
# budget 4: confirming a window DRAINED costs one page. The six rows of
# [0, 105] fill pages 1-3 exactly, so the only evidence nothing remains
# is a fourth, short page.
resumed = auradefi.sync(budget=4)
assert (resumed.pages_fetched, resumed.live_pages, resumed.backfill_pages) == (4, 1, 3)
assert resumed.connections[0].backfill_complete is True
assert resumed.connections[0].backfill_cursor == 100
print("after advancing the clock:", resumed.pages_fetched, "pages, history complete:",
      resumed.connections[0].backfill_complete)

no-op sync: 0 pages, transport calls still 3
after advancing the clock: 4 pages, history complete: True

restarted = Auradefi(
    ledger,
    source,
    Inquirer([DefiLlamaOracle(client)]),
    clock,
    Settings(sync_min_interval_s=60),
    sync_state=state,                 # the ONLY thing carried across the "restart"
    sync_page_size=PAGE_SIZE,
)
assert state.tenants() == (user.tenant_id,)

calls = transport.calls
after_restart = restarted.sync(budget=2)
assert transport.calls == calls, "still inside the throttle window: zero requests"

# The proof is the breakdown, not the aggregate: the restarted facade NAMED
# the stored connection. A no_op with no rows at all is the old defect.
assert [row.connection_id for row in after_restart.connections] == [connection.id]
assert after_restart.failed_connections == ()
print("restarted facade saw", len(after_restart.connections), "stored connection(s):",
      after_restart.connections[0].connection_id)


class HalfAState:
    "A 0.1.0-shaped store: four methods, no tenants(): refused at bind."

    def get_state(self, tenant_id, connection_id): ...
    def put_state(self, tenant_id, connection_id, state): ...
    def add_connection(self, tenant_id, connection): ...
    def connections(self, tenant_id): return ()


try:
    Auradefi(ledger, source, Inquirer([]), clock, sync_state=HalfAState())
except ValidationError as exc:
    print("stale port refused at bind:", exc)

restarted facade saw 1 stored connection(s): conn_d0327e21d9b0ea55
stale port refused at bind: sync_state has no 'tenants': see SyncStatePort

with session_factory() as session:
    rows = list(session.exec(
        select(LedgerTransactionRow)
        .where(LedgerTransactionRow.tenant_id == user.tenant_id)
        .order_by(LedgerTransactionRow.block_number)
    ))

assert len(rows) == 7
assert [row.block_number for row in rows] == [100, 101, 102, 103, 104, 105, 106]
assert {row.account_id for row in rows} == {connection.id}
assert not any(row.removed for row in rows)
assert rows[0].id == "txn_b3618169bbd2dd6b"
assert rows[0].initiated_at == 1_700_000_000_000
for row in rows[:3]:
    print(f"{row.id}  block={row.block_number}  entries={row.entries_json}")
print("...", len(rows), "rows in the host's sqlite")

txn_b3618169bbd2dd6b  block=100  entries=[{"asset_id":"eip155:1/slip44:60","decimals":18,"direction":"in","raw":"1000000000000000000"}]
txn_66000dfdcbfd1e1a  block=101  entries=[{"asset_id":"eip155:1/slip44:60","decimals":18,"direction":"in","raw":"1000000000000000000"}]
txn_ae616c0eebd4f70c  block=102  entries=[{"asset_id":"eip155:1/slip44:60","decimals":18,"direction":"in","raw":"1000000000000000000"}]
... 7 rows in the host's sqlite

from decimal import Decimal

from auradefi.money.fiat import Money

(report,) = auradefi.holdings()
assert report.total_value == Money(Decimal("5025"), "USD")     # 2 ETH @ 2500 + 25 USDC @ 1
assert [holding.symbol for holding in report.holdings] == ["ETH", "USDC"]
assert report.unpriced == ()
print(report.address, report.total_value)

metrics = auradefi.scalar_metrics()
values = {metric.name: metric.value for metric in metrics}
assert len(metrics) == 26
assert values["portfolio_value_usd"] == 5025.0
assert values["transaction_count"] == 7.0
assert sum(values[f"tx_count_hour_{hour:02d}"] for hour in range(24)) == 7.0
active = {hour for hour in range(24) if values[f"tx_count_hour_{hour:02d}"]}
assert active == {22, 23, 0, 1, 2, 3, 4}
print("metrics:", len(metrics), "| active UTC hours:", sorted(active))

0x1111111111111111111111111111111111111111 5025.000000000000000000 USD
metrics: 26 | active UTC hours: [0, 1, 2, 3, 4, 22, 23]