综合项目与最佳实践

10 天学习路径

SDK 模块总结

项目概述

构建一个完整的去中心化投票 DApp，整合所有学习内容。

智能合约

@compiler >= 6

contract VotingSystem =
    
    record proposal = {
        id          : int,
        title       : string,
        description : string,
        creator     : address,
        yes_votes   : int,
        no_votes    : int,
        end_height  : int,
        executed    : bool
    }
    
    record state = {
        owner           : address,
        proposals       : map(int, proposal),
        proposal_count  : int,
        votes           : map(address, map(int, bool)),
        min_vote_period : int
    }
    
    datatype event =
        ProposalCreated(int, string, address)
        | Voted(int, address, bool)
        | ProposalExecuted(int)
    
    entrypoint init() = {
        owner = Call.caller,
        proposals = {},
        proposal_count = 0,
        votes = {},
        min_vote_period = 100
    }
    
    // 创建提案
    stateful entrypoint create_proposal(title: string, description: string, 
                                         vote_period: int) : int =
        require(vote_period >= state.min_vote_period, "Vote period too short")
        
        let id = state.proposal_count + 1
        let new_proposal = {
            id = id,
            title = title,
            description = description,
            creator = Call.caller,
            yes_votes = 0,
            no_votes = 0,
            end_height = Chain.block_height + vote_period,
            executed = false
        }
        
        put(state{
            proposals = state.proposals{[id] = new_proposal},
            proposal_count = id
        })
        
        Chain.event(ProposalCreated(id, title, Call.caller))
        id
    
    // 投票
    stateful entrypoint vote(proposal_id: int, support: bool) =
        require(Map.member(proposal_id, state.proposals), "Proposal not found")
        
        let p = state.proposals[proposal_id]
        require(Chain.block_height < p.end_height, "Voting ended")
        require(!has_voted(Call.caller, proposal_id), "Already voted")
        
        let updated = if(support)
            p{yes_votes = p.yes_votes + 1}
        else
            p{no_votes = p.no_votes + 1}
        
        put(state{
            proposals = state.proposals{[proposal_id] = updated},
            votes = state.votes{[Call.caller] = 
                Map.lookup_default(Call.caller, state.votes, {}){[proposal_id] = true}}
        })
        
        Chain.event(Voted(proposal_id, Call.caller, support))
    
    // 查询提案
    entrypoint get_proposal(id: int) : proposal =
        require(Map.member(id, state.proposals), "Proposal not found")
        state.proposals[id]
    
    // 检查是否已投票
    entrypoint has_voted(voter: address, proposal_id: int) : bool =
        switch(Map.lookup(voter, state.votes))
            None => false
            Some(votes) => Map.lookup_default(proposal_id, votes, false)
    
    // 获取投票结果
    entrypoint get_results(proposal_id: int) : (int, int) =
        let p = state.proposals[proposal_id]
        (p.yes_votes, p.no_votes)

Python 客户端

from aeternity.node import NodeClient, Config
from aeternity.signing import Account
from aeternity.contract_native import ContractManager

class VotingClient:
    """投票系统客户端"""
    
    def __init__(self, node_url: str, account: Account):
        config = Config(external_url=node_url)
        self.client = NodeClient(config)
        self.account = account
        self.manager = None
    
    def deploy(self, source: str) -> str:
        """部署合约"""
        self.manager = ContractManager(
            client=self.client,
            account=self.account,
            source=source
        )
        tx = self.manager.deploy()
        return self.manager.contract_id
    
    def connect(self, contract_id: str, source: str):
        """连接已部署合约"""
        self.manager = ContractManager(
            client=self.client,
            account=self.account,
            contract_id=contract_id,
            source=source
        )
    
    def create_proposal(self, title: str, description: str, 
                        vote_period: int = 200) -> int:
        """创建提案"""
        result = self.manager.call(
            "create_proposal",
            f'"{title}"',
            f'"{description}"',
            str(vote_period)
        )
        return result.return_value
    
    def vote(self, proposal_id: int, support: bool):
        """投票"""
        self.manager.call(
            "vote",
            str(proposal_id),
            "true" if support else "false"
        )
    
    def get_proposal(self, proposal_id: int) -> dict:
        """获取提案信息"""
        result = self.manager.call_static("get_proposal", str(proposal_id))
        return result
    
    def get_results(self, proposal_id: int) -> tuple:
        """获取投票结果"""
        result = self.manager.call_static("get_results", str(proposal_id))
        return result

