Reference

algobase.models.asset_params

A Pydantic model for Algorand Standard Asset parameters.

AssetParams

Bases: BaseModel

A Pydantic model for Algorand Standard Assets (ASAs).

Source code in algobase/models/asset_params.py

class AssetParams(BaseModel):
    """A Pydantic model for Algorand Standard Assets (ASAs)."""

    model_config = ConfigDict(frozen=True)

    total: Uint64 = Field(
        description="The total number of base units of the asset to create."
    )
    decimals: AsaDecimals = Field(
        default=0,
        description="The number of digits to use after the decimal point when displaying the asset. If 0, the asset is not divisible. If 1, the base unit of the asset is in tenths. Must be between 0 and 19, inclusive.",
    )
    default_frozen: bool = Field(
        default=False,
        description="Whether slots for this asset in user accounts are frozen by default.",
    )
    unit_name: AsaUnitName | None = Field(
        default=None,
        description="The name of a unit of this asset. Max size is 8 bytes. Example: 'USDT'.",
    )
    asset_name: AsaAssetName | None = Field(
        default=None,
        description="The name of the asset. Max size is 32 bytes. Example: 'Tether'.",
    )
    url: AsaUrl | None = Field(
        default=None,
        description="Specifies a URL where more information about the asset can be retrieved. Max size is 96 bytes.",
    )
    metadata_hash: AlgorandHash | None = Field(
        default=None,
        description="This field is intended to be a 32-byte hash of some metadata that is relevant to your asset and/or asset holders.",
    )
    manager: AlgorandAddress | None = Field(
        default=None,
        description="The address of the account that can manage the configuration of the asset and destroy it.",
    )
    reserve: AlgorandAddress | None = Field(
        default=None,
        description="The address of the account that holds the reserve (non-minted) units of the asset.",
    )
    freeze: AlgorandAddress | None = Field(
        default=None,
        description="The address of the account used to freeze holdings of this asset. If empty, freezing is not permitted.",
    )
    clawback: AlgorandAddress | None = Field(
        default=None,
        description="The address of the account that can clawback holdings of this asset. If empty, clawback is not permitted.",
    )

algobase.models.arc3

Pydantic models for Algorand ARC-3 metadata.

Reference: https://github.com/algorandfoundation/ARCs/blob/main/ARCs/arc-0003.md

Arc3Localization

Bases: BaseModel

A Pydantic model for Algorand ARC-3 localization.

Source code in algobase/models/arc3.py

class Arc3Localization(BaseModel):
    """A Pydantic model for Algorand ARC-3 localization."""

    model_config = ConfigDict(frozen=True)

    uri: Arc3LocalizedUrl = Field(
        description="The URI pattern to fetch localized data from. This URI should contain the substring `{locale}` which will be replaced with the appropriate locale value before sending the request."
    )
    default: UnicodeLocale = Field(
        description="The locale of the default data within the base JSON."
    )
    locales: list[UnicodeLocale] = Field(
        description="The list of locales for which data is available. These locales should conform to those defined in the Unicode Common UnicodeLocale Data Repository (http://cldr.unicode.org/)."
    )
    integrity: dict[UnicodeLocale, Arc3Sri] | None = Field(
        default=None,
        description="The SHA-256 digests of the localized JSON files (except the default one). The field name is the locale. The field value is a single SHA-256 integrity metadata as defined in the W3C subresource integrity specification (https://w3c.github.io/webappsec-subresource-integrity).",
    )

Arc3Metadata

Bases: BaseModel

A Pydantic model for Algorand ARC-3 metadata.

Source code in algobase/models/arc3.py

class Arc3Metadata(BaseModel):
    """A Pydantic model for Algorand ARC-3 metadata."""

    model_config = ConfigDict(frozen=True)

    arc: Literal[Arc.ARC3] = Field(
        default=Arc.ARC3,
        description="Name of the Algorand ARC standard that the NFT metadata adheres to.",
        exclude=True,
    )

    @property
    def json_str(self) -> str:
        """Returns the model JSON as a string."""
        return self.model_dump_json(exclude_none=True, indent=4)

    @property
    def json_bytes(self, encoding: Literal["utf-8"] = "utf-8") -> bytes:
        """Returns the model JSON encoded as bytes.

        Currently only officially supports UTF-8 encoding.
        """
        return self.json_str.encode(encoding)

    name: str | None = Field(
        default=None, description="Identifies the asset to which this token represents."
    )
    decimals: AsaDecimals | None = Field(
        default=None,
        description="The number of decimal places that the token amount should display - e.g. 18, means to divide the token amount by 1000000000000000000 to get its user representation.",
    )
    description: str | None = Field(
        default=None, description="Describes the asset to which this token represents."
    )
    image: Arc3Url | None = Field(
        default=None,
        description="A URI pointing to a file with MIME type image/* representing the asset to which this token represents. Consider making any images at a width between 320 and 1080 pixels and aspect ratio between 1.91:1 and 4:5 inclusive.",
    )
    image_integrity: Arc3Sri | None = Field(
        default=None,
        description="The SHA-256 digest of the file pointed by the URI image. The field value is a single SHA-256 integrity metadata as defined in the W3C subresource integrity specification (https://w3c.github.io/webappsec-subresource-integrity).",
    )
    image_mimetype: ImageMimeType | None = Field(
        default=None,
        description="The MIME type of the file pointed by the URI image. MUST be of the form 'image/*'.",
    )
    background_color: Arc3Color | None = Field(
        default=None,
        description="Background color do display the asset. MUST be a six-character hexadecimal without a pre-pended #.",
    )
    external_url: Arc3Url | None = Field(
        default=None,
        description="A URI pointing to an external website presenting the asset.",
    )
    external_url_integrity: Arc3Sri | None = Field(
        default=None,
        description="The SHA-256 digest of the file pointed by the URI external_url. The field value is a single SHA-256 integrity metadata as defined in the W3C subresource integrity specification (https://w3c.github.io/webappsec-subresource-integrity).",
    )
    external_url_mimetype: Literal["text/html"] | None = Field(
        default=None,
        description="The MIME type of the file pointed by the URI external_url. It is expected to be 'text/html' in almost all cases.",
    )
    animation_url: Arc3Url | None = Field(
        default=None,
        description="A URI pointing to a multi-media file representing the asset.",
    )
    animation_url_integrity: Arc3Sri | None = Field(
        default=None,
        description="The SHA-256 digest of the file pointed by the URI external_url. The field value is a single SHA-256 integrity metadata as defined in the W3C subresource integrity specification (https://w3c.github.io/webappsec-subresource-integrity).",
    )
    animation_url_mimetype: MimeType | None = Field(
        default=None,
        description="The MIME type of the file pointed by the URI animation_url. If the MIME type is not specified, clients MAY guess the MIME type from the file extension or MAY decide not to display the asset at all. It is STRONGLY RECOMMENDED to include the MIME type.",
    )
    properties: Arc3Properties | None = Field(
        default=None,
        description="Arbitrary properties (also called attributes). Values may be strings, numbers, object or arrays.",
    )
    extra_metadata: Base64Str | None = Field(
        default=None,
        description="Extra metadata in base64. If the field is specified (even if it is an empty string) the asset metadata (am) of the ASA is computed differently than if it is not specified.",
    )
    localization: Arc3Localization | None = Field(
        default=None,
        description="A sub-object that may be used to provide localized values for fields that need it.",
    )

json_bytes property

json_bytes: bytes

Returns the model JSON encoded as bytes.

Currently only officially supports UTF-8 encoding.

json_str property

json_str: str

Returns the model JSON as a string.

Arc3Properties

Bases: BaseModel

A Pydantic model for Algorand ARC-3 properties.

If the traits property is present, it must comply with ARC-16: https://github.com/algorandfoundation/ARCs/blob/main/ARCs/arc-0016.md

Source code in algobase/models/arc3.py

class Arc3Properties(BaseModel):
    """A Pydantic model for Algorand ARC-3 properties.

    If the `traits` property is present, it must comply with ARC-16: https://github.com/algorandfoundation/ARCs/blob/main/ARCs/arc-0016.md
    """

    model_config = ConfigDict(frozen=True, extra="allow")

    # Struggling to get recursive type definition working here.
    # Have defined `Arc3NonTraitProperties` in algobase/types/annotated.py
    # but it doesn't work as an annotation for __pydantic_extra__.
    __pydantic_extra__: dict[str, str | int | float | dict | list]  # type: ignore

    traits: Arc16Traits | None = Field(
        default=None,
        description="Traits (attributes) that can be used to calculate things like rarity. Values may be strings or numbers.",
    )

algobase.models.arc19

Pydantic models for Algorand ARC-19 metadata.

Reference: https://github.com/algorandfoundation/ARCs/blob/main/ARCs/arc-0019.md

Arc19Metadata

Bases: BaseModel

A Pydantic model for Algorand ARC-19 metadata.

Source code in algobase/models/arc19.py

class Arc19Metadata(BaseModel):
    """A Pydantic model for Algorand ARC-19 metadata."""

    model_config = ConfigDict(frozen=True)

    arc: Literal[Arc.ARC19] = Field(
        default=Arc.ARC19,
        description="Name of the Algorand ARC standard that the NFT metadata adheres to.",
        exclude=True,
    )

    arc3_metadata: Arc3Metadata | None = Field(
        default=None,
        description="Optional ARC-3 metadata model.",
    )

    @model_validator(mode="after")
    def validate_arc3_compliance(self) -> "Arc19Metadata":
        """If the ARC-3 metadata is present, ensure it complies with ARC-19.

        Raises:
            ValueError: If the ARC-3 metadata is present and does not comply with ARC-19.

        Returns:
            Arc19Metadata: The model instance.
        """
        if (
            self.arc3_metadata is not None
            and self.arc3_metadata.extra_metadata is not None
        ):
            raise ValueError("Extra metadata is not supported for ARC-19.")
        return self

validate_arc3_compliance

validate_arc3_compliance() -> Arc19Metadata

If the ARC-3 metadata is present, ensure it complies with ARC-19.

Raises:

• ValueError –

  If the ARC-3 metadata is present and does not comply with ARC-19.

Returns:

• Arc19Metadata ( Arc19Metadata ) –

  The model instance.

Source code in algobase/models/arc19.py

