from abc import ABC, abstractmethod
from collections.abc import Sequence
from typing import Any, assert_never, override
from pydantic import TypeAdapter, ValidationError
from typing_extensions import TypeVar
from erc7730.common.output import OutputAdder
from erc7730.common.properties import get_property
from erc7730.model.input.path import ContainerPathStr, DataPathStr
from erc7730.model.input.v2.descriptor import InputERC7730Descriptor
from erc7730.model.input.v2.display import InputMapReference
from erc7730.model.paths import ROOT_DESCRIPTOR_PATH, ArrayElement, ContainerPath, DataPath, DescriptorPath, Field
from erc7730.model.paths.path_ops import descriptor_path_append, to_absolute
from erc7730.model.types import MixedCaseAddress
_T = TypeVar("_T", covariant=True)
[docs]
class ConstantProvider(ABC):
"""
Resolver for constants values referenced by descriptor paths.
"""
[docs]
@abstractmethod
def get(self, path: DescriptorPath, out: OutputAdder) -> Any:
"""
Get the constant for the given path.
:param path: descriptor path
:param out: error handler
:return: constant value, or None if not found
"""
raise NotImplementedError()
[docs]
@abstractmethod
def resolve_map_reference(self, prefix: DataPath, map_ref: InputMapReference, out: OutputAdder) -> Any:
"""
Resolve a map reference to its value.
:param prefix: current path prefix
:param map_ref: map reference to resolve
:param out: error handler
:return: resolved value from map, or None if not found
"""
raise NotImplementedError()
[docs]
def resolve(self, value: _T | DescriptorPath, out: OutputAdder) -> _T:
"""
Resolve the value if it is a descriptor path.
:param value: descriptor path or actual value
:param out: error handler
:return: constant value, or the value itself if not a descriptor path
"""
return self.get(value, out) if isinstance(value, DescriptorPath) else value
[docs]
def resolve_or_none(self, value: _T | DescriptorPath | None, out: OutputAdder) -> _T | None:
"""
Resolve the optional value if it is a descriptor path.
:param value: descriptor path, actual value or None
:param out: error handler
:return: None, constant value, or the value itself if not a descriptor path
"""
return None if value is None else self.resolve(value, out)
[docs]
def resolve_path(
self, value: DataPath | ContainerPath | DescriptorPath, out: OutputAdder
) -> DataPath | ContainerPath | None:
"""
Resolve the value as a data/container path.
:param value: descriptor path or actual data/container path
:param out: error handler
:return: resolved data/container path
"""
def assert_not_address(path: DataPath | ContainerPath) -> bool:
match path:
case ContainerPath():
return True
case DataPath():
if path.absolute:
return True
try:
TypeAdapter(MixedCaseAddress).validate_strings(str(path))
out.error(
title="Invalid data path",
message=f""""{path}" is invalid, it must contain a data path to the address in the """
"transaction data. It seems you are trying to use a constant address value instead, please "
"use the adequate parameter to provide a constant value.",
)
return False
except ValidationError:
return True
case _:
assert_never(path)
if isinstance(value, DataPath | ContainerPath):
if not assert_not_address(value):
return None
return value
resolved_value: Any
if (resolved_value := self.resolve(value, out)) is None:
return None
if not isinstance(resolved_value, str):
return out.error(
title="Invalid constant path",
message=f"Constant path defined at {value} must be a path string, got {type(resolved_value).__name__}.",
)
match TypeAdapter(DataPathStr | ContainerPathStr).validate_strings(resolved_value):
case ContainerPath() as path:
return path
case DataPath() as path:
if not assert_not_address(path):
return None
if not path.absolute:
return out.error(
title="Invalid data path constant",
message=f"Data path defined at {value} must be absolute, please change it to "
f"{to_absolute(path)}.",
)
return path
case _:
assert_never(resolved_value)
# noinspection PyUnreachableCode
return None
[docs]
def resolve_path_or_none(
self, value: DataPath | ContainerPath | DescriptorPath | None, out: OutputAdder
) -> DataPath | ContainerPath | None:
"""
Resolve the value as a data/container path.
:param value: descriptor path or actual data/container path
:param out: error handler
:return: resolved data/container path
"""
return None if value is None else self.resolve_path(value, out)
[docs]
class DefaultConstantProvider(ConstantProvider):
"""
Resolver for constants values from a provided v2 descriptor.
"""
def __init__(self, descriptor: InputERC7730Descriptor) -> None:
self.descriptor: InputERC7730Descriptor = descriptor
[docs]
@override
def get(self, path: DescriptorPath, out: OutputAdder) -> Any:
current_target = self.descriptor
parent_path = ROOT_DESCRIPTOR_PATH
current_path = ROOT_DESCRIPTOR_PATH
for element in path.elements:
current_path = descriptor_path_append(current_path, element)
match element:
case Field(identifier=field):
if isinstance(current_target, Sequence):
return out.error(
title="Invalid constant path",
message=f"""Path {current_path} is invalid, {parent_path} is an array.""",
)
else:
try:
current_target = get_property(current_target, field)
except (AttributeError, KeyError):
return out.error(
title="Invalid constant path",
message=f"""Path {current_path} is invalid, {parent_path} has no "{field}" field.""",
)
case ArrayElement(index=i):
if not isinstance(current_target, Sequence):
return out.error(
title="Invalid constant path",
message=f"Path {current_path} is invalid, {parent_path} is not an array.",
)
if i >= len(current_target):
return out.error(
title="Invalid constant path",
message=f"""Path {current_path} is invalid, index {i} is out of bounds.""",
)
current_target = current_target[i]
case _:
assert_never(element)
parent_path = descriptor_path_append(parent_path, element)
return current_target
[docs]
@override
def resolve_map_reference(self, prefix: DataPath, map_ref: InputMapReference, out: OutputAdder) -> Any:
"""
Resolve a map reference to its value by looking up the map and resolving the keyPath.
:param prefix: current path prefix
:param map_ref: map reference with map descriptor path and keyPath
:param out: error handler
:return: resolved value from map, or None if not found
"""
# Get the map definition
if (map_def := self.get(map_ref.map, out)) is None:
return out.error(
title="Invalid map reference",
message=f"Map at {map_ref.map} does not exist.",
)
# Ensure map has the expected structure
if not hasattr(map_def, "values") or not isinstance(map_def.values, dict):
return out.error(
title="Invalid map reference",
message=f"Map at {map_ref.map} is not a valid map definition.",
)
# Resolve the key path to get the key value
if (self.resolve_path(map_ref.keyPath, out)) is None:
return None
# For map references, the key path is either a DataPath or ContainerPath
# We can't actually resolve the runtime value here during conversion,
# so we store the path for runtime resolution. However, for constant validation
# we could check if it's a constant path.
# Since this is input-to-resolved conversion, we pass through the structure.
# The actual key lookup happens at display time, not conversion time.
# Return the map reference as-is for the resolved model to handle at runtime
return map_ref