# 使用示例
client = VotingClient(
    node_url='https://testnet.aeternity.io',
    account=Account.generate()
)

# 部署合约
contract_id = client.deploy(voting_source)
print(f"合约地址: {contract_id}")

# 创建提案
proposal_id = client.create_proposal(
    title="增加社区基金",
    description="提议将 10% 的收益分配给社区发展基金",
    vote_period=500
)
print(f"提案 ID: {proposal_id}")

# 投票
client.vote(proposal_id, support=True)

# 查询结果
yes, no = client.get_results(proposal_id)
print(f"赞成: {yes}, 反对: {no}")

错误处理

from aeternity.exceptions import (
    AEException,
    TransactionError,
    ContractError,
    InsufficientFundsError
)
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def safe_transaction(func):
    """交易安全装饰器"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except InsufficientFundsError as e:
            logger.error(f"余额不足: {e}")
            raise
        except TransactionError as e:
            logger.error(f"交易失败: {e}")
            raise
        except ContractError as e:
            logger.error(f"合约错误: {e}")
            raise
        except AEException as e:
            logger.error(f"SDK错误: {e}")
            raise
    return wrapper

@safe_transaction
def transfer_with_retry(client, account, to, amount, max_retries=3):
    """带重试的转账"""
    for attempt in range(max_retries):
        try:
            tx = client.spend(account, to, amount)
            logger.info(f"转账成功: {tx.hash}")
            return tx
        except TransactionError as e:
            if attempt < max_retries - 1:
                logger.warning(f"重试 {attempt + 1}/{max_retries}")
                time.sleep(2)
            else:
                raise

配置管理

import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class AppConfig:
    """应用配置"""
    node_url: str
    compiler_url: str
    network_id: str
    debug: bool = False
    
    @classmethod
    def from_env(cls, env: str = "testnet") -> "AppConfig":
        """从环境变量加载配置"""
        configs = {
            "mainnet": cls(
                node_url="https://mainnet.aeternity.io",
                compiler_url="https://compiler.aepps.com",
                network_id="ae_mainnet"
            ),
            "testnet": cls(
                node_url="https://testnet.aeternity.io",
                compiler_url="https://compiler.aepps.com",
                network_id="ae_uat",
                debug=True
            )
        }
        return configs.get(env, configs["testnet"])

# 使用
config = AppConfig.from_env(os.getenv("NETWORK", "testnet"))

安全考量

代码安全示例

# 使用环境变量存储敏感信息
import os
from getpass import getpass

def load_account_secure():
    """安全加载账户"""
    keystore_path = os.getenv("KEYSTORE_PATH")
    
    if keystore_path:
        password = getpass("输入钱包密码: ")
        return Account.from_keystore(keystore_path, password)
    
    secret_key = os.getenv("SECRET_KEY")
    if secret_key:
        return Account.from_secret_key_string(secret_key)
    
    raise ValueError("未配置账户信息")

# 等待交易确认
def wait_for_confirmation(client, tx_hash, confirmations=3, timeout=120):
    """等待交易确认"""
    start = time.time()
    
    while time.time() - start < timeout:
        try:
            tx_info = client.get_transaction_info_by_hash(tx_hash)
            if tx_info.block_height:
                current = client.get_current_key_block_height()
                if current - tx_info.block_height >= confirmations:
                    return tx_info
        except:
            pass
        time.sleep(3)
    
    raise TimeoutError(f"交易确认超时: {tx_hash}")