@model_validator(mode="after")
def validate_arc3_compliance(self) -> "Arc19Metadata":
    """If the ARC-3 metadata is present, ensure it complies with ARC-19.

    Raises:
        ValueError: If the ARC-3 metadata is present and does not comply with ARC-19.

    Returns:
        Arc19Metadata: The model instance.
    """
    if (
        self.arc3_metadata is not None
        and self.arc3_metadata.extra_metadata is not None
    ):
        raise ValueError("Extra metadata is not supported for ARC-19.")
    return self

algobase.models.asa

Pydantic models for Algorand Standard Assets (ASAs).

Asa

Bases: BaseModel

A Pydantic model for Algorand Standard Assets (ASAs).

Source code in algobase/models/asa.py

class Asa(BaseModel):
    """A Pydantic model for Algorand Standard Assets (ASAs)."""

    model_config = ConfigDict(frozen=True)

    asa_type: AsaTypeChoice | None = Field(
        default=None, description="The type of the ASA."
    )
    asset_params: AssetParams = Field(description="AssetParams Pydantic model.")
    metadata: Arc3Metadata | Arc19Metadata | None = Field(
        default=None, description="Metadata Pydantic model.", discriminator="arc"
    )

    @property
    def derived_asa_type(self) -> AsaTypeChoice:
        """The derived type of the ASA."""
        match self.asset_params:
            case asa if asa.total == 1 and asa.decimals == 0:
                return AsaType.NON_FUNGIBLE_PURE
            # Means the total supply is 1
            case asa if asa.decimals == math.log10(asa.total):
                return AsaType.NON_FUNGIBLE_FRACTIONAL
            case _:
                return AsaType.FUNGIBLE

    @property
    def derived_arc3_metadata(self) -> Arc3Metadata | None:
        """The derived ARC-3 metadata."""
        match self.metadata:
            case Arc3Metadata():
                return self.metadata
            case Arc19Metadata() if isinstance(
                self.metadata.arc3_metadata, Arc3Metadata
            ):
                return self.metadata.arc3_metadata
            case _:
                return None

    @computed_field  # type: ignore[misc]
    @property
    def metadata_hash(self) -> AlgorandHash | None:
        """The hash of the JSON metadata."""
        if (metadata := self.derived_arc3_metadata) is None:
            return None
        if metadata.extra_metadata is None:
            return sha256(metadata.json_bytes)
        else:
            # am = SHA-512/256("arc0003/am" || SHA-512/256("arc0003/amj" || content of JSON Metadata file) || e)
            base_hash = sha512_256(b"arc0003/amj" + metadata.json_bytes)
            extra_metadata_bytes = b64decode(metadata.extra_metadata)
            return sha512_256(b"arc0003/am" + base_hash + extra_metadata_bytes)

    @model_validator(mode="after")
    def check_asa_type_constraints(self) -> "Asa":
        """Validate the ASA type against the relevant constraints."""
        if self.asa_type is not None and self.asa_type != self.derived_asa_type:
            match self.asa_type:
                case AsaType.NON_FUNGIBLE_PURE:
                    raise ValueError(
                        "Total number of units must be 1 and number of digits after the decimal point must be 0 for a pure NFT."
                    )
                case AsaType.NON_FUNGIBLE_FRACTIONAL:
                    raise ValueError(
                        "Number of digits after the decimal point must be equal to the logarithm in base 10 of total number of units. In other words, the total supply of the ASA must be exactly 1."
                    )
                case AsaType.FUNGIBLE:
                    raise ValueError(
                        "Total supply of the ASA must be greater than 1, for a fungible asset."
                    )
        return self

    @model_validator(mode="after")
    def check_arc_constraints(self) -> "Asa":
        """Validate fields against ARC constraints, if applicable."""
        if self.derived_arc3_metadata is not None:
            self.check_arc3_metadata_constraints(self.derived_arc3_metadata)
        match self.metadata:
            case Arc3Metadata():
                self.check_arc3_asset_url()
            case Arc19Metadata():
                if self.asset_params.url is None:
                    raise ValueError("Asset URL must not be `None`.")
                validate_arc19_asset_url(self.asset_params.url)
                self.check_arc19_reserve()

        return self

    def check_arc3_decimals(self, metadata: Arc3Metadata) -> "Asa":
        """Raise an error if the decimals in the asset parameters doesn't match the deimals in the metadata."""
        if (
            metadata.decimals is not None
            and metadata.decimals != self.asset_params.decimals
        ):
            raise ValueError(
                f"Decimals in the asset parameters ({self.asset_params.decimals}) must match the decimals in the metadata ({metadata.decimals})."
            )
        return self

    def check_arc3_unit_name(self, metadata: Arc3Metadata) -> "Asa":
        """Raise a warning if the metadata 'name' property is not related to the asset unit name.

        Uses the difflib `SequenceMatcher` for string similarity.
        """
        if (
            self.asset_params.unit_name is not None
            and metadata is not None
            and metadata.name is not None
            and SequenceMatcher(
                None, self.asset_params.unit_name.lower(), metadata.name.lower()
            ).ratio()
            < 0.5
        ):
            warnings.warn(
                UserWarning(
                    "Asset unit name should be related to the name in the ARC-3 JSON metadata."
                )
            )
        return self

    def check_arc3_asset_url(self) -> "Asa":
        """Checks that the asset URL is valid for ARC-3 ASAs."""
        if self.asset_params.url is None:
            raise ValueError("Asset URL must not be `None`.")
        if not self.asset_params.url.endswith("#arc3"):
            raise ValueError(
                f"Asset URL must end with '#arc3' if asset name is '{self.asset_params.asset_name}'."
            )
        validate_type_compatibility(self.asset_params.url, Url)
        return self

    def check_arc3_metadata_constraints(self, metadata: Arc3Metadata) -> "Asa":
        """Validate fields against ARC constraints, if applicable.

        Raises warnings for values/formats that are allowed but not recommended in ARC specs.
        Raises errors for values/formats that are not allowed in ARC specs.
        """
        # Currently only ARC3 is supported
        if isinstance(metadata, Arc3Metadata):
            self.check_arc3_unit_name(metadata)
            self.check_arc3_decimals(metadata)

            # Asset name constraints
            match self.asset_params.asset_name:
                case None:
                    raise ValueError("Asset name must not be `None` for ARC-3 ASAs.")
                case "arc3":
                    warnings.warn(
                        UserWarning(
                            "Asset name 'arc3' is not recommended for ARC-3 ASAs."
                        )
                    )
                case x if x.endswith("@arc3"):
                    warnings.warn(
                        UserWarning(
                            "Asset name format <name>@arc3 is not recommended for ARC-3 ASAs."
                        )
                    )
                # Constraints on combination of asset name and metadata name
                case _:
                    match metadata.name:
                        case None:
                            raise ValueError(
                                f"Metadata name must not be `None` if asset name is '{self.asset_params.asset_name}'."
                            )
                        case x if x != self.asset_params.asset_name:
                            if is_valid(
                                validate_type_compatibility,
                                metadata.name,
                                AsaAssetName,
                            ):
                                raise ValueError(
                                    f"Asset name '{self.asset_params.asset_name}' must match the metadata name '{x}'."
                                )
                            elif not metadata.name.startswith(
                                self.asset_params.asset_name
                            ):
                                raise ValueError(
                                    f"Asset name must be a shortened version of the metadata name '{metadata.name}'."
                                )
        return self

    def check_arc19_reserve(self) -> "Asa":
        """Checks that the CID encoded in the reserve address field is valid for ARC-19 ASAs."""
        if self.asset_params.reserve is None:
            raise ValueError("Reserve address must not be `None`.")
        return self

derived_arc3_metadata property

derived_arc3_metadata: Arc3Metadata | None

The derived ARC-3 metadata.

derived_asa_type property

derived_asa_type: AsaTypeChoice

The derived type of the ASA.

metadata_hash property

metadata_hash: AlgorandHash | None

The hash of the JSON metadata.

check_arc19_reserve

check_arc19_reserve() -> Asa

Checks that the CID encoded in the reserve address field is valid for ARC-19 ASAs.

Source code in algobase/models/asa.py

def check_arc19_reserve(self) -> "Asa":
    """Checks that the CID encoded in the reserve address field is valid for ARC-19 ASAs."""
    if self.asset_params.reserve is None:
        raise ValueError("Reserve address must not be `None`.")
    return self

check_arc3_asset_url

check_arc3_asset_url() -> Asa

Checks that the asset URL is valid for ARC-3 ASAs.

Source code in algobase/models/asa.py

def check_arc3_asset_url(self) -> "Asa":
    """Checks that the asset URL is valid for ARC-3 ASAs."""
    if self.asset_params.url is None:
        raise ValueError("Asset URL must not be `None`.")
    if not self.asset_params.url.endswith("#arc3"):
        raise ValueError(
            f"Asset URL must end with '#arc3' if asset name is '{self.asset_params.asset_name}'."
        )
    validate_type_compatibility(self.asset_params.url, Url)
    return self

check_arc3_decimals

check_arc3_decimals(metadata: Arc3Metadata) -> Asa

Raise an error if the decimals in the asset parameters doesn't match the deimals in the metadata.

Source code in algobase/models/asa.py

def check_arc3_decimals(self, metadata: Arc3Metadata) -> "Asa":
    """Raise an error if the decimals in the asset parameters doesn't match the deimals in the metadata."""
    if (
        metadata.decimals is not None
        and metadata.decimals != self.asset_params.decimals
    ):
        raise ValueError(
            f"Decimals in the asset parameters ({self.asset_params.decimals}) must match the decimals in the metadata ({metadata.decimals})."
        )
    return self

check_arc3_metadata_constraints

check_arc3_metadata_constraints(
    metadata: Arc3Metadata,
) -> Asa

Validate fields against ARC constraints, if applicable.

Raises warnings for values/formats that are allowed but not recommended in ARC specs. Raises errors for values/formats that are not allowed in ARC specs.

Source code in algobase/models/asa.py

