stocra.models

  1from dataclasses import dataclass
  2from decimal import Decimal
  3from enum import Enum, unique
  4from typing import Any, Awaitable, Callable, Dict, List, Optional, Union
  5
  6from pydantic import BaseModel, root_validator, validator
  7
  8Address = str
  9TransactionHash = str
 10OutputIndex = int
 11
 12
 13class Amount(BaseModel):
 14    value: Decimal
 15    currency_symbol: str
 16
 17    def __add__(self, other: Any) -> "Amount":
 18        if isinstance(other, Amount):
 19            if self.currency_symbol != other.currency_symbol:
 20                raise ValueError(
 21                    f"Amounts have different currencies: {self.currency_symbol} != {other.currency_symbol}"
 22                )
 23
 24            return Amount(value=self.value + other.value, currency_symbol=self.currency_symbol)
 25
 26        raise TypeError(f"Cannot add Amount and {type(other)}")
 27
 28    def __eq__(self, other: Any) -> bool:
 29        if not isinstance(other, Amount):
 30            raise TypeError(f"Cannot compare Amount and {type(other)}")
 31
 32        if self.value != other.value:
 33            return False
 34
 35        if self.currency_symbol != other.currency_symbol:
 36            return False
 37
 38        return True
 39
 40
 41class TransactionPointer(BaseModel):
 42    transaction_hash: TransactionHash
 43    output_index: OutputIndex
 44
 45
 46class Input(BaseModel):
 47    address: Optional[Address] = None
 48    amount: Optional[Amount] = None
 49    transaction_pointer: Optional[TransactionPointer] = None
 50
 51    @classmethod
 52    @root_validator
 53    def validate_address_or_pointer(cls, values: Dict) -> Dict:  # pylint: disable=no-self-argument
 54        if not (values["address"] or values["transaction_pointer"]):
 55            raise ValueError("Either address or transaction pointer must be specified")
 56
 57        return values
 58
 59
 60class Output(BaseModel):
 61    address: Address
 62    amount: Amount
 63
 64
 65class Transaction(BaseModel):
 66    hash: TransactionHash
 67    inputs: List[Input]
 68    outputs: List[Output]
 69    fee: Amount
 70
 71    @classmethod
 72    @validator("fee")
 73    def validate_fee(cls, value: Amount) -> Amount:
 74        if value.value < 0:
 75            raise ValueError(f"Fee cannot be negative! Got {value.value}")
 76
 77        return value
 78
 79
 80class Block(BaseModel):
 81    height: int
 82    hash: str
 83    timestamp_ms: int
 84    transactions: List[str] = []
 85
 86    @classmethod
 87    @validator("timestamp_ms")
 88    def validate_timestamp_ms(cls, value: int) -> int:  # pylint: disable=no-self-argument
 89        # I don't really care if this runs even after 2286
 90        if len(str(value)) != 13:
 91            raise ValueError("Timestamp must be in miliseconds")
 92
 93        return value
 94
 95
 96class Currency(BaseModel):
 97    symbol: str
 98    name: str
 99
100    class Config:
101        frozen = True
102
103
104@unique
105class TokenType(Enum):
106    ERC20 = "ERC20"
107
108
109class Token(BaseModel):
110    currency: Currency
111    scaling: Decimal
112    type: TokenType
113
114
115@dataclass(frozen=True)
116class StocraHTTPError:
117    endpoint: str
118    iteration: int
119    exception: Exception
120
121
122ErrorHandler = Callable[[StocraHTTPError], Union[bool, Awaitable[bool]]]