anchorpy
anchorpy copied to clipboard
anchorpy parser problem
context: processing transaction logs into events for archival purposes with anchorpy
code:
import asyncio
import base64
import binascii
from solana.rpc.async_api import AsyncClient # type: ignore
from solders.signature import Signature # type: ignore
from anchorpy import Wallet, Program, Provider # type: ignore
DRIFT_PROGRAM_ID: str = "dRiftyHA39MWEi3m9aunc5MzRF1JYuBsbn6VPcn33UH"
PROGRAM_LOG: str = "Program log: "
PROGRAM_DATA: str = "Program data: "
PROGRAM_LOG_START_INDEX: int = len(PROGRAM_LOG)
PROGRAM_DATA_START_INDEX: int = len(PROGRAM_DATA)
async def parse_logs(program: Program, logs: list[str]) -> None:
for log in logs:
if log.startswith("Log truncated"):
break
handle_log(log, program)
def handle_log(
log: str, program: Program
):
if log.startswith(PROGRAM_LOG) or log.startswith(PROGRAM_DATA):
log_str = (
log[PROGRAM_LOG_START_INDEX:]
if log.startswith(PROGRAM_LOG)
else log[PROGRAM_DATA_START_INDEX:]
)
try:
decoded = base64.b64decode(log_str)
except binascii.Error:
return
# throws here
event = program.coder.events.parse(decoded)
print(event)
return
else:
return
async def main():
connection = AsyncClient("https://api.mainnet-beta.solana.com")
signature = Signature.from_string("gjVEiXqtWPCbfKtsypL3xvNk8ef2j6D1bo5pHjvL5Bms66TbDozjs7jqusfWj2GhPgDmTojPYgYXtr7AuTk5Y3c")
wallet = Wallet.dummy()
provider = Provider(connection, wallet)
program = await Program.at(DRIFT_PROGRAM_ID, provider)
logs = (await connection.get_transaction(signature, max_supported_transaction_version=0)).value.transaction.meta.log_messages
await parse_logs(program, logs)
if __name__ == "__main__":
asyncio.run(main())
error:
raise StreamError("stream read less than specified amount, expected %d, found %d" % (length, len(data)), path=path)
construct.core.StreamError: Error in path (parsing) -> discriminator
stream read less than specified amount, expected 8, found 6