def check_arc3_metadata_constraints(self, metadata: Arc3Metadata) -> "Asa":
    """Validate fields against ARC constraints, if applicable.

    Raises warnings for values/formats that are allowed but not recommended in ARC specs.
    Raises errors for values/formats that are not allowed in ARC specs.
    """
    # Currently only ARC3 is supported
    if isinstance(metadata, Arc3Metadata):
        self.check_arc3_unit_name(metadata)
        self.check_arc3_decimals(metadata)

        # Asset name constraints
        match self.asset_params.asset_name:
            case None:
                raise ValueError("Asset name must not be `None` for ARC-3 ASAs.")
            case "arc3":
                warnings.warn(
                    UserWarning(
                        "Asset name 'arc3' is not recommended for ARC-3 ASAs."
                    )
                )
            case x if x.endswith("@arc3"):
                warnings.warn(
                    UserWarning(
                        "Asset name format <name>@arc3 is not recommended for ARC-3 ASAs."
                    )
                )
            # Constraints on combination of asset name and metadata name
            case _:
                match metadata.name:
                    case None:
                        raise ValueError(
                            f"Metadata name must not be `None` if asset name is '{self.asset_params.asset_name}'."
                        )
                    case x if x != self.asset_params.asset_name:
                        if is_valid(
                            validate_type_compatibility,
                            metadata.name,
                            AsaAssetName,
                        ):
                            raise ValueError(
                                f"Asset name '{self.asset_params.asset_name}' must match the metadata name '{x}'."
                            )
                        elif not metadata.name.startswith(
                            self.asset_params.asset_name
                        ):
                            raise ValueError(
                                f"Asset name must be a shortened version of the metadata name '{metadata.name}'."
                            )
    return self

check_arc3_unit_name

check_arc3_unit_name(metadata: Arc3Metadata) -> Asa

Raise a warning if the metadata 'name' property is not related to the asset unit name.

Uses the difflib SequenceMatcher for string similarity.

Source code in algobase/models/asa.py

def check_arc3_unit_name(self, metadata: Arc3Metadata) -> "Asa":
    """Raise a warning if the metadata 'name' property is not related to the asset unit name.

    Uses the difflib `SequenceMatcher` for string similarity.
    """
    if (
        self.asset_params.unit_name is not None
        and metadata is not None
        and metadata.name is not None
        and SequenceMatcher(
            None, self.asset_params.unit_name.lower(), metadata.name.lower()
        ).ratio()
        < 0.5
    ):
        warnings.warn(
            UserWarning(
                "Asset unit name should be related to the name in the ARC-3 JSON metadata."
            )
        )
    return self

check_arc_constraints

check_arc_constraints() -> Asa

Validate fields against ARC constraints, if applicable.

Source code in algobase/models/asa.py

@model_validator(mode="after")
def check_arc_constraints(self) -> "Asa":
    """Validate fields against ARC constraints, if applicable."""
    if self.derived_arc3_metadata is not None:
        self.check_arc3_metadata_constraints(self.derived_arc3_metadata)
    match self.metadata:
        case Arc3Metadata():
            self.check_arc3_asset_url()
        case Arc19Metadata():
            if self.asset_params.url is None:
                raise ValueError("Asset URL must not be `None`.")
            validate_arc19_asset_url(self.asset_params.url)
            self.check_arc19_reserve()

    return self

check_asa_type_constraints

check_asa_type_constraints() -> Asa

Validate the ASA type against the relevant constraints.

Source code in algobase/models/asa.py

@model_validator(mode="after")
def check_asa_type_constraints(self) -> "Asa":
    """Validate the ASA type against the relevant constraints."""
    if self.asa_type is not None and self.asa_type != self.derived_asa_type:
        match self.asa_type:
            case AsaType.NON_FUNGIBLE_PURE:
                raise ValueError(
                    "Total number of units must be 1 and number of digits after the decimal point must be 0 for a pure NFT."
                )
            case AsaType.NON_FUNGIBLE_FRACTIONAL:
                raise ValueError(
                    "Number of digits after the decimal point must be equal to the logarithm in base 10 of total number of units. In other words, the total supply of the ASA must be exactly 1."
                )
            case AsaType.FUNGIBLE:
                raise ValueError(
                    "Total supply of the ASA must be greater than 1, for a fungible asset."
                )
    return self

algobase.models.algod

Pydantic models for the Algod API (v2).

Mostly auto-generated using datamodel-codegen. Spec: https://github.com/algorand/go-algorand/blob/master/daemon/algod/api/algod.oas3.yml

Account

Bases: BaseModel

Account information.

Source code in algobase/models/algod.py

class Account(BaseModel):
    """Account information."""

    address: str = Field(..., description="the account public key")
    amount: int = Field(
        ..., description="[algo] total number of MicroAlgos in the account"
    )
    amount_without_pending_rewards: int = Field(
        ...,
        alias="amount-without-pending-rewards",
        description="specifies the amount of MicroAlgos in the account, without the pending rewards.",
    )
    apps_local_state: list[ApplicationLocalState] | None = Field(
        None,
        alias="apps-local-state",
        description="[appl] applications local data stored in this account.\n\nNote the raw object uses `map[int] -> AppLocalState` for this type.",
    )
    apps_total_extra_pages: int | None = Field(
        None,
        alias="apps-total-extra-pages",
        description="[teap] the sum of all extra application program pages for this account.",
    )
    apps_total_schema: ApplicationStateSchema | None = Field(
        None, alias="apps-total-schema"
    )
    assets: list[AssetHolding] | None = Field(
        None,
        description="[asset] assets held by this account.\n\nNote the raw object uses `map[int] -> AssetHolding` for this type.",
    )
    auth_addr: str | None = Field(
        None,
        alias="auth-addr",
        description="[spend] the address against which signing should be checked. If empty, the address of the current account is used. This field can be updated in any transaction by setting the RekeyTo field.",
    )
    created_apps: list[Application] | None = Field(
        None,
        alias="created-apps",
        description="[appp] parameters of applications created by this account including app global data.\n\nNote: the raw account uses `map[int] -> AppParams` for this type.",
    )
    created_assets: list[Asset] | None = Field(
        None,
        alias="created-assets",
        description="[apar] parameters of assets created by this account.\n\nNote: the raw account uses `map[int] -> Asset` for this type.",
    )
    min_balance: int = Field(
        ...,
        alias="min-balance",
        description="MicroAlgo balance required by the account.\n\nThe requirement grows based on asset and application usage.",
    )
    participation: AccountParticipation | None = None
    pending_rewards: int = Field(
        ...,
        alias="pending-rewards",
        description="amount of MicroAlgos of pending rewards in this account.",
    )
    reward_base: int | None = Field(
        None,
        alias="reward-base",
        description="[ebase] used as part of the rewards computation. Only applicable to accounts which are participating.",
    )
    rewards: int = Field(
        ...,
        description="[ern] total rewards of MicroAlgos the account has received, including pending rewards.",
    )
    round: int = Field(
        ..., description="The round for which this information is relevant."
    )
    sig_type: SigType | None = Field(
        None,
        alias="sig-type",
        description="Indicates what type of signature is used by this account, must be one of:\n* sig\n* msig\n* lsig",
    )
    status: str = Field(
        ...,
        description="[onl] delegation status of the account's MicroAlgos\n* Offline - indicates that the associated account is delegated.\n*  Online  - indicates that the associated account used as part of the delegation pool.\n*   NotParticipating - indicates that the associated account is neither a delegator nor a delegate.",
    )
    total_apps_opted_in: int = Field(
        ...,
        alias="total-apps-opted-in",
        description="The count of all applications that have been opted in, equivalent to the count of application local data (AppLocalState objects) stored in this account.",
    )
    total_assets_opted_in: int = Field(
        ...,
        alias="total-assets-opted-in",
        description="The count of all assets that have been opted in, equivalent to the count of AssetHolding objects held by this account.",
    )
    total_box_bytes: int | None = Field(
        None,
        alias="total-box-bytes",
        description="\\[tbxb\\] The total number of bytes used by this account's app's box keys and values.",
    )
    total_boxes: int | None = Field(
        None,
        alias="total-boxes",
        description="\\[tbx\\] The number of existing boxes created by this account's app.",
    )
    total_created_apps: int = Field(
        ...,
        alias="total-created-apps",
        description="The count of all apps (AppParams objects) created by this account.",
    )
    total_created_assets: int = Field(
        ...,
        alias="total-created-assets",
        description="The count of all assets (AssetParams objects) created by this account.",
    )

AccountParticipation

Bases: BaseModel

Account participation information.

Source code in algobase/models/algod.py

class AccountParticipation(BaseModel):
    """Account participation information."""

    selection_participation_key: str = Field(
        ...,
        alias="selection-participation-key",
        description="[sel] Selection public key (if any) currently registered for this round.",
    )
    state_proof_key: str | None = Field(
        None,
        alias="state-proof-key",
        description="[stprf] Root of the state proof key (if any)",
    )
    vote_first_valid: int = Field(
        ...,
        alias="vote-first-valid",
        description="[voteFst] First round for which this participation is valid.",
    )
    vote_key_dilution: int = Field(
        ...,
        alias="vote-key-dilution",
        description="[voteKD] Number of subkeys in each batch of participation keys.",
    )
    vote_last_valid: int = Field(
        ...,
        alias="vote-last-valid",
        description="[voteLst] Last round for which this participation is valid.",
    )
    vote_participation_key: str = Field(
        ...,
        alias="vote-participation-key",
        description="[vote] root participation public key (if any) currently registered for this round.",
    )

AccountStateDelta

Bases: BaseModel

AccountStateDelta.

Source code in algobase/models/algod.py

class AccountStateDelta(BaseModel):
    """AccountStateDelta."""

    address: str
    delta: StateDelta

Application

Bases: BaseModel

Application information.

Source code in algobase/models/algod.py

class Application(BaseModel):
    """Application information."""

    id: int = Field(..., description="[appidx] application index.")
    params: ApplicationParams

ApplicationLocalState

Bases: BaseModel

Application local state.

Source code in algobase/models/algod.py

class ApplicationLocalState(BaseModel):
    """Application local state."""

    id: int = Field(..., description="The application which this local state is for.")
    key_value: TealKeyValueStore | None = Field(None, alias="key-value")
    schema_: ApplicationStateSchema = Field(..., alias="schema")

ApplicationParams

Bases: BaseModel

Application parameters.

Source code in algobase/models/algod.py

class ApplicationParams(BaseModel):
    """Application parameters."""

    approval_program: str = Field(
        ..., alias="approval-program", description="[approv] approval program."
    )
    clear_state_program: str = Field(
        ..., alias="clear-state-program", description="[clearp] approval program."
    )
    creator: str = Field(
        ...,
        description="The address that created this application. This is the address where the parameters and global state for this application can be found.",
    )
    extra_program_pages: int | None = Field(
        None,
        alias="extra-program-pages",
        description="[epp] the amount of extra program pages available to this app.",
    )
    global_state: TealKeyValueStore | None = Field(None, alias="global-state")
    global_state_schema: ApplicationStateSchema | None = Field(
        None, alias="global-state-schema"
    )
    local_state_schema: ApplicationStateSchema | None = Field(
        None, alias="local-state-schema"
    )

