Master Algorand smart contract development with comprehensive examples in Python (Puya) and TypeScript (TEALScript). From beginner-friendly contracts to advanced DeFi protocols.
from puyapy import *
class HelloWorld(Contract):
def __init__(self) -> None:
self.greeting = GlobalState(Bytes, key="greeting")
@create
def create(self) -> None:
self.greeting.value = Bytes(b"Hello, Algorand!")
@external
def set_greeting(self, new_greeting: Bytes) -> None:
self.greeting.value = new_greeting
@external
def get_greeting(self) -> Bytes:
return self.greeting.valuefrom puyapy import *
class Calculator(Contract):
def __init__(self) -> None:
self.result = GlobalState(UInt64, key="result")
@create
def create(self) -> None:
self.result.value = UInt64(0)
@external
def add(self, a: UInt64, b: UInt64) -> UInt64:
result = a + b
self.result.value = result
return result
@external
def subtract(self, a: UInt64, b: UInt64) -> UInt64:
assert a >= b, "Cannot subtract larger number from smaller"
result = a - b
self.result.value = result
return result
@external
def multiply(self, a: UInt64, b: UInt64) -> UInt64:
result = a * b
self.result.value = result
return result
@external
def divide(self, a: UInt64, b: UInt64) -> UInt64:
assert b > 0, "Cannot divide by zero"
result = a // b
self.result.value = result
return result
@external
def get_result(self) -> UInt64:
return self.result.valuefrom puyapy import *
class CustomToken(Contract):
def __init__(self) -> None:
self.total_supply = GlobalState(UInt64, key="total_supply")
self.creator = GlobalState(Account, key="creator")
self.balances = BoxMap(Account, UInt64, key_prefix="balance")
@create
def create(self, initial_supply: UInt64) -> None:
self.total_supply.value = initial_supply
self.creator.value = Txn.sender
self.balances[Txn.sender] = initial_supply
@external
def transfer(self, to: Account, amount: UInt64) -> None:
sender_balance = self.balances.get(Txn.sender, UInt64(0))
assert sender_balance >= amount, "Insufficient balance"
self.balances[Txn.sender] = sender_balance - amount
receiver_balance = self.balances.get(to, UInt64(0))
self.balances[to] = receiver_balance + amount
@external
def mint(self, amount: UInt64) -> None:
assert Txn.sender == self.creator.value, "Only creator can mint"
self.total_supply.value += amount
creator_balance = self.balances.get(self.creator.value, UInt64(0))
self.balances[self.creator.value] = creator_balance + amount
@external
def get_balance(self, account: Account) -> UInt64:
return self.balances.get(account, UInt64(0))
@external
def get_total_supply(self) -> UInt64:
return self.total_supply.valuefrom puyapy import *
class VotingSystem(Contract):
def __init__(self) -> None:
self.proposal_count = GlobalState(UInt64, key="proposal_count")
self.voting_end = GlobalState(UInt64, key="voting_end")
self.yes_votes = GlobalState(UInt64, key="yes_votes")
self.no_votes = GlobalState(UInt64, key="no_votes")
self.has_voted = LocalState(UInt64, key="has_voted")
@create
def create(self) -> None:
self.proposal_count.value = UInt64(0)
self.yes_votes.value = UInt64(0)
self.no_votes.value = UInt64(0)
@external
def create_proposal(self, duration: UInt64) -> None:
self.proposal_count.value += UInt64(1)
self.voting_end.value = Global.latest_timestamp + duration
self.yes_votes.value = UInt64(0)
self.no_votes.value = UInt64(0)
@external
def vote(self, choice: bool) -> None:
assert Global.latest_timestamp < self.voting_end.value, "Voting period ended"
assert self.has_voted[Txn.sender] == UInt64(0), "Already voted"
self.has_voted[Txn.sender] = UInt64(1)
if choice:
self.yes_votes.value += UInt64(1)
else:
self.no_votes.value += UInt64(1)
@external
def get_results(self) -> tuple[UInt64, UInt64]:
return self.yes_votes.value, self.no_votes.value
@external
def has_user_voted(self, user: Account) -> bool:
return self.has_voted[user] == UInt64(1)from puyapy import *
class Auction(Contract):
def __init__(self) -> None:
self.seller = GlobalState(Account, key="seller")
self.highest_bidder = GlobalState(Account, key="highest_bidder")
self.highest_bid = GlobalState(UInt64, key="highest_bid")
self.auction_end = GlobalState(UInt64, key="auction_end")
self.ended = GlobalState(bool, key="ended")
self.asset_id = GlobalState(UInt64, key="asset_id")
@create
def create(self, asset_id: UInt64, duration: UInt64, starting_bid: UInt64) -> None:
self.seller.value = Txn.sender
self.asset_id.value = asset_id
self.auction_end.value = Global.latest_timestamp + duration
self.highest_bid.value = starting_bid
self.ended.value = False
@external
def bid(self, payment: PaymentTransaction) -> None:
assert not self.ended.value, "Auction has ended"
assert Global.latest_timestamp < self.auction_end.value, "Auction time expired"
assert payment.amount > self.highest_bid.value, "Bid too low"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
# Refund previous highest bidder
if self.highest_bidder.value != Global.zero_address:
itxn.Payment(
receiver=self.highest_bidder.value,
amount=self.highest_bid.value,
).submit()
self.highest_bidder.value = payment.sender
self.highest_bid.value = payment.amount
@external
def end_auction(self) -> None:
assert Global.latest_timestamp >= self.auction_end.value, "Auction not yet ended"
assert not self.ended.value, "Auction already ended"
self.ended.value = True
if self.highest_bidder.value != Global.zero_address:
# Transfer asset to winner
itxn.AssetTransfer(
asset_receiver=self.highest_bidder.value,
asset_amount=1,
xfer_asset=self.asset_id.value,
).submit()
# Transfer payment to seller
itxn.Payment(
receiver=self.seller.value,
amount=self.highest_bid.value,
).submit()
@external
def get_auction_info(self) -> tuple[Account, UInt64, UInt64, bool]:
return (
self.highest_bidder.value,
self.highest_bid.value,
self.auction_end.value,
self.ended.value,
)from puyapy import *
class EscrowService(Contract):
def __init__(self) -> None:
self.buyer = GlobalState(Account, key="buyer")
self.seller = GlobalState(Account, key="seller")
self.amount = GlobalState(UInt64, key="amount")
self.status = GlobalState(UInt64, key="status") # 0: pending, 1: completed, 2: disputed
@create
def create(self, buyer: Account, seller: Account, amount: UInt64) -> None:
self.buyer.value = buyer
self.seller.value = seller
self.amount.value = amount
self.status.value = UInt64(0)
@external
def release_funds(self) -> None:
assert Txn.sender == self.buyer.value, "Only buyer can release funds"
assert self.status.value == UInt64(0), "Escrow not in pending state"
self.status.value = UInt64(1)
itxn.Payment(
receiver=self.seller.value,
amount=self.amount.value,
).submit()
@external
def refund_buyer(self) -> None:
assert Txn.sender == self.seller.value, "Only seller can initiate refund"
assert self.status.value == UInt64(0), "Escrow not in pending state"
self.status.value = UInt64(1)
itxn.Payment(
receiver=self.buyer.value,
amount=self.amount.value,
).submit()
@external
def dispute(self) -> None:
assert (
Txn.sender == self.buyer.value or Txn.sender == self.seller.value
), "Only buyer or seller can dispute"
self.status.value = UInt64(2)
@external
def get_escrow_info(self) -> tuple[Account, Account, UInt64, UInt64]:
return self.buyer.value, self.seller.value, self.amount.value, self.status.valuefrom puyapy import *
class NFTMarketplace(Contract):
def __init__(self) -> None:
self.nft_count = GlobalState(UInt64, key="nft_count")
self.marketplace_fee = GlobalState(UInt64, key="marketplace_fee")
self.nft_prices = BoxMap(UInt64, UInt64, key_prefix="price")
self.nft_sellers = BoxMap(UInt64, Account, key_prefix="seller")
self.nft_listed = BoxMap(UInt64, bool, key_prefix="listed")
@create
def create(self) -> None:
self.nft_count.value = UInt64(0)
self.marketplace_fee.value = UInt64(250) # 2.5% fee (basis points)
@external
def mint_nft(self, metadata_url: Bytes) -> UInt64:
nft_id = self.nft_count.value + UInt64(1)
self.nft_count.value = nft_id
# Create NFT asset
itxn.AssetConfig(
config_asset_total=1,
config_asset_decimals=0,
config_asset_default_frozen=False,
config_asset_name=Bytes(b"NFT #") + itoa(nft_id),
config_asset_url=metadata_url,
).submit()
return nft_id
@external
def list_nft(self, nft_id: UInt64, price: UInt64) -> None:
assert not self.nft_listed[nft_id], "NFT already listed"
self.nft_prices[nft_id] = price
self.nft_sellers[nft_id] = Txn.sender
self.nft_listed[nft_id] = True
@external
def buy_nft(self, nft_id: UInt64, payment: PaymentTransaction) -> None:
assert self.nft_listed[nft_id], "NFT not listed"
price = self.nft_prices[nft_id]
seller = self.nft_sellers[nft_id]
fee = (price * self.marketplace_fee.value) // UInt64(10000)
assert payment.amount >= price, "Insufficient payment"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
# Transfer payment to seller (minus fee)
itxn.Payment(
receiver=seller,
amount=price - fee,
).submit()
# Transfer NFT to buyer
# Note: In practice, you'd need the actual asset ID from creation
self.nft_listed[nft_id] = False
@external
def unlist_nft(self, nft_id: UInt64) -> None:
assert Txn.sender == self.nft_sellers[nft_id], "Only seller can unlist"
self.nft_listed[nft_id] = False
@external
def get_nft_info(self, nft_id: UInt64) -> tuple[UInt64, Account, bool]:
return (
self.nft_prices.get(nft_id, UInt64(0)),
self.nft_sellers.get(nft_id, Global.zero_address),
self.nft_listed.get(nft_id, False),
)from puyapy import *
class StakingPool(Contract):
def __init__(self) -> None:
self.total_staked = GlobalState(UInt64, key="total_staked")
self.reward_rate = GlobalState(UInt64, key="reward_rate")
self.last_update = GlobalState(UInt64, key="last_update")
self.staked_amount = LocalState(UInt64, key="staked_amount")
self.last_claim = LocalState(UInt64, key="last_claim")
self.pending_rewards = LocalState(UInt64, key="pending_rewards")
@create
def create(self, initial_reward_rate: UInt64) -> None:
self.total_staked.value = UInt64(0)
self.reward_rate.value = initial_reward_rate
self.last_update.value = Global.latest_timestamp
@external
def stake(self, amount: UInt64, payment: PaymentTransaction) -> None:
assert amount > UInt64(0), "Amount must be positive"
assert payment.amount == amount, "Payment amount mismatch"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self._update_rewards()
self.staked_amount[Txn.sender] += amount
self.total_staked.value += amount
self.last_claim[Txn.sender] = Global.latest_timestamp
@external
def claim_rewards(self) -> UInt64:
self._update_rewards()
rewards = self.pending_rewards[Txn.sender]
self.pending_rewards[Txn.sender] = UInt64(0)
self.last_claim[Txn.sender] = Global.latest_timestamp
if rewards > UInt64(0):
itxn.Payment(
receiver=Txn.sender,
amount=rewards,
).submit()
return rewards
@external
def unstake(self, amount: UInt64) -> None:
assert self.staked_amount[Txn.sender] >= amount, "Insufficient staked amount"
# Claim pending rewards first
self.claim_rewards()
self.staked_amount[Txn.sender] -= amount
self.total_staked.value -= amount
# Return staked tokens
itxn.Payment(
receiver=Txn.sender,
amount=amount,
).submit()
@subroutine
def _update_rewards(self) -> None:
time_elapsed = Global.latest_timestamp - self.last_claim[Txn.sender]
staked_amount = self.staked_amount[Txn.sender]
if staked_amount > UInt64(0) and time_elapsed > UInt64(0):
rewards = (staked_amount * self.reward_rate.value * time_elapsed) // (UInt64(86400) * UInt64(10000))
self.pending_rewards[Txn.sender] += rewards
@external
def get_staked_amount(self) -> UInt64:
return self.staked_amount[Txn.sender]
@external
def get_pending_rewards(self) -> UInt64:
return self.pending_rewards[Txn.sender]from puyapy import *
class Lottery(Contract):
def __init__(self) -> None:
self.ticket_price = GlobalState(UInt64, key="ticket_price")
self.total_tickets = GlobalState(UInt64, key="total_tickets")
self.lottery_end = GlobalState(UInt64, key="lottery_end")
self.winner = GlobalState(Account, key="winner")
self.ended = GlobalState(bool, key="ended")
self.tickets = BoxMap(UInt64, Account, key_prefix="ticket")
self.user_tickets = LocalState(UInt64, key="user_tickets")
@create
def create(self, ticket_price: UInt64, duration: UInt64) -> None:
self.ticket_price.value = ticket_price
self.total_tickets.value = UInt64(0)
self.lottery_end.value = Global.latest_timestamp + duration
self.ended.value = False
@external
def buy_ticket(self, payment: PaymentTransaction) -> UInt64:
assert not self.ended.value, "Lottery has ended"
assert Global.latest_timestamp < self.lottery_end.value, "Lottery time expired"
assert payment.amount == self.ticket_price.value, "Incorrect payment amount"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
ticket_id = self.total_tickets.value + UInt64(1)
self.total_tickets.value = ticket_id
self.tickets[ticket_id] = payment.sender
self.user_tickets[payment.sender] += UInt64(1)
return ticket_id
@external
def draw_winner(self) -> Account:
assert Global.latest_timestamp >= self.lottery_end.value, "Lottery not yet ended"
assert not self.ended.value, "Lottery already ended"
assert self.total_tickets.value > UInt64(0), "No tickets sold"
# Simple random number generation (in practice, use VRF)
random_seed = Global.latest_timestamp + Global.round
winning_ticket = (random_seed % self.total_tickets.value) + UInt64(1)
winner = self.tickets[winning_ticket]
self.winner.value = winner
self.ended.value = True
# Transfer prize to winner (90% of total, 10% fee)
total_prize = self.ticket_price.value * self.total_tickets.value
winner_prize = (total_prize * UInt64(90)) // UInt64(100)
itxn.Payment(
receiver=winner,
amount=winner_prize,
).submit()
return winner
@external
def get_lottery_info(self) -> tuple[UInt64, UInt64, UInt64, Account, bool]:
return (
self.ticket_price.value,
self.total_tickets.value,
self.lottery_end.value,
self.winner.value,
self.ended.value,
)
@external
def get_user_tickets(self, user: Account) -> UInt64:
return self.user_tickets[user]from puyapy import *
class TimelockVault(Contract):
def __init__(self) -> None:
self.owner = GlobalState(Account, key="owner")
self.beneficiary = GlobalState(Account, key="beneficiary")
self.release_time = GlobalState(UInt64, key="release_time")
self.amount = GlobalState(UInt64, key="amount")
self.released = GlobalState(bool, key="released")
@create
def create(self, beneficiary: Account, lock_duration: UInt64) -> None:
self.owner.value = Txn.sender
self.beneficiary.value = beneficiary
self.release_time.value = Global.latest_timestamp + lock_duration
self.amount.value = UInt64(0)
self.released.value = False
@external
def deposit(self, payment: PaymentTransaction) -> None:
assert not self.released.value, "Vault already released"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self.amount.value += payment.amount
@external
def release(self) -> None:
assert Global.latest_timestamp >= self.release_time.value, "Timelock not yet expired"
assert not self.released.value, "Already released"
assert self.amount.value > UInt64(0), "No funds to release"
self.released.value = True
itxn.Payment(
receiver=self.beneficiary.value,
amount=self.amount.value,
).submit()
@external
def emergency_withdraw(self) -> None:
assert Txn.sender == self.owner.value, "Only owner can emergency withdraw"
assert not self.released.value, "Already released"
self.released.value = True
itxn.Payment(
receiver=self.owner.value,
amount=self.amount.value,
).submit()
@external
def extend_timelock(self, additional_time: UInt64) -> None:
assert Txn.sender == self.owner.value, "Only owner can extend timelock"
assert not self.released.value, "Already released"
self.release_time.value += additional_time
@external
def get_vault_info(self) -> tuple[Account, Account, UInt64, UInt64, bool]:
return (
self.owner.value,
self.beneficiary.value,
self.release_time.value,
self.amount.value,
self.released.value,
)from puyapy import *
class MultiSigWallet(Contract):
def __init__(self) -> None:
self.owners = BoxMap(Account, bool, key_prefix="owner")
self.required_signatures = GlobalState(UInt64, key="required_sigs")
self.transaction_count = GlobalState(UInt64, key="tx_count")
self.transactions = BoxMap(UInt64, Bytes, key_prefix="tx")
self.confirmations = BoxMap(Bytes, UInt64, key_prefix="confirm") # tx_id -> confirmation count
self.user_confirmations = BoxMap(Bytes, bool, key_prefix="user_confirm") # user+tx_id -> bool
@create
def create(self, owners: Bytes, required_sigs: UInt64) -> None:
# owners is a concatenated list of addresses
self.required_signatures.value = required_sigs
self.transaction_count.value = UInt64(0)
# Parse and add owners (simplified - in practice, you'd parse the bytes)
# For demo, assume first owner is the creator
self.owners[Txn.sender] = True
@external
def add_owner(self, new_owner: Account) -> None:
assert self.owners[Txn.sender], "Only owners can add new owners"
self.owners[new_owner] = True
@external
def submit_transaction(self, to: Account, amount: UInt64, data: Bytes) -> UInt64:
assert self.owners[Txn.sender], "Only owners can submit transactions"
tx_id = self.transaction_count.value + UInt64(1)
self.transaction_count.value = tx_id
# Store transaction data (simplified)
tx_data = concat(to.bytes, itoa(amount), data)
self.transactions[tx_id] = tx_data
self.confirmations[itoa(tx_id)] = UInt64(0)
return tx_id
@external
def confirm_transaction(self, tx_id: UInt64) -> None:
assert self.owners[Txn.sender], "Only owners can confirm transactions"
tx_key = itoa(tx_id)
user_confirm_key = concat(Txn.sender.bytes, tx_key)
assert not self.user_confirmations[user_confirm_key], "Already confirmed"
self.user_confirmations[user_confirm_key] = True
self.confirmations[tx_key] += UInt64(1)
@external
def execute_transaction(self, tx_id: UInt64, to: Account, amount: UInt64) -> None:
tx_key = itoa(tx_id)
confirmations = self.confirmations[tx_key]
assert confirmations >= self.required_signatures.value, "Insufficient confirmations"
assert self.transactions[tx_id] != Bytes(b""), "Transaction does not exist"
# Execute the transaction
itxn.Payment(
receiver=to,
amount=amount,
).submit()
# Clear transaction data
self.transactions[tx_id] = Bytes(b"")
@external
def get_confirmation_count(self, tx_id: UInt64) -> UInt64:
return self.confirmations[itoa(tx_id)]
@external
def is_owner(self, account: Account) -> bool:
return self.owners[account]from puyapy import *
class Crowdfunding(Contract):
def __init__(self) -> None:
self.creator = GlobalState(Account, key="creator")
self.goal = GlobalState(UInt64, key="goal")
self.deadline = GlobalState(UInt64, key="deadline")
self.total_raised = GlobalState(UInt64, key="total_raised")
self.goal_reached = GlobalState(bool, key="goal_reached")
self.funds_withdrawn = GlobalState(bool, key="funds_withdrawn")
self.contributions = BoxMap(Account, UInt64, key_prefix="contrib")
@create
def create(self, goal: UInt64, duration: UInt64) -> None:
self.creator.value = Txn.sender
self.goal.value = goal
self.deadline.value = Global.latest_timestamp + duration
self.total_raised.value = UInt64(0)
self.goal_reached.value = False
self.funds_withdrawn.value = False
@external
def contribute(self, payment: PaymentTransaction) -> None:
assert Global.latest_timestamp < self.deadline.value, "Campaign has ended"
assert not self.goal_reached.value, "Goal already reached"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self.contributions[payment.sender] += payment.amount
self.total_raised.value += payment.amount
if self.total_raised.value >= self.goal.value:
self.goal_reached.value = True
@external
def withdraw_funds(self) -> None:
assert Txn.sender == self.creator.value, "Only creator can withdraw"
assert self.goal_reached.value, "Goal not reached"
assert not self.funds_withdrawn.value, "Funds already withdrawn"
self.funds_withdrawn.value = True
itxn.Payment(
receiver=self.creator.value,
amount=self.total_raised.value,
).submit()
@external
def refund(self) -> None:
assert Global.latest_timestamp >= self.deadline.value, "Campaign still active"
assert not self.goal_reached.value, "Goal was reached"
contribution = self.contributions[Txn.sender]
assert contribution > UInt64(0), "No contribution to refund"
self.contributions[Txn.sender] = UInt64(0)
itxn.Payment(
receiver=Txn.sender,
amount=contribution,
).submit()
@external
def get_campaign_info(self) -> tuple[Account, UInt64, UInt64, UInt64, bool, bool]:
return (
self.creator.value,
self.goal.value,
self.deadline.value,
self.total_raised.value,
self.goal_reached.value,
self.funds_withdrawn.value,
)from puyapy import *
class SubscriptionService(Contract):
def __init__(self) -> None:
self.service_provider = GlobalState(Account, key="provider")
self.subscription_fee = GlobalState(UInt64, key="fee")
self.billing_period = GlobalState(UInt64, key="period")
self.subscriber_count = GlobalState(UInt64, key="sub_count")
self.subscribers = BoxMap(Account, UInt64, key_prefix="subscriber") # subscriber -> expiry
self.last_payment = BoxMap(Account, UInt64, key_prefix="last_pay")
@create
def create(self, fee: UInt64, period: UInt64) -> None:
self.service_provider.value = Txn.sender
self.subscription_fee.value = fee
self.billing_period.value = period
self.subscriber_count.value = UInt64(0)
@external
def subscribe(self, payment: PaymentTransaction) -> None:
assert payment.amount >= self.subscription_fee.value, "Insufficient payment"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
current_expiry = self.subscribers.get(payment.sender, UInt64(0))
if current_expiry == UInt64(0):
self.subscriber_count.value += UInt64(1)
# Extend subscription from current time or existing expiry
start_time = max(Global.latest_timestamp, current_expiry)
new_expiry = start_time + self.billing_period.value
self.subscribers[payment.sender] = new_expiry
self.last_payment[payment.sender] = Global.latest_timestamp
# Transfer payment to service provider
itxn.Payment(
receiver=self.service_provider.value,
amount=payment.amount,
).submit()
@external
def renew_subscription(self, payment: PaymentTransaction) -> None:
assert payment.amount >= self.subscription_fee.value, "Insufficient payment"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
assert self.subscribers[payment.sender] > UInt64(0), "Not a subscriber"
# Extend subscription from current expiry
current_expiry = self.subscribers[payment.sender]
new_expiry = current_expiry + self.billing_period.value
self.subscribers[payment.sender] = new_expiry
self.last_payment[payment.sender] = Global.latest_timestamp
# Transfer payment to service provider
itxn.Payment(
receiver=self.service_provider.value,
amount=payment.amount,
).submit()
@external
def cancel_subscription(self) -> None:
assert self.subscribers[Txn.sender] > UInt64(0), "Not a subscriber"
self.subscribers[Txn.sender] = UInt64(0)
self.subscriber_count.value -= UInt64(1)
@external
def is_active_subscriber(self, user: Account) -> bool:
expiry = self.subscribers.get(user, UInt64(0))
return expiry > Global.latest_timestamp
@external
def get_subscription_info(self, user: Account) -> tuple[UInt64, UInt64, bool]:
expiry = self.subscribers.get(user, UInt64(0))
last_pay = self.last_payment.get(user, UInt64(0))
is_active = expiry > Global.latest_timestamp
return expiry, last_pay, is_active
@external
def get_service_stats(self) -> tuple[UInt64, UInt64, UInt64]:
return (
self.subscription_fee.value,
self.billing_period.value,
self.subscriber_count.value,
)from puyapy import *
class DecentralizedInsurance(Contract):
def __init__(self) -> None:
self.insurer = GlobalState(Account, key="insurer")
self.premium_amount = GlobalState(UInt64, key="premium")
self.coverage_amount = GlobalState(UInt64, key="coverage")
self.policy_duration = GlobalState(UInt64, key="duration")
self.total_pool = GlobalState(UInt64, key="pool")
self.active_policies = GlobalState(UInt64, key="active")
self.policies = BoxMap(Account, UInt64, key_prefix="policy") # policyholder -> expiry
self.claims = BoxMap(Account, bool, key_prefix="claim") # policyholder -> claimed
self.oracle_address = GlobalState(Account, key="oracle")
@create
def create(self, premium: UInt64, coverage: UInt64, duration: UInt64, oracle: Account) -> None:
self.insurer.value = Txn.sender
self.premium_amount.value = premium
self.coverage_amount.value = coverage
self.policy_duration.value = duration
self.oracle_address.value = oracle
self.total_pool.value = UInt64(0)
self.active_policies.value = UInt64(0)
@external
def buy_policy(self, payment: PaymentTransaction) -> None:
assert payment.amount >= self.premium_amount.value, "Insufficient premium"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
assert self.policies[payment.sender] <= Global.latest_timestamp, "Policy already active"
# Set policy expiry
policy_expiry = Global.latest_timestamp + self.policy_duration.value
self.policies[payment.sender] = policy_expiry
self.claims[payment.sender] = False
# Add to insurance pool
self.total_pool.value += payment.amount
self.active_policies.value += UInt64(1)
@external
def file_claim(self) -> None:
assert self.policies[Txn.sender] > Global.latest_timestamp, "No active policy"
assert not self.claims[Txn.sender], "Claim already filed"
self.claims[Txn.sender] = True
@external
def process_claim(self, policyholder: Account, approved: bool) -> None:
assert Txn.sender == self.oracle_address.value, "Only oracle can process claims"
assert self.policies[policyholder] > Global.latest_timestamp, "No active policy"
assert self.claims[policyholder], "No claim filed"
if approved:
assert self.total_pool.value >= self.coverage_amount.value, "Insufficient pool funds"
# Pay claim
itxn.Payment(
receiver=policyholder,
amount=self.coverage_amount.value,
).submit()
self.total_pool.value -= self.coverage_amount.value
# Invalidate policy after claim
self.policies[policyholder] = UInt64(0)
self.active_policies.value -= UInt64(1)
# Reset claim status
self.claims[policyholder] = False
@external
def add_to_pool(self, payment: PaymentTransaction) -> None:
assert Txn.sender == self.insurer.value, "Only insurer can add to pool"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self.total_pool.value += payment.amount
@external
def withdraw_excess(self, amount: UInt64) -> None:
assert Txn.sender == self.insurer.value, "Only insurer can withdraw"
# Ensure minimum pool coverage
required_reserve = self.active_policies.value * self.coverage_amount.value
available = self.total_pool.value - required_reserve
assert amount <= available, "Insufficient excess funds"
self.total_pool.value -= amount
itxn.Payment(
receiver=self.insurer.value,
amount=amount,
).submit()
@external
def get_policy_info(self, user: Account) -> tuple[UInt64, bool, bool]:
expiry = self.policies.get(user, UInt64(0))
is_active = expiry > Global.latest_timestamp
has_claim = self.claims.get(user, False)
return expiry, is_active, has_claim
@external
def get_pool_stats(self) -> tuple[UInt64, UInt64, UInt64, UInt64]:
return (
self.total_pool.value,
self.active_policies.value,
self.premium_amount.value,
self.coverage_amount.value,
)from puyapy import *
class DAOGovernance(Contract):
def __init__(self) -> None:
self.token_address = GlobalState(UInt64, key="token")
self.proposal_count = GlobalState(UInt64, key="prop_count")
self.voting_period = GlobalState(UInt64, key="vote_period")
self.quorum_threshold = GlobalState(UInt64, key="quorum")
self.proposals = BoxMap(UInt64, Bytes, key_prefix="proposal")
self.proposal_votes_for = BoxMap(UInt64, UInt64, key_prefix="votes_for")
self.proposal_votes_against = BoxMap(UInt64, UInt64, key_prefix="votes_against")
self.proposal_end_time = BoxMap(UInt64, UInt64, key_prefix="end_time")
self.proposal_executed = BoxMap(UInt64, bool, key_prefix="executed")
self.user_votes = BoxMap(Bytes, UInt64, key_prefix="user_vote") # user+proposal_id -> vote_power
@create
def create(self, token_id: UInt64, vote_period: UInt64, quorum: UInt64) -> None:
self.token_address.value = token_id
self.proposal_count.value = UInt64(0)
self.voting_period.value = vote_period
self.quorum_threshold.value = quorum
@external
def create_proposal(self, description: Bytes, target: Account, call_data: Bytes) -> UInt64:
# Check if sender has governance tokens (simplified check)
proposal_id = self.proposal_count.value + UInt64(1)
self.proposal_count.value = proposal_id
# Store proposal data
proposal_data = concat(description, target.bytes, call_data)
self.proposals[proposal_id] = proposal_data
self.proposal_end_time[proposal_id] = Global.latest_timestamp + self.voting_period.value
self.proposal_votes_for[proposal_id] = UInt64(0)
self.proposal_votes_against[proposal_id] = UInt64(0)
self.proposal_executed[proposal_id] = False
return proposal_id
@external
def vote(self, proposal_id: UInt64, support: bool, vote_power: UInt64) -> None:
assert proposal_id <= self.proposal_count.value, "Invalid proposal"
assert Global.latest_timestamp < self.proposal_end_time[proposal_id], "Voting period ended"
user_vote_key = concat(Txn.sender.bytes, itoa(proposal_id))
assert self.user_votes[user_vote_key] == UInt64(0), "Already voted"
# In practice, verify vote_power against token balance
self.user_votes[user_vote_key] = vote_power
if support:
self.proposal_votes_for[proposal_id] += vote_power
else:
self.proposal_votes_against[proposal_id] += vote_power
@external
def execute_proposal(self, proposal_id: UInt64) -> None:
assert proposal_id <= self.proposal_count.value, "Invalid proposal"
assert Global.latest_timestamp >= self.proposal_end_time[proposal_id], "Voting still active"
assert not self.proposal_executed[proposal_id], "Already executed"
votes_for = self.proposal_votes_for[proposal_id]
votes_against = self.proposal_votes_against[proposal_id]
total_votes = votes_for + votes_against
assert total_votes >= self.quorum_threshold.value, "Quorum not reached"
assert votes_for > votes_against, "Proposal rejected"
self.proposal_executed[proposal_id] = True
# In practice, execute the proposal action here
# This would involve calling the target contract with call_data
@external
def get_proposal_info(self, proposal_id: UInt64) -> tuple[UInt64, UInt64, UInt64, bool, bool]:
votes_for = self.proposal_votes_for.get(proposal_id, UInt64(0))
votes_against = self.proposal_votes_against.get(proposal_id, UInt64(0))
end_time = self.proposal_end_time.get(proposal_id, UInt64(0))
executed = self.proposal_executed.get(proposal_id, False)
is_active = Global.latest_timestamp < end_time
return votes_for, votes_against, end_time, executed, is_active
@external
def has_voted(self, user: Account, proposal_id: UInt64) -> bool:
user_vote_key = concat(user.bytes, itoa(proposal_id))
return self.user_votes.get(user_vote_key, UInt64(0)) > UInt64(0)
@external
def get_dao_stats(self) -> tuple[UInt64, UInt64, UInt64]:
return (
self.proposal_count.value,
self.voting_period.value,
self.quorum_threshold.value,
)from puyapy import *
class PredictionMarket(Contract):
def __init__(self) -> None:
self.creator = GlobalState(Account, key="creator")
self.oracle = GlobalState(Account, key="oracle")
self.question = GlobalState(Bytes, key="question")
self.resolution_time = GlobalState(UInt64, key="resolution_time")
self.total_yes_bets = GlobalState(UInt64, key="yes_bets")
self.total_no_bets = GlobalState(UInt64, key="no_bets")
self.resolved = GlobalState(bool, key="resolved")
self.outcome = GlobalState(bool, key="outcome")
self.market_fee = GlobalState(UInt64, key="fee") # basis points
self.user_yes_bets = BoxMap(Account, UInt64, key_prefix="yes_bet")
self.user_no_bets = BoxMap(Account, UInt64, key_prefix="no_bet")
self.claimed = BoxMap(Account, bool, key_prefix="claimed")
@create
def create(self, question: Bytes, resolution_time: UInt64, oracle: Account, fee: UInt64) -> None:
self.creator.value = Txn.sender
self.oracle.value = oracle
self.question.value = question
self.resolution_time.value = resolution_time
self.total_yes_bets.value = UInt64(0)
self.total_no_bets.value = UInt64(0)
self.resolved.value = False
self.market_fee.value = fee
@external
def bet_yes(self, payment: PaymentTransaction) -> None:
assert not self.resolved.value, "Market already resolved"
assert Global.latest_timestamp < self.resolution_time.value, "Betting period ended"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self.user_yes_bets[payment.sender] += payment.amount
self.total_yes_bets.value += payment.amount
@external
def bet_no(self, payment: PaymentTransaction) -> None:
assert not self.resolved.value, "Market already resolved"
assert Global.latest_timestamp < self.resolution_time.value, "Betting period ended"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self.user_no_bets[payment.sender] += payment.amount
self.total_no_bets.value += payment.amount
@external
def resolve_market(self, result: bool) -> None:
assert Txn.sender == self.oracle.value, "Only oracle can resolve"
assert Global.latest_timestamp >= self.resolution_time.value, "Resolution time not reached"
assert not self.resolved.value, "Already resolved"
self.resolved.value = True
self.outcome.value = result
@external
def claim_winnings(self) -> None:
assert self.resolved.value, "Market not resolved"
assert not self.claimed[Txn.sender], "Already claimed"
user_winning_bet = UInt64(0)
total_winning_bets = UInt64(0)
total_losing_bets = UInt64(0)
if self.outcome.value: # YES outcome
user_winning_bet = self.user_yes_bets.get(Txn.sender, UInt64(0))
total_winning_bets = self.total_yes_bets.value
total_losing_bets = self.total_no_bets.value
else: # NO outcome
user_winning_bet = self.user_no_bets.get(Txn.sender, UInt64(0))
total_winning_bets = self.total_no_bets.value
total_losing_bets = self.total_yes_bets.value
assert user_winning_bet > UInt64(0), "No winning bet to claim"
# Calculate winnings: original bet + proportional share of losing bets (minus fee)
total_pool = total_winning_bets + total_losing_bets
fee_amount = (total_pool * self.market_fee.value) // UInt64(10000)
net_losing_bets = total_losing_bets - fee_amount
proportional_winnings = (user_winning_bet * net_losing_bets) // total_winning_bets
total_payout = user_winning_bet + proportional_winnings
self.claimed[Txn.sender] = True
itxn.Payment(
receiver=Txn.sender,
amount=total_payout,
).submit()
@external
def withdraw_fees(self) -> None:
assert Txn.sender == self.creator.value, "Only creator can withdraw fees"
assert self.resolved.value, "Market not resolved"
total_pool = self.total_yes_bets.value + self.total_no_bets.value
fee_amount = (total_pool * self.market_fee.value) // UInt64(10000)
itxn.Payment(
receiver=self.creator.value,
amount=fee_amount,
).submit()
@external
def get_market_info(self) -> tuple[UInt64, UInt64, UInt64, bool, bool]:
return (
self.total_yes_bets.value,
self.total_no_bets.value,
self.resolution_time.value,
self.resolved.value,
self.outcome.value,
)
@external
def get_user_bets(self, user: Account) -> tuple[UInt64, UInt64, bool]:
yes_bet = self.user_yes_bets.get(user, UInt64(0))
no_bet = self.user_no_bets.get(user, UInt64(0))
has_claimed = self.claimed.get(user, False)
return yes_bet, no_bet, has_claimedfrom puyapy import *
class SupplyChainTracking(Contract):
def __init__(self) -> None:
self.admin = GlobalState(Account, key="admin")
self.product_count = GlobalState(UInt64, key="product_count")
self.products = BoxMap(UInt64, Bytes, key_prefix="product") # product_id -> product_data
self.product_status = BoxMap(UInt64, UInt64, key_prefix="status") # 0: created, 1: shipped, 2: delivered
self.product_owner = BoxMap(UInt64, Account, key_prefix="owner")
self.product_location = BoxMap(UInt64, Bytes, key_prefix="location")
self.product_timestamp = BoxMap(UInt64, UInt64, key_prefix="timestamp")
self.authorized_parties = BoxMap(Account, bool, key_prefix="authorized")
@create
def create(self) -> None:
self.admin.value = Txn.sender
self.product_count.value = UInt64(0)
self.authorized_parties[Txn.sender] = True
@external
def authorize_party(self, party: Account) -> None:
assert Txn.sender == self.admin.value, "Only admin can authorize parties"
self.authorized_parties[party] = True
@external
def revoke_authorization(self, party: Account) -> None:
assert Txn.sender == self.admin.value, "Only admin can revoke authorization"
self.authorized_parties[party] = False
@external
def create_product(self, product_data: Bytes, initial_location: Bytes) -> UInt64:
assert self.authorized_parties[Txn.sender], "Not authorized"
product_id = self.product_count.value + UInt64(1)
self.product_count.value = product_id
self.products[product_id] = product_data
self.product_status[product_id] = UInt64(0) # Created
self.product_owner[product_id] = Txn.sender
self.product_location[product_id] = initial_location
self.product_timestamp[product_id] = Global.latest_timestamp
return product_id
@external
def update_location(self, product_id: UInt64, new_location: Bytes) -> None:
assert self.authorized_parties[Txn.sender], "Not authorized"
assert product_id <= self.product_count.value, "Invalid product ID"
self.product_location[product_id] = new_location
self.product_timestamp[product_id] = Global.latest_timestamp
@external
def transfer_ownership(self, product_id: UInt64, new_owner: Account) -> None:
assert product_id <= self.product_count.value, "Invalid product ID"
assert Txn.sender == self.product_owner[product_id], "Only current owner can transfer"
assert self.authorized_parties[new_owner], "New owner not authorized"
self.product_owner[product_id] = new_owner
self.product_timestamp[product_id] = Global.latest_timestamp
@external
def update_status(self, product_id: UInt64, new_status: UInt64) -> None:
assert self.authorized_parties[Txn.sender], "Not authorized"
assert product_id <= self.product_count.value, "Invalid product ID"
assert new_status <= UInt64(2), "Invalid status"
current_status = self.product_status[product_id]
assert new_status > current_status, "Cannot downgrade status"
self.product_status[product_id] = new_status
self.product_timestamp[product_id] = Global.latest_timestamp
@external
def get_product_info(self, product_id: UInt64) -> tuple[Bytes, UInt64, Account, Bytes, UInt64]:
assert product_id <= self.product_count.value, "Invalid product ID"
return (
self.products[product_id],
self.product_status[product_id],
self.product_owner[product_id],
self.product_location[product_id],
self.product_timestamp[product_id],
)
@external
def verify_authenticity(self, product_id: UInt64) -> bool:
assert product_id <= self.product_count.value, "Invalid product ID"
return self.products[product_id] != Bytes(b"")
@external
def is_authorized(self, party: Account) -> bool:
return self.authorized_parties.get(party, False)
@external
def get_supply_chain_stats(self) -> tuple[UInt64, Account]:
return self.product_count.value, self.admin.valuefrom puyapy import *
class CarbonCreditsTrading(Contract):
def __init__(self) -> None:
self.admin = GlobalState(Account, key="admin")
self.verifier = GlobalState(Account, key="verifier")
self.credit_count = GlobalState(UInt64, key="credit_count")
self.total_retired = GlobalState(UInt64, key="total_retired")
self.credits = BoxMap(UInt64, Bytes, key_prefix="credit") # credit_id -> project_data
self.credit_owner = BoxMap(UInt64, Account, key_prefix="owner")
self.credit_amount = BoxMap(UInt64, UInt64, key_prefix="amount") # tons of CO2
self.credit_price = BoxMap(UInt64, UInt64, key_prefix="price")
self.credit_verified = BoxMap(UInt64, bool, key_prefix="verified")
self.credit_retired = BoxMap(UInt64, bool, key_prefix="retired")
self.credit_for_sale = BoxMap(UInt64, bool, key_prefix="for_sale")
@create
def create(self, verifier: Account) -> None:
self.admin.value = Txn.sender
self.verifier.value = verifier
self.credit_count.value = UInt64(0)
self.total_retired.value = UInt64(0)
@external
def issue_credits(self, project_data: Bytes, amount: UInt64, recipient: Account) -> UInt64:
assert Txn.sender == self.admin.value, "Only admin can issue credits"
credit_id = self.credit_count.value + UInt64(1)
self.credit_count.value = credit_id
self.credits[credit_id] = project_data
self.credit_owner[credit_id] = recipient
self.credit_amount[credit_id] = amount
self.credit_verified[credit_id] = False
self.credit_retired[credit_id] = False
self.credit_for_sale[credit_id] = False
self.credit_price[credit_id] = UInt64(0)
return credit_id
@external
def verify_credits(self, credit_id: UInt64) -> None:
assert Txn.sender == self.verifier.value, "Only verifier can verify credits"
assert credit_id <= self.credit_count.value, "Invalid credit ID"
assert not self.credit_retired[credit_id], "Cannot verify retired credits"
self.credit_verified[credit_id] = True
@external
def list_for_sale(self, credit_id: UInt64, price: UInt64) -> None:
assert credit_id <= self.credit_count.value, "Invalid credit ID"
assert Txn.sender == self.credit_owner[credit_id], "Only owner can list for sale"
assert self.credit_verified[credit_id], "Credits must be verified"
assert not self.credit_retired[credit_id], "Cannot sell retired credits"
self.credit_for_sale[credit_id] = True
self.credit_price[credit_id] = price
@external
def remove_from_sale(self, credit_id: UInt64) -> None:
assert credit_id <= self.credit_count.value, "Invalid credit ID"
assert Txn.sender == self.credit_owner[credit_id], "Only owner can remove from sale"
self.credit_for_sale[credit_id] = False
self.credit_price[credit_id] = UInt64(0)
@external
def buy_credits(self, credit_id: UInt64, payment: PaymentTransaction) -> None:
assert credit_id <= self.credit_count.value, "Invalid credit ID"
assert self.credit_for_sale[credit_id], "Credits not for sale"
assert not self.credit_retired[credit_id], "Cannot buy retired credits"
assert payment.amount >= self.credit_price[credit_id], "Insufficient payment"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
seller = self.credit_owner[credit_id]
# Transfer payment to seller
itxn.Payment(
receiver=seller,
amount=payment.amount,
).submit()
# Transfer ownership
self.credit_owner[credit_id] = payment.sender
self.credit_for_sale[credit_id] = False
self.credit_price[credit_id] = UInt64(0)
@external
def retire_credits(self, credit_id: UInt64) -> None:
assert credit_id <= self.credit_count.value, "Invalid credit ID"
assert Txn.sender == self.credit_owner[credit_id], "Only owner can retire credits"
assert self.credit_verified[credit_id], "Credits must be verified"
assert not self.credit_retired[credit_id], "Credits already retired"
self.credit_retired[credit_id] = True
self.credit_for_sale[credit_id] = False
self.total_retired.value += self.credit_amount[credit_id]
@external
def get_credit_info(self, credit_id: UInt64) -> tuple[Bytes, Account, UInt64, UInt64, bool, bool, bool]:
assert credit_id <= self.credit_count.value, "Invalid credit ID"
return (
self.credits[credit_id],
self.credit_owner[credit_id],
self.credit_amount[credit_id],
self.credit_price[credit_id],
self.credit_verified[credit_id],
self.credit_retired[credit_id],
self.credit_for_sale[credit_id],
)
@external
def get_market_stats(self) -> tuple[UInt64, UInt64]:
return self.credit_count.value, self.total_retired.valuefrom puyapy import *
class RealEstateTokenization(Contract):
def __init__(self) -> None:
self.property_owner = GlobalState(Account, key="owner")
self.property_value = GlobalState(UInt64, key="value")
self.total_tokens = GlobalState(UInt64, key="total_tokens")
self.tokens_sold = GlobalState(UInt64, key="tokens_sold")
self.token_price = GlobalState(UInt64, key="token_price")
self.rental_pool = GlobalState(UInt64, key="rental_pool")
self.last_distribution = GlobalState(UInt64, key="last_distribution")
self.property_data = GlobalState(Bytes, key="property_data")
self.token_balances = BoxMap(Account, UInt64, key_prefix="balance")
self.claimed_dividends = BoxMap(Account, UInt64, key_prefix="claimed")
@create
def create(self, property_val: UInt64, total_supply: UInt64, price_per_token: UInt64, prop_data: Bytes) -> None:
self.property_owner.value = Txn.sender
self.property_value.value = property_val
self.total_tokens.value = total_supply
self.token_price.value = price_per_token
self.property_data.value = prop_data
self.tokens_sold.value = UInt64(0)
self.rental_pool.value = UInt64(0)
self.last_distribution.value = Global.latest_timestamp
@external
def buy_tokens(self, quantity: UInt64, payment: PaymentTransaction) -> None:
assert quantity > UInt64(0), "Quantity must be positive"
assert self.tokens_sold.value + quantity <= self.total_tokens.value, "Not enough tokens available"
total_cost = quantity * self.token_price.value
assert payment.amount >= total_cost, "Insufficient payment"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
# Transfer payment to property owner
itxn.Payment(
receiver=self.property_owner.value,
amount=payment.amount,
).submit()
# Update token balances
self.token_balances[payment.sender] += quantity
self.tokens_sold.value += quantity
@external
def transfer_tokens(self, to: Account, quantity: UInt64) -> None:
assert quantity > UInt64(0), "Quantity must be positive"
assert self.token_balances[Txn.sender] >= quantity, "Insufficient token balance"
self.token_balances[Txn.sender] -= quantity
self.token_balances[to] += quantity
@external
def add_rental_income(self, payment: PaymentTransaction) -> None:
assert Txn.sender == self.property_owner.value, "Only property owner can add rental income"
assert payment.receiver == Global.current_application_address, "Payment must go to contract"
self.rental_pool.value += payment.amount
@external
def distribute_dividends(self) -> None:
assert Txn.sender == self.property_owner.value, "Only property owner can distribute dividends"
assert self.rental_pool.value > UInt64(0), "No rental income to distribute"
self.last_distribution.value = Global.latest_timestamp
# Note: In practice, this would trigger individual dividend calculations
@external
def claim_dividends(self) -> UInt64:
user_tokens = self.token_balances[Txn.sender]
assert user_tokens > UInt64(0), "No tokens owned"
# Calculate user's share of rental income
user_share = (self.rental_pool.value * user_tokens) // self.tokens_sold.value
already_claimed = self.claimed_dividends.get(Txn.sender, UInt64(0))
claimable = user_share - already_claimed
assert claimable > UInt64(0), "No dividends to claim"
self.claimed_dividends[Txn.sender] = user_share
itxn.Payment(
receiver=Txn.sender,
amount=claimable,
).submit()
return claimable
@external
def update_property_value(self, new_value: UInt64) -> None:
assert Txn.sender == self.property_owner.value, "Only property owner can update value"
self.property_value.value = new_value
@external
def get_token_info(self, user: Account) -> tuple[UInt64, UInt64, UInt64]:
user_balance = self.token_balances.get(user, UInt64(0))
user_claimed = self.claimed_dividends.get(user, UInt64(0))
# Calculate pending dividends
user_share = UInt64(0)
if self.tokens_sold.value > UInt64(0):
user_share = (self.rental_pool.value * user_balance) // self.tokens_sold.value
pending_dividends = user_share - user_claimed
return user_balance, user_claimed, pending_dividends
@external
def get_property_info(self) -> tuple[UInt64, UInt64, UInt64, UInt64, UInt64, Bytes]:
return (
self.property_value.value,
self.total_tokens.value,
self.tokens_sold.value,
self.token_price.value,
self.rental_pool.value,
self.property_data.value,
)from puyapy import *
from algopy import ARC4Contract, GlobalState, UInt64
from algopy.arc4 import abimethod
class CustomCreate(ARC4Contract):
def __init__(self) -> None:
self.age = GlobalState(UInt64)
@abimethod(create="require")
def custom_create(self, age: UInt64) -> None:
self.age.value = age
@abimethod()
def get_age(self) -> UInt64:
return self.age.valuefrom puyapy import *
from algopy import BoxMap, arc4
class UserStruct(arc4.Struct):
name: arc4.String
id: arc4.UInt64
asset: arc4.UInt64
class StructInBoxMap(arc4.ARC4Contract):
def __init__(self) -> None:
self.user_map = BoxMap(arc4.UInt64, UserStruct, key_prefix="users")
@arc4.abimethod
def box_map_test(self) -> bool:
key_0 = arc4.UInt64(0)
value = UserStruct(arc4.String("testName"), arc4.UInt64(70), arc4.UInt64(2))
self.user_map[key_0] = value.copy()
assert self.user_map[key_0].bytes.length == value.bytes.length
assert self.user_map.length(key_0) == value.bytes.length
return True
@arc4.abimethod
def box_map_set(self, key: arc4.UInt64, value: UserStruct) -> bool:
self.user_map[key] = value.copy()
assert self.user_map[key] == value
return True
@arc4.abimethod
def box_map_get(self, key: arc4.UInt64) -> UserStruct:
return self.user_map[key]
@arc4.abimethod
def box_map_exists(self, key: arc4.UInt64) -> bool:
return key in self.user_mapfrom puyapy import *
from algopy import (
Bytes,
Global,
TemplateVar,
TransactionType,
Txn,
UInt64,
logicsig,
op,
)
@logicsig
def self_payment() -> bool:
"""
This Delegated Account will authorize a single empty self payment in a block known ahead of time.
"""
return (
Txn.type_enum == TransactionType.Payment
and Txn.receiver == Txn.sender
and Txn.amount == 0
and Txn.rekey_to == Global.zero_address
and Txn.close_remainder_to == Global.zero_address
and Txn.fee == Global.min_txn_fee
and Global.genesis_hash == TemplateVar[Bytes]("TARGET_NETWORK_GENESIS")
# Acquiring a lease with last_round and a non-empty lease field prevents replay attacks.
and Txn.last_valid == TemplateVar[UInt64]("LAST_ROUND")
and Txn.lease == op.sha256(b"self-payment")
)