如何在borsh-python的Schema中指定Pubkey作为预期类型？

尝试用borsh-python库反序列化指定十六进制字符串时，在Schema中直接使用solders.pubkey.Pubkey类型触发错误：TypeError: value '<class 'solders.pubkey.Pubkey'>' is not a valid Borsh type，需要解决类型替代方案或自定义Pubkey的Borsh类型，以得到预期的序列化输出。

当前触发错误的代码：

import borsh
from borsh import types
from solders.pubkey import Pubkey

# Define the schema for the dictionary
example_dict_schema = borsh.schema({
    'idk': types.u64,
    'mint': Pubkey,
    'solAmount': types.u64,
    'tokenAmount': types.u64,
    'isBuy': types.u8,
    'user': Pubkey,
    'timestamp': types.i64,
    'virtualSolReserves': types.u64,
    'virtualTokenReserves': types.u64
})

serialized_bytes = bytes.fromhex('e445a52e51cb9a1dbddb7fd34ee661eecf5078301148ff160d7d39c2eb39bd8a65336260014e92375510898fccc8b08f00d2496b000000005b49b52a3d3700000118fd4318e5c5d0ebafd594a987aa6bd95ced29e4b785c4ef9de66783fc08cdc54a1a656600000000007e6d6707000000a5c6221da698030000d2496b00000000a52e10d1149a0200')

deserialized_data = borsh.deserialize(example_dict_schema, serialized_bytes)

print(deserialized_data)

预期输出：

{
  "mint": {
    "type": "publicKey",
    "data": "ExGZYPhHkhQ4jCbYPicCZz89tEmtVKiS6j9MhSqypump"
  },
  "solAmount": {
    "type": "u64",
    "data": "1800000000"
  },
  "tokenAmount": {
    "type": "u64",
    "data": "60735849056603"
  },
  "isBuy": {
    "type": "bool",
    "data": true
  },
  "user": {
    "type": "publicKey",
    "data": "2gYmikFDQno2o7shxA6CgDF5AVM6Px6DpHRNc6RNMiDS"
  },
  "timestamp": {
    "type": "i64",
    "data": "1717901898"
  },
  "virtualSolReserves": {
    "type": "u64",
    "data": "31800000000"
  },
  "virtualTokenReserves": {
    "type": "u64",
    "data": "1012264150943397"
  }
}

报错原因是borsh-python原生不支持solders.pubkey.Pubkey类型，而Pubkey本质是32字节的固定长度二进制数据，可以通过以下两种方式解决：

方法一：用固定长度数组替代，手动转换 #

直接在Schema中用types.array(types.u8, 32)表示32字节数组，反序列化后再将字节数组转换为Pubkey对象，最后整理成预期格式：

import borsh
from borsh import types
from solders.pubkey import Pubkey
import json

# 修改Schema，用32字节数组替代Pubkey类型
example_dict_schema = borsh.schema({
    'idk': types.u64,
    'mint': types.array(types.u8, 32),
    'solAmount': types.u64,
    'tokenAmount': types.u64,
    'isBuy': types.u8,
    'user': types.array(types.u8, 32),
    'timestamp': types.i64,
    'virtualSolReserves': types.u64,
    'virtualTokenReserves': types.u64
})

serialized_bytes = bytes.fromhex('e445a52e51cb9a1dbddb7fd34ee661eecf5078301148ff160d7d39c2eb39bd8a65336260014e92375510898fccc8b08f00d2496b000000005b49b52a3d3700000118fd4318e5c5d0ebafd594a987aa6bd95ced29e4b785c4ef9de66783fc08cdc54a1a656600000000007e6d6707000000a5c6221da698030000d2496b00000000a52e10d1149a0200')

# 反序列化原始数据
deserialized_raw = borsh.deserialize(example_dict_schema, serialized_bytes)

# 转换为预期的输出格式
formatted_output = {
    "mint": {
        "type": "publicKey",
        "data": str(Pubkey(deserialized_raw['mint']))
    },
    "solAmount": {
        "type": "u64",
        "data": str(deserialized_raw['solAmount'])
    },
    "tokenAmount": {
        "type": "u64",
        "data": str(deserialized_raw['tokenAmount'])
    },
    "isBuy": {
        "type": "bool",
        "data": bool(deserialized_raw['isBuy'])
    },
    "user": {
        "type": "publicKey",
        "data": str(Pubkey(deserialized_raw['user']))
    },
    "timestamp": {
        "type": "i64",
        "data": str(deserialized_raw['timestamp'])
    },
    "virtualSolReserves": {
        "type": "u64",
        "data": str(deserialized_raw['virtualSolReserves'])
    },
    "virtualTokenReserves": {
        "type": "u64",
        "data": str(deserialized_raw['virtualTokenReserves'])
    }
}

print(json.dumps(formatted_output, indent=2))

方法二：自定义Pubkey的Borsh类型 #

通过继承borsh的types.Type类，实现Pubkey的序列化/反序列化逻辑，之后可以在Schema中直接使用自定义类型，更适合重复使用的场景：

import borsh
from borsh import types
from solders.pubkey import Pubkey
import json

# 自定义支持Pubkey的Borsh类型
class PubkeyType(types.Type):
    @classmethod
    def deserialize(cls, data, offset=0):
        # 读取32字节数据并转换为Pubkey
        pubkey_bytes = data[offset:offset+32]
        return Pubkey(pubkey_bytes), offset + 32
    
    @classmethod
    def serialize(cls, value):
        # 序列化时输出Pubkey的字节数组
        return bytes(value)
    
    @classmethod
    def get_size(cls, value=None):
        # Pubkey固定为32字节
        return 32

# 使用自定义类型定义Schema
example_dict_schema = borsh.schema({
    'idk': types.u64,
    'mint': PubkeyType,
    'solAmount': types.u64,
    'tokenAmount': types.u64,
    'isBuy': types.u8,
    'user': PubkeyType,
    'timestamp': types.i64,
    'virtualSolReserves': types.u64,
    'virtualTokenReserves': types.u64
})

serialized_bytes = bytes.fromhex('e445a52e51cb9a1dbddb7fd34ee661eecf5078301148ff160d7d39c2eb39bd8a65336260014e92375510898fccc8b08f00d2496b000000005b49b52a3d3700000118fd4318e5c5d0ebafd594a987aa6bd95ced29e4b785c4ef9de66783fc08cdc54a1a656600000000007e6d6707000000a5c6221da698030000d2496b00000000a52e10d1149a0200')

# 反序列化原始数据
deserialized_raw = borsh.deserialize(example_dict_schema, serialized_bytes)

# 转换为预期的输出格式
formatted_output = {
    "mint": {
        "type": "publicKey",
        "data": str(deserialized_raw['mint'])
    },
    "solAmount": {
        "type": "u64",
        "data": str(deserialized_raw['solAmount'])
    },
    "tokenAmount": {
        "type": "u64",
        "data": str(deserialized_raw['tokenAmount'])
    },
    "isBuy": {
        "type": "bool",
        "data": bool(deserialized_raw['isBuy'])
    },
    "user": {
        "type": "publicKey",
        "data": str(deserialized_raw['user'])
    },
    "timestamp": {
        "type": "i64",
        "data": str(deserialized_raw['timestamp'])
    },
    "virtualSolReserves": {
        "type": "u64",
        "data": str(deserialized_raw['virtualSolReserves'])
    },
    "virtualTokenReserves": {
        "type": "u64",
        "data": str(deserialized_raw['virtualTokenReserves'])
    }
}

print(json.dumps(formatted_output, indent=2))

内容的提问来源于stack exchange，提问作者Shane Marotta