ApplicationStateSchema

Bases: BaseModel

Application state schema.

Source code in algobase/models/algod.py

class ApplicationStateSchema(BaseModel):
    """Application state schema."""

    num_byte_slice: int = Field(
        ..., alias="num-byte-slice", description="[nbs] num of byte slices."
    )
    num_uint: int = Field(..., alias="num-uint", description="[nui] num of uints.")

Asset

Bases: BaseModel

Asset information.

Source code in algobase/models/algod.py

class Asset(BaseModel):
    """Asset information."""

    index: int = Field(..., description="unique asset identifier")
    params: AssetParams

AssetHolding

Bases: BaseModel

Asset holding information.

Source code in algobase/models/algod.py

class AssetHolding(BaseModel):
    """Asset holding information."""

    amount: int = Field(..., description="[a] number of units held.")
    asset_id: int = Field(..., alias="asset-id", description="Asset ID of the holding.")
    is_frozen: bool = Field(
        ...,
        alias="is-frozen",
        description="[f] whether or not the holding is frozen.",
    )

AssetParams

Bases: BaseModel

Asset parameters.

Source code in algobase/models/algod.py

class AssetParams(BaseModel):
    """Asset parameters."""

    clawback: str | None = Field(
        None,
        description="[c] Address of account used to clawback holdings of this asset.  If empty, clawback is not permitted.",
    )
    creator: str = Field(
        ...,
        description="The address that created this asset. This is the address where the parameters for this asset can be found, and also the address where unwanted asset units can be sent in the worst case.",
    )
    decimals: Annotated[int, Field(strict=True, ge=0, le=19)] = Field(
        ...,
        description="[dc] The number of digits to use after the decimal point when displaying this asset. If 0, the asset is not divisible. If 1, the base unit of the asset is in tenths. If 2, the base unit of the asset is in hundredths, and so on. This value must be between 0 and 19 (inclusive).",
    )
    default_frozen: bool | None = Field(
        None,
        alias="default-frozen",
        description="[df] Whether holdings of this asset are frozen by default.",
    )
    freeze: str | None = Field(
        None,
        description="[f] Address of account used to freeze holdings of this asset.  If empty, freezing is not permitted.",
    )
    manager: str | None = Field(
        None,
        description="[m] Address of account used to manage the keys of this asset and to destroy it.",
    )
    metadata_hash: str | None = Field(
        None,
        alias="metadata-hash",
        description="[am] A commitment to some unspecified asset metadata. The format of this metadata is up to the application.",
    )
    name: str | None = Field(
        None,
        description="[an] Name of this asset, as supplied by the creator. Included only when the asset name is composed of printable utf-8 characters.",
    )
    name_b64: str | None = Field(
        None,
        alias="name-b64",
        description="Base64 encoded name of this asset, as supplied by the creator.",
    )
    reserve: str | None = Field(
        None,
        description="[r] Address of account holding reserve (non-minted) units of this asset.",
    )
    total: int = Field(..., description="[t] The total number of units of this asset.")
    unit_name: str | None = Field(
        None,
        alias="unit-name",
        description="[un] Name of a unit of this asset, as supplied by the creator. Included only when the name of a unit of this asset is composed of printable utf-8 characters.",
    )
    unit_name_b64: str | None = Field(
        None,
        alias="unit-name-b64",
        description="Base64 encoded name of a unit of this asset, as supplied by the creator.",
    )
    url: str | None = Field(
        None,
        description="[au] URL where more information about the asset can be retrieved. Included only when the URL is composed of printable utf-8 characters.",
    )
    url_b64: str | None = Field(
        None,
        alias="url-b64",
        description="Base64 encoded URL where more information about the asset can be retrieved.",
    )

EvalDelta

Bases: BaseModel

EvalDelta.

Source code in algobase/models/algod.py

class EvalDelta(BaseModel):
    """EvalDelta."""

    action: int = Field(..., description="\\[at\\] delta action.")
    bytes: str | None = Field(None, description="\\[bs\\] bytes value.")
    uint: int | None = Field(None, description="\\[ui\\] uint value.")

EvalDeltaKeyValue

Bases: BaseModel

EvalDeltaKeyValue.

Source code in algobase/models/algod.py

class EvalDeltaKeyValue(BaseModel):
    """EvalDeltaKeyValue."""

    key: str
    value: EvalDelta

PendingTransactionResponse

Bases: BaseModel

PendingTransactionResponse.

Source code in algobase/models/algod.py

class PendingTransactionResponse(BaseModel):
    """PendingTransactionResponse."""

    application_index: int | None = Field(
        None,
        alias="application-index",
        description="The application index if the transaction was found and it created an application.",
    )
    asset_closing_amount: int | None = Field(
        None,
        alias="asset-closing-amount",
        description="The number of the asset's unit that were transferred to the close-to address.",
    )
    asset_index: int | None = Field(
        None,
        alias="asset-index",
        description="The asset index if the transaction was found and it created an asset.",
    )
    close_rewards: int | None = Field(
        None,
        alias="close-rewards",
        description="Rewards in microalgos applied to the close remainder to account.",
    )
    closing_amount: int | None = Field(
        None, alias="closing-amount", description="Closing amount for the transaction."
    )
    confirmed_round: int | None = Field(
        None,
        alias="confirmed-round",
        description="The round where this transaction was confirmed, if present.",
    )
    global_state_delta: StateDelta | None = Field(None, alias="global-state-delta")
    inner_txns: list["PendingTransactionResponse"] | None = Field(
        None,
        alias="inner-txns",
        description="Inner transactions produced by application execution.",
    )
    local_state_delta: list[AccountStateDelta] | None = Field(
        None,
        alias="local-state-delta",
        description="Local state key/value changes for the application being executed by this transaction.",
    )
    logs: list[str] | None = Field(
        None, description="Logs for the application being executed by this transaction."
    )
    pool_error: str = Field(
        ...,
        alias="pool-error",
        description="Indicates that the transaction was kicked out of this node's transaction pool (and specifies why that happened).  An empty string indicates the transaction wasn't kicked out of this node's txpool due to an error.\n",
    )
    receiver_rewards: int | None = Field(
        None,
        alias="receiver-rewards",
        description="Rewards in microalgos applied to the receiver account.",
    )
    sender_rewards: int | None = Field(
        None,
        alias="sender-rewards",
        description="Rewards in microalgos applied to the sender account.",
    )
    txn: dict[str, Any] = Field(..., description="The raw signed transaction.")

SigType

Bases: StrEnum

Enumeration for signature types.

Source code in algobase/models/algod.py

class SigType(StrEnum):
    """Enumeration for signature types."""

    SIG = auto()
    MISG = auto()
    LSIG = auto()

TealKeyValue

Bases: BaseModel

Teal key-value pair.

Source code in algobase/models/algod.py

class TealKeyValue(BaseModel):
    """Teal key-value pair."""

    key: str
    value: TealValue

TealValue

Bases: BaseModel

Teal value.

Source code in algobase/models/algod.py

class TealValue(BaseModel):
    """Teal value."""

    bytes: str = Field(..., description="[tb] bytes value.")
    type: int = Field(
        ...,
        description="[tt] value type. Value `1` refers to **bytes**, value `2` refers to **uint**",
    )
    uint: int = Field(..., description="[ui] uint value.")

algobase.models.kmd

Pydantic models for the KMD API (v1).

Mostly auto-generated using datamodel-codegen. Spec: https://github.com/algorand/go-algorand/blob/master/daemon/kmd/api/swagger.json

APIV1GETWalletsResponse

Bases: BaseModel

The response from the GET /v1/wallets endpoint.

Source code in algobase/models/kmd.py

class APIV1GETWalletsResponse(BaseModel):
    """The response from the `GET /v1/wallets` endpoint."""

    error: bool | None
    message: str | None
    wallets: list[APIV1Wallet] | None

APIV1Wallet

Bases: BaseModel

A KMD wallet.

Source code in algobase/models/kmd.py

class APIV1Wallet(BaseModel):
    """A KMD wallet."""

    driver_name: str | None
    driver_version: int | None
    id: str | None
    mnemonic_ux: bool | None
    name: str | None
    supported_txs: list[TxType] | None

algobase.algorand.client

Classes and functions to configure and create Algorand API clients.

ClientConfig dataclass

Configuration for an Algorand API client.

Source code in algobase/algorand/client.py

@dataclass(frozen=True, slots=True)
class ClientConfig:
    """Configuration for an Algorand API client."""

    url: str
    credential: str
    headers: dict[str, str] | None = None

create_algod_client

create_algod_client(config: ClientConfig) -> AlgodClient

Create an AlgodClient instance from the given configuration.

Parameters:

• config (ClientConfig) –

  The configuration to use.

Returns:

• AlgodClient ( AlgodClient ) –

  The AlgodClient instance.

Source code in algobase/algorand/client.py

def create_algod_client(config: ClientConfig) -> AlgodClient:
    """Create an AlgodClient instance from the given configuration.

    Args:
        config (ClientConfig): The configuration to use.

    Returns:
        AlgodClient: The AlgodClient instance.
    """
    return AlgodClient(
        algod_token=config.credential, algod_address=config.url, headers=config.headers
    )

create_indexer_client

create_indexer_client(
    config: ClientConfig,
) -> IndexerClient

Create an IndexerClient instance from the given configuration.

Parameters:

• config (ClientConfig) –

  The configuration to use.

Returns:

• IndexerClient ( IndexerClient ) –

  The IndexerClient instance.

Source code in algobase/algorand/client.py

def create_indexer_client(config: ClientConfig) -> IndexerClient:
    """Create an IndexerClient instance from the given configuration.

    Args:
        config (ClientConfig): The configuration to use.

    Returns:
        IndexerClient: The IndexerClient instance.
    """
    return IndexerClient(
        indexer_token=config.credential,
        indexer_address=config.url,
        headers=config.headers,
    )

create_kmd_client

create_kmd_client(config: ClientConfig) -> KMDClient

Create a KMDClient instance from the given configuration.

Parameters:

• config (ClientConfig) –

  The configuration to use.

Returns:

• KMDClient ( KMDClient ) –

  The KMDClient instance.

Source code in algobase/algorand/client.py

def create_kmd_client(config: ClientConfig) -> KMDClient:
    """Create a KMDClient instance from the given configuration.

    Args:
        config (ClientConfig): The configuration to use.

    Returns:
        KMDClient: The KMDClient instance.
    """
    return KMDClient(kmd_token=config.credential, kmd_address=config.url)

