Merge pull request #1 from antazoey/feat/hexstr

feat: HexStr, Bip122Uri, and HashStr types

.github/workflows/docs.yaml

@@ -11,6 +11,8 @@ on:
 jobs:
     docs:
         runs-on: ubuntu-latest
+        permissions:
+          contents: write
 
         steps:
         - uses: actions/checkout@v3

.github/workflows/draft.yaml

@@ -8,6 +8,8 @@ on:
 jobs:
   update-draft:
     runs-on: ubuntu-latest
+    permissions:
+        contents: write
     steps:
       # Drafts your next Release notes as Pull Requests are merged into "main"
       - uses: release-drafter/release-drafter@v5

README.md

@@ -2,25 +2,28 @@
 
 The types in this package are pydantic types for Ethereum inspired from [eth-typing](https://github.com/ethereum/eth-typing/blob/master/eth_typing/evm.py).
 
-## Hash32
+## Hash
 
-Hash32 is a good type to use for Ethereum transaction hashes.
-When using the Hash32 type, your inputs will be validated by length and your schema will declare the input a string with a binary format.
-Use Hash32 like this:
+`HashBytes{n}` and `HashStr{n}` are good types to use when your hex values are sized.
+Both types serialize to `string` in the JSON schema.
+Use `HashBytes` types when you want types to serialize to bytes in the Pydantic core schema and `HashStr` types when you want to serialize to `str` in the core Pydantic schema.
 
 ```python
 from pydantic import BaseModel
 
-from eth_pydantic_types import Hash32
+from eth_pydantic_types import HashBytes32, HashStr20
 
+# When serializing to JSON, both types are hex strings.
 class Transaction(BaseModel):
-    tx_hash: Hash32
+    tx_hash: HashBytes32  # Will be bytes
+    address: HashStr20  # Will be str
 
 
 # NOTE: I am able to pass an int-hash as the value and it will
 #  get validated and type-coerced.
 tx = Transaction(
-    tx_hash=0x1031f0c9ac54dcb64b4f121a27957c14263c5cb49ed316d568e41e19c34d7b28
+    tx_hash=0x1031f0c9ac54dcb64b4f121a27957c14263c5cb49ed316d568e41e19c34d7b28,
+    address=0x1031f0c9ac54dcb64b4f121a27957c14263c5cb4,
 )
 ```
 
@@ -29,6 +32,7 @@ tx = Transaction(
 A thin-wrapper around an already thin-wrapper `hexbytes.HexBytes`.
 The difference here is that this HexBytes properly serializes.
 Use HexBytes any place where you would actually use `hexbytes.HexBytes`.
+`HexBytes` serializes to bytes in the Pydantic core schema and `string` in the JSON schema with a binary format.
 
 ```python
 from pydantic import BaseModel
@@ -45,6 +49,7 @@ storage = MyStorage(cid="0x123")
 
 Use the Address class for working with checksummed-addresses.
 Addresses get validated and checksummed in model construction.
+Addresses serialize to `str` in the Pydantic core schema and `string` in the JSON schema with a binary format.
 
 ```python
 from pydantic import BaseModel
@@ -57,3 +62,38 @@ class Account(BaseModel):
 #   ("0x0837207e343277CBd6c114a45EC0e9Ec56a1AD84")
 account = Account(address="0x837207e343277cbd6c114a45ec0e9ec56a1ad84")
 ```
+
+## HexStr
+
+Use hex str when you only care about un-sized hex strings.
+The `HexStr` type serializes to `str` in the Pydantic core schema and a `string` in the JSON schema with a binary format.
+
+```python
+from eth_pydantic_types import HexStr
+from pydantic import BaseModel
+
+class Tx(BaseModel):
+    data: HexStr
+
+tx = Tx(data="0x0123")
+```
+
+## Bip122Uri
+
+Use BIP-122 URIs in your models by annotating with the `Bip122Uri` type.
+This type serializes to a `str` in the Pydantic core schema as well as a `string` in the JSON schema, however the individual hashes are validated.
+
+```python
+from eth_pydantic_types import Bip122Uri
+from pydantic import BaseModel
+
+class Message(BaseModel):
+    path: Bip122Uri
+
+message = Message(
+    path=(
+        "blockchain://d4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
+        "/block/752820c0ad7abc1200f9ad42c4adc6fbb4bd44b5bed4667990e64565102c1ba6"
+    )
+)
+```

eth_pydantic_types/__init__.py

@@ -1,5 +1,36 @@
 from .address import Address
-from .hash import Hash4, Hash8, Hash16, Hash20, Hash32, Hash64
-from .hexbytes import HexBytes
+from .bip122 import Bip122Uri
+from .hash import (
+    HashBytes4,
+    HashBytes8,
+    HashBytes16,
+    HashBytes20,
+    HashBytes32,
+    HashBytes64,
+    HashStr4,
+    HashStr8,
+    HashStr16,
+    HashStr20,
+    HashStr32,
+    HashStr64,
+)
+from .hex import HexBytes, HexStr
 
-__all__ = ["Address", "Hash4", "Hash8", "Hash16", "Hash20", "Hash32", "Hash64", "HexBytes"]
+__all__ = [
+    "Address",
+    "Bip122Uri",
+    "HashBytes4",
+    "HashBytes8",
+    "HashBytes16",
+    "HashBytes20",
+    "HashBytes32",
+    "HashBytes64",
+    "HashStr4",
+    "HashStr8",
+    "HashStr16",
+    "HashStr20",
+    "HashStr32",
+    "HashStr64",
+    "HexBytes",
+    "HexStr",
+]

eth_pydantic_types/_error.py

@@ -0,0 +1,27 @@
+from typing import Any, Callable
+
+from pydantic_core import PydanticCustomError
+
+# NOTE: We use the factory approach because PydanticCustomError is a final class.
+#   That is also why this module is internal.
+
+
+def CustomError(fn: Callable, invalid_tag: str, **kwargs) -> PydanticCustomError:
+    return PydanticCustomError(fn.__name__, f"Invalid {invalid_tag}", kwargs)
+
+
+def HexValueError(value: Any) -> PydanticCustomError:
+    return CustomError(HexValueError, "hex value", value=value)
+
+
+def SizeError(size: Any, value: Any) -> PydanticCustomError:
+    return CustomError(SizeError, "size of value", size=size, value=value)
+
+
+def Bip122UriFormatError(value: str) -> PydanticCustomError:
+    return CustomError(
+        Bip122UriFormatError,
+        "BIP-122 URI format",
+        uri=value,
+        format="blockchain://<genesis_hash>/block/<block_hash>.",
+    )

eth_pydantic_types/address.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional
+from typing import Any, ClassVar, Optional, Tuple
 
 from eth_utils import is_checksum_address, to_checksum_address
 from pydantic_core import CoreSchema
@@ -8,20 +8,34 @@
     with_info_before_validator_function,
 )
 
-from eth_pydantic_types.hexbytes import HexBytes
+from eth_pydantic_types.hex import BaseHexStr, HexBytes
 from eth_pydantic_types.validators import validate_address_size
 
+ADDRESS_PATTERN = "^0x[a-fA-F0-9]{40}$"
 
-class Address(str):
+
+def address_schema():
+    return str_schema(min_length=42, max_length=42, pattern=ADDRESS_PATTERN)
+
+
+class Address(BaseHexStr):
     """
     Use for address-types. Validates as a checksummed address. Left-pads zeroes
     if necessary.
     """
 
+    schema_pattern: ClassVar[str] = ADDRESS_PATTERN
+    schema_examples: ClassVar[Tuple[str, ...]] = (
+        "0x0000000000000000000000000000000000000000",  # Zero address
+        "0x02c84e944F97F4A4f60221e6fb5d5DbAE49c7aaB",  # Leading zero
+        "0xa5a13f62ce1113838e0d9b4559b8caf5f76463c0",  # Trailing zero
+        "0x1e59ce931B4CFea3fe4B875411e280e173cB7A9C",
+    )
+
     def __get_pydantic_core_schema__(self, *args, **kwargs) -> CoreSchema:
         schema = with_info_before_validator_function(
             self._validate_address,
-            str_schema(min_length=42, max_length=42, pattern="^0x[a-fA-F0-9]{40}$"),
+            address_schema(),
         )
         return schema
 
@@ -34,12 +48,6 @@ def _validate_address(cls, value: Any, info: Optional[ValidationInfo] = None) ->
             value = HexBytes(value).hex()
 
         number = value[2:] if value.startswith("0x") else value
-        number_padded = validate_address_size(number, 40)
+        number_padded = validate_address_size(number)
         value = f"0x{number_padded}"
         return to_checksum_address(value)
-
-    def __int__(self) -> int:
-        return int(self, 16)
-
-    def __bytes__(self) -> HexBytes:
-        return HexBytes(self)

eth_pydantic_types/bip122.py

@@ -0,0 +1,83 @@
+from enum import Enum
+from functools import cached_property
+from typing import Any, Optional, Tuple
+
+from pydantic_core import CoreSchema
+from pydantic_core.core_schema import (
+    ValidationInfo,
+    str_schema,
+    with_info_before_validator_function,
+)
+
+from eth_pydantic_types._error import Bip122UriFormatError
+from eth_pydantic_types.hex import validate_hex_str
+
+
+class Bip122UriType(Enum):
+    TX = "tx"
+    BLOCK = "block"
+    ADDRESS = "address"
+
+
+class Bip122Uri(str):
+    prefix: str = "blockchain://"
+
+    @classmethod
+    def __get_pydantic_json_schema__(cls, core_schema, handler):
+        json_schema = handler(core_schema)
+        example = (
+            f"{cls.prefix}d4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
+            f"/{Bip122UriType.BLOCK.value}/"
+            f"752820c0ad7abc1200f9ad42c4adc6fbb4bd44b5bed4667990e64565102c1ba6"
+        )
+        pattern = f"^{cls.prefix}[0-9a-f]{{64}}/{Bip122UriType.BLOCK.value}/[0-9a-f]{{64}}$"
+        json_schema.update(examples=[example], pattern=pattern)
+        return json_schema
+
+    def __get_pydantic_core_schema__(self, *args, **kwargs) -> CoreSchema:
+        return with_info_before_validator_function(
+            self._validate,
+            str_schema(),
+        )
+
+    @classmethod
+    def _validate(cls, value: Any, info: Optional[ValidationInfo] = None) -> str:
+        if not value.startswith(cls.prefix):
+            raise Bip122UriFormatError(value)
+
+        genesis_hash, block_keyword, block_hash = cls.parse(value)
+        return f"{cls.prefix}{genesis_hash[2:]}/{block_keyword.value}/{block_hash[2:]}"
+
+    @classmethod
+    def parse(cls, value: str) -> Tuple[str, Bip122UriType, str]:
+        protocol_suffix = value.replace(cls.prefix, "")
+        protocol_parsed = protocol_suffix.split("/")
+        if len(protocol_parsed) != 3:
+            raise Bip122UriFormatError(value)
+
+        genesis_hash, block_keyword, block_hash = protocol_parsed
+        block_keyword = block_keyword.lower()
+        if block_keyword not in [x.value for x in Bip122UriType]:
+            raise Bip122UriFormatError(value)
+
+        return (
+            validate_hex_str(genesis_hash),
+            Bip122UriType(block_keyword),
+            validate_hex_str(block_hash),
+        )
+
+    @cached_property
+    def parsed(self) -> Tuple[str, Bip122UriType, str]:
+        return self.parse(self)
+
+    @property
+    def chain(self) -> str:
+        return self.parsed[0]
+
+    @property
+    def uri_type(self) -> Bip122UriType:
+        return self.parsed[1]
+
+    @property
+    def hash(self) -> str:
+        return self.parsed[2]