-
Notifications
You must be signed in to change notification settings - Fork 48
Open
Description
1.18 introduced a TypeError in one of my libs that worked fine on 1.17
super(type, obj): obj must be an instance or subtype of type
While I haven't had a chance to minimize this failure case, the relevant source code is below. It's currently known to occur in a non-native class that inherits from a non-native dataclass, and uses super():
@mypyc_attr(native_class=False)
@dataclass(kw_only=True, frozen=True)
class _SortRule:
"""Base class for defining transaction matching rules.
When instantiated, a rule validates its inputs, determines which transaction
attributes to match (or uses a custom function), and registers itself
in the global `SORT_RULES` mapping under its class.
Matched transactions are assigned to the specified `txgroup`.
See Also:
:const:`dao_treasury.sorting.rule.SORT_RULES`
"""
txgroup: TxGroupName
"""Name of the transaction group to assign upon match."""
hash: Optional[HexStr] = None
"""Exact transaction hash to match."""
from_address: Optional[EthAddress] = None
"""Source wallet address to match."""
from_nickname: Optional[str] = None
"""Sender nickname (alias) to match."""
to_address: Optional[EthAddress] = None
"""Recipient wallet address to match."""
to_nickname: Optional[str] = None
"""Recipient nickname (alias) to match."""
token_address: Optional[EthAddress] = None
"""Token contract address to match."""
symbol: Optional[str] = None
"""Token symbol to match."""
log_index: Optional[int] = None
"""Log index within the transaction receipt to match."""
func: Optional[SortFunction] = None
"""Custom matching function that takes a `TreasuryTx` and returns a bool or an awaitable that returns a bool."""
# __instances__: ClassVar[List[Self]] = []
def __post_init__(self) -> None:
"""Validate inputs, checksum addresses, and register the rule.
- Ensures no duplicate rule exists for the same `txgroup`.
- Converts address fields to checksummed format.
- Determines which attributes will be used for direct matching.
- Validates that exactly one of attribute-based or function-based matching is provided.
- Registers the instance in :attr:`SORT_RULES` and :data:`_match_all`.
"""
if self.txgroup in _match_all:
raise ValueError(
f"there is already a matcher defined for txgroup {self.txgroup}: {self}"
)
# ensure addresses are checksummed if applicable
for attr in ["from_address", "to_address", "token_address"]:
value = getattr(self, attr)
if value is not None:
checksummed = EthAddress(value)
# NOTE: we must use object.__setattr__ to modify a frozen dataclass instance
object.__setattr__(self, attr, checksummed)
# define matchers used for this instance
matchers = [attr for attr in _MATCHING_ATTRS if getattr(self, attr) is not None]
_match_all[self.txgroup] = matchers
if self.func is not None and matchers:
raise ValueError(
"You must specify attributes for matching or pass in a custom matching function, not both."
)
if self.func is None and not matchers:
raise ValueError(
"You must specify attributes for matching or pass in a custom matching function."
)
if self.func is not None and not callable(self.func):
raise TypeError(f"func must be callable. You passed {self.func}")
# append new instance to instances classvar
# TODO: fix dataclass ClassVar handling in mypyc and reenable
# self.__instances__.append(self)
# append new instance under its class key
SORT_RULES[type(self)].append(self)
@property
def txgroup_dbid(self) -> TxGroupDbid:
"""Compute the database ID for this rule's `txgroup`.
Splits the `txgroup` string on ':' and resolves or creates the hierarchical
`TxGroup` entries in the database, returning the final group ID.
See Also:
:class:`~dao_treasury.db.TxGroup`.
"""
from dao_treasury.db import TxGroup
txgroup = None
for part in self.txgroup.split(":"):
txgroup = TxGroup.get_dbid(part, txgroup)
return txgroup
async def match(self, tx: "TreasuryTx") -> bool:
"""Determine if the given transaction matches this rule.
Args:
tx: A `TreasuryTx` entity to test against this rule.
Returns:
True if the transaction matches the rule criteria; otherwise False.
Examples:
# match by symbol and recipient
>>> rule = _SortRule(txgroup='Foo', symbol='DAI', to_address='0xabc...')
>>> await rule.match(tx) # where tx.symbol == 'DAI' and tx.to_address == '0xabc...'
True
See Also:
:attr:`_match_all`
"""
if matchers := _match_all[self.txgroup]:
return all(
getattr(tx, matcher) == getattr(self, matcher) for matcher in matchers
)
_log_debug("checking %s for %s", tx, self.func)
match = self.func(tx) # type: ignore [misc]
return match if isinstance(match, bool) else await match
@mypyc_attr(native_class=False)
class _InboundSortRule(_SortRule):
"""Sort rule that applies only to inbound transactions (to the DAO's wallet).
Checks that the transaction's `to_address` belongs to a known `TreasuryWallet`
before applying the base matching logic.
"""
async def match(self, tx: "TreasuryTx") -> bool:
return (
tx.to_address is not None
and TreasuryWallet.check_membership(tx.to_address.address, tx.block)
and await super().match(tx)
)Metadata
Metadata
Assignees
Labels
No labels