create_localnet_algod_client

create_localnet_algod_client() -> AlgodClient

Create an AlgodClient instance for the localnet.

Returns:

• AlgodClient ( AlgodClient ) –

  The AlgodClient instance.

Source code in algobase/algorand/client.py

def create_localnet_algod_client() -> AlgodClient:
    """Create an AlgodClient instance for the localnet.

    Returns:
        AlgodClient: The AlgodClient instance.
    """
    return flow(AlgorandApi.ALGOD, create_localnet_default_config, create_algod_client)

create_localnet_default_config

create_localnet_default_config(
    api: AlgorandApiChoice,
) -> ClientConfig

Create a default configuration for the localnet.

Parameters:

• api (AlgorandApiChoice) –

  The API to configure.

Returns:

• ClientConfig ( ClientConfig ) –

  The default configuration.

Source code in algobase/algorand/client.py

def create_localnet_default_config(api: AlgorandApiChoice) -> ClientConfig:
    """Create a default configuration for the localnet.

    Args:
        api (AlgorandApiChoice): The API to configure.

    Returns:
        ClientConfig: The default configuration.
    """
    port = {AlgorandApi.ALGOD: 4001, AlgorandApi.INDEXER: 8980, AlgorandApi.KMD: 4002}[
        api
    ]
    return ClientConfig(url=f"http://localhost:{port}", credential="a" * 64)

create_localnet_indexer_client

create_localnet_indexer_client() -> IndexerClient

Create an IndexerClient instance for the localnet.

Returns:

• IndexerClient ( IndexerClient ) –

  The IndexerClient instance.

Source code in algobase/algorand/client.py

def create_localnet_indexer_client() -> IndexerClient:
    """Create an IndexerClient instance for the localnet.

    Returns:
        IndexerClient: The IndexerClient instance.
    """
    return flow(
        AlgorandApi.INDEXER, create_localnet_default_config, create_indexer_client
    )

create_localnet_kmd_client

create_localnet_kmd_client() -> KMDClient

Create a KMDClient instance for the localnet.

Returns:

• KMDClient ( KMDClient ) –

  The KMDClient instance.

Source code in algobase/algorand/client.py

def create_localnet_kmd_client() -> KMDClient:
    """Create a KMDClient instance for the localnet.

    Returns:
        KMDClient: The KMDClient instance.
    """
    return flow(AlgorandApi.KMD, create_localnet_default_config, create_kmd_client)

find_wallet_id

find_wallet_id(
    kmd_client: KMDClient, wallet_name: str
) -> str | None

Get the ID of a wallet from the KMD client.

Parameters:

• kmd_client (KMDClient) –

  The KMD client.

• wallet_name (str) –

  The name of the wallet.

Returns:

• str | None –

  str | None: The ID of the wallet if found, else None.

Source code in algobase/algorand/client.py

def find_wallet_id(kmd_client: KMDClient, wallet_name: str) -> str | None:
    """Get the ID of a wallet from the KMD client.

    Args:
        kmd_client (KMDClient): The KMD client.
        wallet_name (str): The name of the wallet.

    Returns:
        str | None: The ID of the wallet if found, else None.
    """
    return next(
        x.id
        for x in map(kmd.APIV1Wallet.model_validate, kmd_client.list_wallets())
        if x.name == wallet_name
    )

get_algonode_config

get_algonode_config(
    network: Literal[
        AlgorandNetwork.BETANET,
        AlgorandNetwork.TESTNET,
        AlgorandNetwork.MAINNET,
    ],
    api: Literal[AlgorandApi.ALGOD, AlgorandApi.INDEXER],
) -> ClientConfig

Get the client config for Algonode API.

Parameters:

• network (AlgorandNetworkChoice) –

  The Algorand network.

• api (Literal[ALGOD, INDEXER]) –

  The Algorand API.

Returns:

• ClientConfig ( ClientConfig ) –

  The client config object.

Source code in algobase/algorand/client.py

def get_algonode_config(
    network: Literal[
        AlgorandNetwork.BETANET, AlgorandNetwork.TESTNET, AlgorandNetwork.MAINNET
    ],
    api: Literal[AlgorandApi.ALGOD, AlgorandApi.INDEXER],
) -> ClientConfig:
    """Get the client config for Algonode API.

    Args:
        network (AlgorandNetworkChoice): The Algorand network.
        api (Literal[AlgorandApi.ALGOD, AlgorandApi.INDEXER]): The Algorand API.

    Returns:
        ClientConfig: The client config object.
    """
    return ClientConfig(
        url=f"https://{network}-{('idx', 'algod')[api == AlgorandApi.ALGOD]}.algonode.cloud",
        credential="",
    )

get_default_account

get_default_account(
    algod_client: AlgodClient,
    kmd_client: KMDClient | None = None,
) -> Account

Return an Account instance for the default account.

Parameters:

• algod_client (AlgodClient) –

  The Algod client.

• kmd_client (KMDClient | None, default: None ) –

  The KMD client. If None, a default instance will be created. Defaults to None.

Raises:

• ValueError –

  If the Algod client instance isn't connected to a localnet network.

Returns:

• Account ( Account ) –

  The default account.

Source code in algobase/algorand/client.py

def get_default_account(
    algod_client: AlgodClient, kmd_client: KMDClient | None = None
) -> Account:
    """Return an Account instance for the default account.

    Args:
        algod_client (AlgodClient): The Algod client.
        kmd_client (KMDClient | None, optional): The KMD client. If None, a default instance will be created. Defaults to None.

    Raises:
        ValueError: If the Algod client instance isn't connected to a localnet network.

    Returns:
        Account: The default account.
    """
    if not is_localnet(algod_client):
        raise ValueError("Algod client must be connected to a localnet network.")

    kmd_client = kmd_client or create_localnet_kmd_client()

    wallet_handle = kmd_client.init_wallet_handle(
        find_wallet_id(kmd_client, "unencrypted-default-wallet"), ""
    )

    return flow(
        wallet_handle,
        kmd_client.list_keys,
        lambda keys: match_account(algod_client, keys, is_default_account),
        partial(kmd_client.export_key, wallet_handle, ""),
        Account.from_private_key,
    )

is_default_account

is_default_account(account: algod.Account) -> bool

Check if an account is the default account.

Parameters:

• account (AlgodResponseType) –

  The account info.

Returns:

• bool ( bool ) –

  True if the account is the default account, else False.

Source code in algobase/algorand/client.py

def is_default_account(account: algod.Account) -> bool:
    """Check if an account is the default account.

    Args:
        account (AlgodResponseType): The account info.

    Returns:
        bool: True if the account is the default account, else False.
    """
    return account.status != "Offline" and account.amount > 1_000_000_000

is_localnet

is_localnet(algod_client: AlgodClient) -> bool

Check if the AlgodClient is connected to a localnet.

Parameters:

• algod_client (AlgodClient) –

  The AlgodClient instance.

Returns:

• bool ( bool ) –

  True if the client is connected to a localnet, else False.

Source code in algobase/algorand/client.py

def is_localnet(algod_client: AlgodClient) -> bool:
    """Check if the AlgodClient is connected to a localnet.

    Args:
        algod_client (AlgodClient): The AlgodClient instance.

    Returns:
        bool: True if the client is connected to a localnet, else False.
    """
    return algod_client.suggested_params().gen in {
        "devnet-v1",
        "sandnet-v1",
        "dockernet-v1",
    }

match_account

match_account(
    algod_client: AlgodClient,
    addresses: list[str],
    predicate: Callable[[algod.Account], bool],
) -> str

Find the first account that matches the predicate, given a list of addresses to lookup.

Parameters:

• algod_client (AlgodClient) –

  The Algod client.

• addresses (list[str]) –

  The addresses to check.

• predicate (Callable[[Account], bool]) –

  The predicate function.

Raises:

• ValueError –

  If no account is found where predicate(account) is True.

Returns:

• str ( str ) –

  The address of the matching account if found, else None.

Source code in algobase/algorand/client.py

def match_account(
    algod_client: AlgodClient,
    addresses: list[str],
    predicate: Callable[[algod.Account], bool],
) -> str:
    """Find the first account that matches the predicate, given a list of addresses to lookup.

    Args:
        algod_client (AlgodClient): The Algod client.
        addresses (list[str]): The addresses to check.
        predicate (Callable[[algod.Account], bool]): The predicate function.

    Raises:
        ValueError: If no account is found where predicate(account) is True.

    Returns:
        str: The address of the matching account if found, else None.
    """
    matched = first_true(
        addresses,
        predicate=lambda x: flow(
            x, algod_client.account_info, algod.Account.model_validate, predicate
        ),
    )
    if matched is None:
        raise ValueError("No account found.")
    return matched

algobase.ipfs.client_base

Abstract base class for IPFS clients.

IpfsClient

Bases: ABC

Abstract base class for IPFS clients.

Source code in algobase/ipfs/client_base.py

class IpfsClient(ABC):
    """Abstract base class for IPFS clients."""

    def __post_init__(self) -> None:
        """If an API key is required, check that it is present."""
        if self.is_api_key_required:
            self.check_api_key_is_present()

    @classmethod
    @abstractmethod
    def from_settings(cls, settings: Settings) -> Self:
        """Create an instance of the IPFS client from a settings object."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def ipfs_provider_name(self) -> IpfsProviderChoice:
        """The name of the IPFS provider."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def api_version(self) -> str:
        """The version of the IPFS provider's API."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def base_url(self) -> httpx.URL:
        """The base URL of the IPFS provider's API."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def is_api_key_required(self) -> bool:
        """Whether the IPFS provider requires an API key."""
        ...  # pragma: no cover

    @property
    @abstractmethod
    def api_key(self) -> str | None:
        """The API key."""
        ...  # pragma: no cover

    def check_api_key_is_present(self) -> None:
        """Checks that the IPFS provider's API key is present."""
        if self.is_api_key_required and self.api_key is None:
            raise ValueError(
                f"API key for {self.ipfs_provider_name} must be defined in .env file."
            )

    @abstractmethod
    def store_json(self, json: str | bytes) -> str:
        """Stores JSON data in IPFS.

        Args:
            json (str | bytes): The JSON to store.

        Returns:
            str: The IPFS CID of the stored data.
        """
        ...  # pragma: no cover

    @abstractmethod
    def fetch_pin_status(self, cid: str) -> IpfsPinStatusChoice:
        """Returns the pinning status of a file, by CID.

        Args:
            cid (str): The CID of the file to check.

        Returns:
            IpfsPinStatusChoice: The pin status of the CID.
        """
        ...  # pragma: no cover

api_key abstractmethod property

api_key: str | None

The API key.

api_version abstractmethod property

api_version: str

The version of the IPFS provider's API.

base_url abstractmethod property

base_url: URL

The base URL of the IPFS provider's API.

ipfs_provider_name abstractmethod property

ipfs_provider_name: IpfsProviderChoice

The name of the IPFS provider.

is_api_key_required abstractmethod property

is_api_key_required: bool

Whether the IPFS provider requires an API key.

check_api_key_is_present

check_api_key_is_present() -> None

Checks that the IPFS provider's API key is present.

Source code in algobase/ipfs/client_base.py

def check_api_key_is_present(self) -> None:
    """Checks that the IPFS provider's API key is present."""
    if self.is_api_key_required and self.api_key is None:
        raise ValueError(
            f"API key for {self.ipfs_provider_name} must be defined in .env file."
        )

fetch_pin_status abstractmethod

fetch_pin_status(cid: str) -> IpfsPinStatusChoice

Returns the pinning status of a file, by CID.

Parameters:

• cid (str) –

  The CID of the file to check.

Returns:

• IpfsPinStatusChoice ( IpfsPinStatusChoice ) –

  The pin status of the CID.

Source code in algobase/ipfs/client_base.py

@abstractmethod
def fetch_pin_status(self, cid: str) -> IpfsPinStatusChoice:
    """Returns the pinning status of a file, by CID.

    Args:
        cid (str): The CID of the file to check.

    Returns:
        IpfsPinStatusChoice: The pin status of the CID.
    """
    ...  # pragma: no cover

from_settings abstractmethod classmethod

from_settings(settings: Settings) -> Self

Create an instance of the IPFS client from a settings object.

Source code in algobase/ipfs/client_base.py

@classmethod
@abstractmethod
def from_settings(cls, settings: Settings) -> Self:
    """Create an instance of the IPFS client from a settings object."""
    ...  # pragma: no cover

store_json abstractmethod

store_json(json: str | bytes) -> str

Stores JSON data in IPFS.

Parameters:

• json (str | bytes) –

  The JSON to store.

Returns:

• str ( str ) –

  The IPFS CID of the stored data.

Source code in algobase/ipfs/client_base.py

@abstractmethod
def store_json(self, json: str | bytes) -> str:
    """Stores JSON data in IPFS.

    Args:
        json (str | bytes): The JSON to store.

    Returns:
        str: The IPFS CID of the stored data.
    """
    ...  # pragma: no cover

algobase.ipfs.nft_storage

IPFS client for nft.storage.

NftStorage dataclass

Bases: IpfsClient

IPFS client for nft.storage.

Requires the NFT_STORAGE_API_KEY environment variable to be set.

Source code in algobase/ipfs/nft_storage.py

@dataclass
class NftStorage(IpfsClient):
    """IPFS client for nft.storage.

    Requires the `NFT_STORAGE_API_KEY` environment variable to be set.
    """

    _api_key: str | None

    @classmethod
    def from_settings(cls, settings: Settings) -> Self:
        """Create an instance of the IPFS client from the settings object."""
        return cls(_api_key=settings.nft_storage_api_key)

    @property
    def ipfs_provider_name(self) -> IpfsProviderChoice:
        """The name of the IPFS provider."""
        return IpfsProvider.NFT_STORAGE

    @property
    def api_version(self) -> str:
        """The version of the IPFS provider's API."""
        return "1.0"

    @property
    def base_url(self) -> httpx.URL:
        """The base URL of the IPFS provider's API."""
        return httpx.URL("https://api.nft.storage")

    @property
    def is_api_key_required(self) -> bool:
        """Whether the IPFS provider requires an API key."""
        return True

    @property
    def api_key(self) -> str | None:
        """The API key."""
        return self._api_key

    @property
    def headers(self) -> dict[str, str]:
        """The headers to use for the HTTP requests."""
        return {"Authorization": f"Bearer {self.api_key}"}

    def store_json(self, json: str | bytes) -> str:
        """Stores JSON data in IPFS.

        Args:
            json (str | bytes): The JSON to store.

        Returns:
            str: The IPFS CID of the stored data.
        """
        with httpx.Client() as client:
            response = client.post(
                url=self.base_url.join("upload"),
                content=json,
                headers=self.headers,
                timeout=10.0,
            )
            data = response.json()
            if response.status_code == httpx.codes.OK:
                if (
                    data.get("ok") is True
                    and (cid := data.get("value").get("cid")) is not None
                ):
                    return str(cid)
                else:
                    raise httpx.HTTPError(
                        f"HTTP Exception for {response.request.url}: Failed to store JSON in IPFS using {self.ipfs_provider_name}."
                    )
            else:
                raise httpx.HTTPError(
                    f"HTTP Exception for {response.request.url}: {response.status_code} {data.get('error').get('message')}"
                )

    def fetch_pin_status(self, cid: str) -> IpfsPinStatusChoice:
        """Returns the pinning status of a file, by CID.

        Args:
            cid (str): The CID of the file to check.

        Returns:
            IpfsPinStatusChoice: The pin status of the CID.
        """
        with httpx.Client() as client:
            response = client.get(
                url=self.base_url.join(f"check/{cid}"),
                headers=self.headers,
                timeout=10.0,
            )
            data = response.json()
            if response.status_code == httpx.codes.OK:
                pin_status = data.get("value").get("pin").get("status")
                if (
                    data.get("ok") is True
                    and pin_status is not None
                    and hasattr(IpfsPinStatus, str(pin_status).upper())
                ):
                    return IpfsPinStatus(pin_status)
                else:
                    raise httpx.HTTPError(
                        f"HTTP Exception for {response.request.url}: {pin_status} is not a valid pin status."
                    )
            else:
                raise httpx.HTTPError(
                    f"HTTP Exception for {response.request.url}: {response.status_code} {data.get('error').get('message')}"
                )

api_key property

api_key: str | None

The API key.

api_version property

api_version: str

The version of the IPFS provider's API.

base_url property

base_url: URL

The base URL of the IPFS provider's API.

headers property

headers: dict[str, str]

The headers to use for the HTTP requests.

ipfs_provider_name property

ipfs_provider_name: IpfsProviderChoice

The name of the IPFS provider.

is_api_key_required property

is_api_key_required: bool

Whether the IPFS provider requires an API key.

fetch_pin_status

fetch_pin_status(cid: str) -> IpfsPinStatusChoice

Returns the pinning status of a file, by CID.

Parameters:

• cid (str) –

  The CID of the file to check.

Returns:

• IpfsPinStatusChoice ( IpfsPinStatusChoice ) –

  The pin status of the CID.

Source code in algobase/ipfs/nft_storage.py

def fetch_pin_status(self, cid: str) -> IpfsPinStatusChoice:
    """Returns the pinning status of a file, by CID.

    Args:
        cid (str): The CID of the file to check.

    Returns:
        IpfsPinStatusChoice: The pin status of the CID.
    """
    with httpx.Client() as client:
        response = client.get(
            url=self.base_url.join(f"check/{cid}"),
            headers=self.headers,
            timeout=10.0,
        )
        data = response.json()
        if response.status_code == httpx.codes.OK:
            pin_status = data.get("value").get("pin").get("status")
            if (
                data.get("ok") is True
                and pin_status is not None
                and hasattr(IpfsPinStatus, str(pin_status).upper())
            ):
                return IpfsPinStatus(pin_status)
            else:
                raise httpx.HTTPError(
                    f"HTTP Exception for {response.request.url}: {pin_status} is not a valid pin status."
                )
        else:
            raise httpx.HTTPError(
                f"HTTP Exception for {response.request.url}: {response.status_code} {data.get('error').get('message')}"
            )

from_settings classmethod

from_settings(settings: Settings) -> Self

Create an instance of the IPFS client from the settings object.

Source code in algobase/ipfs/nft_storage.py

@classmethod
def from_settings(cls, settings: Settings) -> Self:
    """Create an instance of the IPFS client from the settings object."""
    return cls(_api_key=settings.nft_storage_api_key)

store_json

store_json(json: str | bytes) -> str

Stores JSON data in IPFS.

Parameters:

• json (str | bytes) –

  The JSON to store.

Returns:

• str ( str ) –

  The IPFS CID of the stored data.

Source code in algobase/ipfs/nft_storage.py

def store_json(self, json: str | bytes) -> str:
    """Stores JSON data in IPFS.

    Args:
        json (str | bytes): The JSON to store.

    Returns:
        str: The IPFS CID of the stored data.
    """
    with httpx.Client() as client:
        response = client.post(
            url=self.base_url.join("upload"),
            content=json,
            headers=self.headers,
            timeout=10.0,
        )
        data = response.json()
        if response.status_code == httpx.codes.OK:
            if (
                data.get("ok") is True
                and (cid := data.get("value").get("cid")) is not None
            ):
                return str(cid)
            else:
                raise httpx.HTTPError(
                    f"HTTP Exception for {response.request.url}: Failed to store JSON in IPFS using {self.ipfs_provider_name}."
                )
        else:
            raise httpx.HTTPError(
                f"HTTP Exception for {response.request.url}: {response.status_code} {data.get('error').get('message')}"
            )

algobase.utils.hash

Utility functions for hashing data.

sha256

sha256(data: bytes) -> bytes

Returns a SHA-256 hash digest of the input data.

Parameters:

• data (bytes) –

  The data to hash.

Returns:

• bytes ( bytes ) –

  The hash digest.

Source code in algobase/utils/hash.py

def sha256(data: bytes) -> bytes:
    """Returns a SHA-256 hash digest of the input data.

    Args:
        data (bytes): The data to hash.

    Returns:
        bytes: The hash digest.
    """
    return hashlib.sha256(data).digest()

sha512_256

sha512_256(data: bytes) -> bytes

Returns a SHA-512/256 hash digest of the input data.

Parameters:

• data (bytes) –

  The data to hash.

Returns:

• bytes ( bytes ) –

  The hash digest.

Source code in algobase/utils/hash.py

def sha512_256(data: bytes) -> bytes:
    """Returns a SHA-512/256 hash digest of the input data.

    Args:
        data (bytes): The data to hash.

    Returns:
        bytes: The hash digest.
    """
    return hashlib.new("sha512_256", data).digest()

algobase.utils.read

Functions for reading and caching reference data files.

read_ipfs_gateways

read_ipfs_gateways() -> list[str]

Read IPFS gateways from the reference data file.

Returns:

• list[str] –

  list[str]: The list of IPFS gateways.

Source code in algobase/utils/read.py

def read_ipfs_gateways() -> list[str]:
    """Read IPFS gateways from the reference data file.

    Returns:
        list[str]: The list of IPFS gateways.
    """
    with open("algobase/data/ipfs.toml", "rb") as f:
        data = tomllib.load(f)
    return list(data["ipfs_gateways"])

read_mime_types

read_mime_types() -> list[str]

Read MIME types from the reference data file.

Returns:

• list[str] –

  list[str]: The list of MIME types.

Source code in algobase/utils/read.py

def read_mime_types() -> list[str]:
    """Read MIME types from the reference data file.

    Returns:
        list[str]: The list of MIME types.
    """
    mimetypes.init()
    return list(mimetypes.types_map.values())

algobase.utils.url

Functions for working with URLs.

decode_url_braces

decode_url_braces(url: str) -> str

Decodes curly braces in a URL string.

This allows for arbitrary parameters to be passed in URL strings, as specified in some Algorand standards. For example, ARC-3 asset URLs may contain the string '{id}', which clients must replace with the asset ID in decimal form.

Parameters:

• url (str) –

  The URL string to decode.

Returns:

• str ( str ) –

  The decoded URL string.

Source code in algobase/utils/url.py

def decode_url_braces(url: str) -> str:
    """Decodes curly braces in a URL string.

    This allows for arbitrary parameters to be passed in URL strings, as specified in some Algorand standards.
    For example, ARC-3 asset URLs may contain the string '{id}', which clients must replace with the asset ID in decimal form.

    Args:
        url (str): The URL string to decode.

    Returns:
        str: The decoded URL string.
    """
    parsed_url = urlparse(url)
    decoded_path = parsed_url.path.replace(quote("{"), "{").replace(quote("}"), "}")
    decoded_url = parsed_url._replace(path=decoded_path).geturl()
    return decoded_url

algobase.utils.validate

Functions for data validation.

is_valid

is_valid(
    func: Callable[..., Any], *args: Any, **kwargs: Any
) -> bool

Checks if a function call is valid.

The other functions in this module raise errors when the input is not valid. This is a convenience function to check if a function call is valid without raising an error.

Parameters:

• func (Callable[..., Any]) –

  The function to call.

• *args (Any, default: () ) –

  Variable length argument list.

• **kwargs (Any, default: {} ) –

  Arbitrary keyword arguments.

Returns:

• bool ( bool ) –

  True if the function call doesn't raise a ValueError, else False.

Source code in algobase/utils/validate.py

def is_valid(func: Callable[..., Any], *args: Any, **kwargs: Any) -> bool:
    """Checks if a function call is valid.

    The other functions in this module raise errors when the input is not valid.
    This is a convenience function to check if a function call is valid without raising an error.

    Args:
        func (Callable[..., Any]): The function to call.
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        bool: True if the function call doesn't raise a ValueError, else False.
    """
    try:
        func(*args, **kwargs)
        return True
    except ValueError:
        return False

validate_address

validate_address(value: str) -> str

Checks that the value is a valid Algorand address.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid Algorand address.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_address(value: str) -> str:
    """Checks that the value is a valid Algorand address.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid Algorand address.

    Returns:
        str: The value passed in.
    """
    if not is_valid_address(value):
        raise ValueError(f"'{value}' is not a valid Algorand address.")
    return value

validate_arc19_asset_url

validate_arc19_asset_url(value: str) -> str

Checks that the value is a valid URL for Algorand ARC-19.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid URL for Algorand ARC-19.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_arc19_asset_url(value: str) -> str:
    """Checks that the value is a valid URL for Algorand ARC-19.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid URL for Algorand ARC-19.

    Returns:
        str: The value passed in.
    """
    # Extract the template substring from the URL, e.g. {ipfscid:0:dag-pb:reserve:sha2-256}
    template = value[value.find("{") + 1 : value.find("}")]
    match template.split(":"):
        case ["ipfscid", "0", "dag-pb", "reserve", "sha2-256"]:
            ...
        case ["ipfscid", "1", "raw" | "dag-pb", "reserve", "sha2-256"]:
            ...
        case _:
            raise ValueError("Asset URL template must follow ARC-19 specification.")
    return value

validate_arc3_sri

validate_arc3_sri(value: str) -> str

Checks that the value is a valid SHA-256 Subresource Integrity (SRI) value.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid SRI.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_arc3_sri(value: str) -> str:
    """Checks that the value is a valid SHA-256 Subresource Integrity (SRI) value.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid SRI.

    Returns:
        str: The value passed in.
    """
    if not value.startswith("sha256-"):
        raise ValueError(
            f"'{value}' is not a valid ARC-3 SRI. String must start with 'sha256-'."
        )
    return validate_sri(value)

validate_base64

validate_base64(value: str) -> str

Checks that the value is a valid base64 string.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid base64 string.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_base64(value: str) -> str:
    """Checks that the value is a valid base64 string.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid base64 string.

    Returns:
        str: The value passed in.
    """
    try:
        base64.b64decode(value, validate=True)
    except binascii.Error:
        raise ValueError(f"'{value}' is not valid base64.")
    return value

validate_contains_substring

validate_contains_substring(
    value: str | Url, substring: str
) -> str | Url

Checks that the value contains the substring.

Parameters:

• value (str | Url) –

  The value to check.

• substring (str) –

  The substring to check for.

Raises:

• ValueError –

  If the value does not contain the substring.

Returns:

• str | Url –

  str | Url: The value passed in.

Source code in algobase/utils/validate.py

def validate_contains_substring(value: str | Url, substring: str) -> str | Url:
    """Checks that the value contains the substring.

    Args:
        value (str | Url): The value to check.
        substring (str): The substring to check for.

    Raises:
        ValueError: If the value does not contain the substring.

    Returns:
        str | Url: The value passed in.
    """
    value_string = value if isinstance(value, str) else value.unicode_string()
    if substring not in value_string:
        raise ValueError(f"'{value_string}' does not contain subtring '{substring}'.")
    return value

validate_encoded_length

validate_encoded_length(
    value: str | Url, max_length: int
) -> str | Url

Checks that the value is not longer than max_length when encoded in UTF-8.

Parameters:

• value (str | Url) –

  The value to check.

• max_length (int) –

  The maximum length of the value when encoded in UTF-8.

Raises:

• ValueError –

  If the value is longer than max_length when encoded in UTF-8.

Returns:

• str | Url –

  str | Url: The value passed in.

Source code in algobase/utils/validate.py

def validate_encoded_length(value: str | Url, max_length: int) -> str | Url:
    """Checks that the value is not longer than `max_length` when encoded in UTF-8.

    Args:
        value (str | Url): The value to check.
        max_length (int): The maximum length of the value when encoded in UTF-8.

    Raises:
        ValueError: If the value is longer than `max_length` when encoded in UTF-8.

    Returns:
        str | Url: The value passed in.
    """
    url = value if isinstance(value, str) else value.unicode_string()
    if len(url.encode("utf-8")) > max_length:
        raise ValueError(f"'{value}' is > {max_length} bytes when encoded in UTF-8.")
    return value

validate_hex

validate_hex(value: str) -> str

Checks that the value is a valid hexadecimal string.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid hexadecimal string.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_hex(value: str) -> str:
    """Checks that the value is a valid hexadecimal string.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid hexadecimal string.

    Returns:
        str: The value passed in.
    """
    if not all(x in string.hexdigits for x in value):
        raise ValueError(f"'{value}' is not a valid hex string.")
    return value

validate_is_power_of_10

validate_is_power_of_10(n: int) -> int

Checks that the value is a power of 10.

Parameters:

• n (int) –

  The value to check.

Raises:

• ValueError –

  If the value is not a power of 10.

Returns:

• int ( int ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_is_power_of_10(n: int) -> int:
    """Checks that the value is a power of 10.

    Args:
        n (int): The value to check.

    Raises:
        ValueError: If the value is not a power of 10.

    Returns:
        int: The value passed in.
    """
    if not (n > 0 and math.log10(n).is_integer()):
        raise ValueError(f"{n} is not a power of 10.")
    return n

validate_locale

validate_locale(value: str) -> str

Checks that the value is a valid Unicode CLDR locale.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid locale identifier.

• UnknownLocaleError –

  If the value is not a valid Unicode CLDR locale.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_locale(value: str) -> str:
    """Checks that the value is a valid Unicode CLDR locale.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid locale identifier.
        UnknownLocaleError: If the value is not a valid Unicode CLDR locale.

    Returns:
        str: The value passed in.
    """
    try:
        Locale.parse(value)
    except ValueError as e:
        raise ValueError(f"'{value}' is not a valid locale identifier: {e}")
    except UnknownLocaleError:
        raise ValueError(f"'{value}' is not a valid Unicode CLDR locale.")
    return value

validate_mime_type cached

validate_mime_type(
    value: str, primary_type: str | None = None
) -> str

Checks that the value is a valid MIME type.

If primary_type is not None, then the value must be a valid MIME type with the specified primary type. E.g. if primary_type is 'image', then the value must be a valid MIME type starting with 'image/'.

Parameters:

• value (str) –

  The value to check.

• primary_type (str | None, default: None ) –

  The primary type of the MIME type. Defaults to None.

Raises:

• ValueError –

  If the value is not a valid MIME type.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

@cache
def validate_mime_type(value: str, primary_type: str | None = None) -> str:
    """Checks that the value is a valid MIME type.

    If `primary_type` is not `None`, then the value must be a valid MIME type with the specified primary type.
    E.g. if `primary_type` is 'image', then the value must be a valid MIME type starting with 'image/'.

    Args:
        value (str): The value to check.
        primary_type (str | None, optional): The primary type of the MIME type. Defaults to None.

    Raises:
        ValueError: If the value is not a valid MIME type.

    Returns:
        str: The value passed in.
    """
    if value not in read_mime_types():
        raise ValueError(f"'{value}' is not a valid MIME type.")
    if primary_type is not None and not value.startswith(f"{primary_type}/"):
        raise ValueError(f"'{value}' is not a valid {primary_type} MIME type.")
    return value

validate_not_in

validate_not_in(
    iterable: Iterable[str], element: str
) -> Iterable[str]

Checks that the element is not in the iterable.

Parameters:

• element (str) –

  The element to check for.

• iterable (Iterable) –

  The iterable to check.

Raises:

• ValueError –

  If the element is in the iterable.

Returns:

• Iterable ( Iterable[str] ) –

  The iterable passed in.

Source code in algobase/utils/validate.py

def validate_not_in(iterable: Iterable[str], element: str) -> Iterable[str]:
    """Checks that the element is not in the iterable.

    Args:
        element (str): The element to check for.
        iterable (Iterable): The iterable to check.

    Raises:
        ValueError: If the element is in the iterable.

    Returns:
        Iterable: The iterable passed in.
    """
    if element in iterable:
        raise ValueError(f"'{element}' is in {iterable}.")
    return iterable

validate_not_ipfs_gateway cached

validate_not_ipfs_gateway(url: str) -> str

Checks that the URL host is not a known public IPFS gateway.

Parameters:

• url (str) –

  The URL to check.

Raises:

• ValueError –

  If the URL host is a known public IPFS gateway.

Returns:

• str ( str ) –

  The URL passed in.

Source code in algobase/utils/validate.py

@cache
def validate_not_ipfs_gateway(url: str) -> str:
    """Checks that the URL host is not a known public IPFS gateway.

    Args:
        url (str): The URL to check.

    Raises:
        ValueError: If the URL host is a known public IPFS gateway.

    Returns:
        str: The URL passed in.
    """
    gateways = read_ipfs_gateways()
    if any(Url(gateway).host == Url(url).host for gateway in gateways):
        raise ValueError(f"'{Url(url).host}' is an IPFS gateway.")
    return url

validate_sri

validate_sri(value: str) -> str

Checks that the value is a valid W3C Subresource Integrity (SRI) value.

Parameters:

• value (str) –

  The value to check.

Raises:

• ValueError –

  If the value is not a valid SRI.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_sri(value: str) -> str:
    """Checks that the value is a valid W3C Subresource Integrity (SRI) value.

    Args:
        value (str): The value to check.

    Raises:
        ValueError: If the value is not a valid SRI.

    Returns:
        str: The value passed in.
    """
    supported_algorithms = {"sha256", "sha384", "sha512"}
    hash_algorithm = next(
        (x for x in supported_algorithms if value.startswith(f"{x}-")), None
    )
    if hash_algorithm is None:
        raise ValueError(
            f"'{value}' is not a valid SRI. String must start with 'sha256-', 'sha384-', or 'sha512-'."
        )
    hasher = hashlib.new(hash_algorithm)
    hash_digest = value.removeprefix(f"{hash_algorithm}-")
    try:
        validate_base64(hash_digest)
    except ValueError as e:
        raise ValueError(f"'{value}' is not a valid SRI. Hash digest {e}")
    if len(base64.b64decode(hash_digest)) != hasher.digest_size:
        raise ValueError(
            f"'{value}' is not a valid SRI. Expected {hasher.digest_size} byte hash digest, got {len(hash_digest)} bytes."
        )
    return value

validate_type_compatibility

validate_type_compatibility(value: str, _type: type) -> str

Checks that the value is compatible with the annotated type.

Parameters:

• value (str) –

  The value to check.

• _type (Type) –

  The type to validate against.

Raises:

• ValidationError –

  If the value is not compatible with the type.

Returns:

• str ( str ) –

  The value passed in.

Source code in algobase/utils/validate.py

def validate_type_compatibility(value: str, _type: type) -> str:
    """Checks that the value is compatible with the annotated type.

    Args:
        value (str): The value to check.
        _type (Type): The type to validate against.

    Raises:
        ValidationError: If the value is not compatible with the type.

    Returns:
        str: The value passed in.
    """
    TypeAdapter(_type).validate_python(value)
    return value

algobase.choices

Enums and enum type aliases for algobase.

AlgorandApi

Bases: StrEnum

An enumeration of Algorand APIs.

Source code in algobase/choices.py

class AlgorandApi(StrEnum):
    """An enumeration of Algorand APIs."""

    ALGOD = auto()
    INDEXER = auto()
    KMD = auto()

AlgorandApiProvider

Bases: StrEnum

An enumeration of Algorand API providers.

Source code in algobase/choices.py

class AlgorandApiProvider(StrEnum):
    """An enumeration of Algorand API providers."""

    LOCALHOST = auto()
    CUSTOM = auto()
    ALGONODE = auto()

AlgorandAsset

Bases: IntEnum

An enumeration of Algorand asset names and IDs.

Source code in algobase/choices.py

class AlgorandAsset(IntEnum):
    """An enumeration of Algorand asset names and IDs."""

    ALGO = 0

AlgorandNetwork

Bases: StrEnum

An enumeration of Algorand networks.

Source code in algobase/choices.py

class AlgorandNetwork(StrEnum):
    """An enumeration of Algorand networks."""

    LOCALNET = auto()
    BETANET = auto()
    TESTNET = auto()
    MAINNET = auto()

Arc

Bases: StrEnum

An enumeration of Algorand ARC standards that are supported in algobase.

Source code in algobase/choices.py

class Arc(StrEnum):
    """An enumeration of Algorand ARC standards that are supported in algobase."""

    ARC3 = auto()
    ARC19 = auto()

AsaType

Bases: StrEnum

An enumeration of Algorand Standard Asset (ASA) types.

Source code in algobase/choices.py

class AsaType(StrEnum):
    """An enumeration of Algorand Standard Asset (ASA) types."""

    FUNGIBLE = auto()
    NON_FUNGIBLE_PURE = auto()
    NON_FUNGIBLE_FRACTIONAL = auto()

IpfsPinStatus

Bases: StrEnum

An enumeration of IPFS pin statuses.

Source code in algobase/choices.py

class IpfsPinStatus(StrEnum):
    """An enumeration of IPFS pin statuses."""

    QUEUED = auto()
    PINNING = auto()
    PINNED = auto()
    FAILED = auto()

IpfsProvider

Bases: StrEnum

An enumeration of IPFS providers.

Source code in algobase/choices.py

class IpfsProvider(StrEnum):
    """An enumeration of IPFS providers."""

    NFT_STORAGE = auto()

algobase.functional

Functions for type casting.

first_true

first_true(
    iterable: Iterable[IT],
    default: IT | None = None,
    predicate: Callable[[IT], bool] | None = None,
) -> IT | None

Returns the first true value in the iterable.

If no true value is found, it returns default. If predicate is not None, it returns the first item for which predicate(item) is true.

Parameters:

• iterable (Iterable[IT]) –

  The iterable.

• default (IT | None, default: None ) –

  The default value to return if no true value is found. Defaults to None.

• predicate (Callable[[IT], bool] | None, default: None ) –

  The predicate function. Defaults to None.

Returns:

• IT | None –

  IT | None: The item in the iterable that is true, or default if no true value is found.

Source code in algobase/functional.py

def first_true(
    iterable: Iterable[IT],
    default: IT | None = None,
    predicate: Callable[[IT], bool] | None = None,
) -> IT | None:
    """Returns the first true value in the iterable.

    If no true value is found, it returns `default`.
    If `predicate` is not None, it returns the first item for which predicate(item) is true.

    Args:
        iterable (Iterable[IT]): The iterable.
        default (IT | None, optional): The default value to return if no true value is found. Defaults to None.
        predicate (Callable[[IT], bool] | None, optional): The predicate function. Defaults to None.

    Returns:
        IT | None: The item in the iterable that is true, or `default` if no true value is found.
    """
    return next(filter(predicate, iterable), default)

maybe_apply

maybe_apply(x: A | None, f: Callable[[A], B]) -> B | None

Return the result of applying the function to the value if the value is not None, otherwise return None.

Parameters:

• x (A | None) –

  The value to apply the function to.

• f (Callable[[A], B | None]) –

  The function to apply to the value.

Returns:

• B | None –

  B | None: The result of applying the function to the value, or None if the value is None.

Source code in algobase/functional.py

def maybe_apply(x: A | None, f: Callable[[A], B]) -> B | None:
    """Return the result of applying the function to the value if the value is not None, otherwise return None.

    Args:
        x (A | None): The value to apply the function to.
        f (Callable[[A], B | None]): The function to apply to the value.

    Returns:
        B | None: The result of applying the function to the value, or None if the value is None.
    """
    return f(x) if x is not None else None

provide_context

provide_context(**kwargs: Any) -> Callable[..., T]

A closure that provides context arguments to a function.

Parameters:

• **kwargs (Any, default: {} ) –

  Arbitrary keyword arguments.

Returns:

• Callable[..., T] –

  Callable[..., T]: The wrapped function.

Source code in algobase/functional.py

def provide_context(**kwargs: Any) -> Callable[..., T]:
    """A closure that provides context arguments to a function.

    Args:
        **kwargs: Arbitrary keyword arguments.

    Returns:
        Callable[..., T]: The wrapped function.
    """

    def wrapped(fn: Callable[P, T], *fn_args: P.args, **fn_kwargs: P.kwargs) -> T:
        """Calls the function with the context arguments and any additional arguments passed in.

        Args:
            fn (Callable[P, T]): The function to call.
            *fn_args: Variable length argument list.
            **fn_kwargs: Arbitrary keyword arguments.

        Returns:
            T: The result of calling the function.
        """
        inject = {
            k: v
            for k, v in kwargs.items()
            if k in fn.__code__.co_varnames[len(fn_args) :]
        }
        return fn(*fn_args, **{**inject, **fn_kwargs})

    return wrapped

algobase.settings

Configuration settings for the algobase.

Settings

Bases: BaseSettings

Pydantic model for algobase settings.

Source code in algobase/settings.py

class Settings(BaseSettings):
    """Pydantic model for algobase settings."""

    model_config = SettingsConfigDict(
        env_prefix="AB_", env_file=".env", env_file_encoding="utf-8"
    )

    algorand_network: AlgorandNetworkChoice = Field(
        description="The name of the Algorand network.",
        default=AlgorandNetwork.LOCALNET,
    )
    algorand_provider: AlgorandApiProviderChoice = Field(
        description="The Algorand API provider.", default=AlgorandApiProvider.LOCALHOST
    )
    algod_token: str = Field(description="The Algod API token.", default="a" * 64)
    nft_storage_api_key: str | None = Field(
        description="API key for nft.storage.", default=None
    )
    testnet_dispenser_access_token: str | None = Field(
        description="Access token for the Algorand TestNet dispenser.", default=None
    )

    def __or__(self, f: Callable[[Self], T]) -> T:
        """Operator overloading to pipe settings into a function or other callable.

        Args:
            f (Callable[[Self], T]): The function that takes `settings` as an argument.

        Returns:
            T: The type returned by the function.
        """
        return f(self)