What changed, and why it matters
This commit only adds Python type hints to function parameters and return values. It does not change any actual program logic, data handling, or security behavior. There is no security issue here.
No action required. This is a non-functional type-hinting change.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit is a pure type-annotation refactor across five files. It adds Callable signatures, parameter types, and return types to functions in coinchooser.py, lnpeer.py, psbt_nostr.py, transaction.py, and util.py. No runtime code paths, validation logic, or cryptographic operations are modified. The only import change is adding Any to typing imports in transaction.py.
Changed components
electrum/coinchooser.pyelectrum/lnpeer.pyelectrum/plugins/psbt_nostr/psbt_nostr.pyelectrum/transaction.pyelectrum/util.pyInspect captured patch +47 / −23
diff --git a/electrum/coinchooser.py b/electrum/coinchooser.py
index ae97d7d..843470f 100644
--- a/electrum/coinchooser.py
+++ b/electrum/coinchooser.py
@@ -90,7 +90,7 @@ class ScoredCandidate(NamedTuple):
buckets: List[Bucket]
-def strip_unneeded(bkts: List[Bucket], sufficient_funds) -> List[Bucket]:
+def strip_unneeded(bkts: List[Bucket], sufficient_funds: Callable) -> List[Bucket]:
'''Remove buckets that are unnecessary in achieving the spend amount'''
if sufficient_funds([], bucket_value_sum=0):
# none of the buckets are needed
@@ -113,7 +113,12 @@ class CoinChooserBase(Logger):
def keys(self, coins: Sequence[PartialTxInput]) -> Sequence[str]:
raise NotImplementedError
- def bucketize_coins(self, coins: Sequence[PartialTxInput], *, fee_estimator_vb):
+ def bucketize_coins(
+ self,
+ coins: Sequence[PartialTxInput],
+ *,
+ fee_estimator_vb: Callable[[int | float | Decimal], int],
+ ):
keys = self.keys(coins)
buckets = defaultdict(list) # type: Dict[str, List[PartialTxInput]]
for key, coin in zip(keys, coins):
@@ -151,9 +156,12 @@ class CoinChooserBase(Logger):
return list(map(make_Bucket, buckets.keys(), buckets.values()))
- def penalty_func(self, base_tx, *,
- tx_from_buckets: Callable[[List[Bucket]], Tuple[PartialTransaction, List[PartialTxOutput]]]) \
- -> Callable[[List[Bucket]], ScoredCandidate]:
+ def penalty_func(
+ self,
+ base_tx: Transaction,
+ *,
+ tx_from_buckets: Callable[[List[Bucket]], Tuple[PartialTransaction, List[PartialTxOutput]]],
+ ) -> Callable[[List[Bucket]], ScoredCandidate]:
raise NotImplementedError
def _change_amounts(self, tx: PartialTransaction, count: int, fee_estimator_numchange) -> List[int]:
@@ -282,7 +290,7 @@ class CoinChooserBase(Logger):
inputs: List[PartialTxInput],
outputs: List[PartialTxOutput],
change_addrs: Sequence[str],
- fee_estimator_vb: Callable,
+ fee_estimator_vb: Callable[[int | float | Decimal], int],
dust_threshold: int,
BIP69_sort: bool = True,
) -> PartialTransaction:
@@ -322,7 +330,7 @@ class CoinChooserBase(Logger):
def fee_estimator_w(weight):
return fee_estimator_vb(Transaction.virtual_size_from_weight(weight))
- def sufficient_funds(buckets, *, bucket_value_sum):
+ def sufficient_funds(buckets: List[Bucket], *, bucket_value_sum: int) -> bool:
'''Given a list of buckets, return True if it has enough
value to pay for the transaction'''
# assert bucket_value_sum == sum(bucket.value for bucket in buckets) # expensive!
@@ -373,7 +381,11 @@ class CoinChooserBase(Logger):
class CoinChooserRandom(CoinChooserBase):
- def bucket_candidates_any(self, buckets: List[Bucket], sufficient_funds) -> List[List[Bucket]]:
+ def bucket_candidates_any(
+ self,
+ buckets: List[Bucket],
+ sufficient_funds: Callable,
+ ) -> List[List[Bucket]]:
'''Returns a list of bucket sets.'''
if not buckets:
if sufficient_funds([], bucket_value_sum=0):
@@ -411,8 +423,11 @@ class CoinChooserRandom(CoinChooserBase):
candidates = [[buckets[n] for n in c] for c in candidates]
return [strip_unneeded(c, sufficient_funds) for c in candidates]
- def bucket_candidates_prefer_confirmed(self, buckets: List[Bucket],
- sufficient_funds) -> List[List[Bucket]]:
+ def bucket_candidates_prefer_confirmed(
+ self,
+ buckets: List[Bucket],
+ sufficient_funds: Callable,
+ ) -> List[List[Bucket]]:
"""Returns a list of bucket sets preferring confirmed coins.
Any bucket can be:
@@ -435,7 +450,7 @@ class CoinChooserRandom(CoinChooserBase):
for bkts_choose_from in bucket_sets:
try:
def sfunds(
- bkts, *, bucket_value_sum,
+ bkts: List[Bucket], *, bucket_value_sum: int,
already_selected_buckets_value_sum=already_selected_buckets_value_sum,
already_selected_buckets=already_selected_buckets,
):
diff --git a/electrum/lnpeer.py b/electrum/lnpeer.py
index cdea45f..3b516ec 100644
--- a/electrum/lnpeer.py
+++ b/electrum/lnpeer.py
@@ -2352,8 +2352,8 @@ class Peer(Logger, EventListener):
chan: Channel,
htlc: UpdateAddHtlc,
processed_onion: ProcessedOnionPacket,
- log_fail_reason: Callable,
- ):
+ log_fail_reason: Callable[[str], None],
+ ) -> tuple[bytes, int, int, OnionRoutingFailure]:
"""
Perform checks that are invariant (results do not depend on height, network conditions, etc).
May raise OnionRoutingFailure
@@ -2379,13 +2379,13 @@ class Peer(Logger, EventListener):
code=OnionFailureCode.FINAL_INCORRECT_CLTV_EXPIRY,
data=htlc.cltv_abs.to_bytes(4, byteorder="big"))
try:
- total_msat = processed_onion.hop_data.payload["payment_data"]["total_msat"]
+ total_msat = processed_onion.hop_data.payload["payment_data"]["total_msat"] # type: int
except Exception:
log_fail_reason(f"'total_msat' missing from onion")
raise exc_incorrect_or_unknown_pd
if chan.opening_fee:
- channel_opening_fee = chan.opening_fee['channel_opening_fee']
+ channel_opening_fee = chan.opening_fee['channel_opening_fee'] # type: int
total_msat -= channel_opening_fee
amt_to_forward -= channel_opening_fee
else:
@@ -2405,7 +2405,16 @@ class Peer(Logger, EventListener):
return payment_secret_from_onion, total_msat, channel_opening_fee, exc_incorrect_or_unknown_pd
- def check_mpp_is_waiting(self, *, payment_secret, short_channel_id, htlc, expected_msat, exc_incorrect_or_unknown_pd, log_fail_reason) -> bool:
+ def check_mpp_is_waiting(
+ self,
+ *,
+ payment_secret: bytes,
+ short_channel_id: ShortChannelID,
+ htlc: UpdateAddHtlc,
+ expected_msat: int,
+ exc_incorrect_or_unknown_pd: OnionRoutingFailure,
+ log_fail_reason: Callable[[str], None],
+ ) -> bool:
from .lnworker import RecvMPPResolution
mpp_resolution = self.lnworker.check_mpp_status(
payment_secret=payment_secret,
diff --git a/electrum/plugins/psbt_nostr/psbt_nostr.py b/electrum/plugins/psbt_nostr/psbt_nostr.py
index b084c0e..bcafa3d 100644
--- a/electrum/plugins/psbt_nostr/psbt_nostr.py
+++ b/electrum/plugins/psbt_nostr/psbt_nostr.py
@@ -275,8 +275,8 @@ class CosignerWallet(Logger):
tx: Union['Transaction', 'PartialTransaction'],
*,
label: str = None,
- on_failure: Callable = None,
- on_success: Callable = None
+ on_failure: Callable[[str], None] = None,
+ on_success: Callable[[], None] = None
) -> None:
try:
# TODO: adding tx should be handled more gracefully here:
diff --git a/electrum/transaction.py b/electrum/transaction.py
index 8bbae09..cdab347 100644
--- a/electrum/transaction.py
+++ b/electrum/transaction.py
@@ -29,7 +29,7 @@ import struct
import io
import base64
from typing import (
- Sequence, Union, NamedTuple, Tuple, Optional, Iterable, Callable, List, Dict, Set, TYPE_CHECKING, Mapping
+ Sequence, Union, NamedTuple, Tuple, Optional, Iterable, Callable, List, Dict, Set, TYPE_CHECKING, Mapping, Any
)
from collections import defaultdict
from enum import IntEnum
@@ -703,7 +703,7 @@ def script_GetOp(_bytes : bytes):
class OPPushDataGeneric:
- def __init__(self, pushlen: Callable=None):
+ def __init__(self, pushlen: Callable[[int], bool] | None = None):
if pushlen is not None:
self.check_data_len = pushlen
@@ -721,7 +721,7 @@ class OPPushDataGeneric:
class OPGeneric:
- def __init__(self, matcher: Callable = None):
+ def __init__(self, matcher: Callable[[Any], bool] | None = None):
if matcher is not None:
self.matcher = matcher
@@ -729,7 +729,7 @@ class OPGeneric:
return self.matcher(op)
@classmethod
- def is_instance(cls, item):
+ def is_instance(cls, item) -> bool:
# accept objects that are instances of this class
# or other classes that are subclasses
return isinstance(item, cls) \
diff --git a/electrum/util.py b/electrum/util.py
index 0105ef3..033283e 100644
--- a/electrum/util.py
+++ b/electrum/util.py
@@ -1742,7 +1742,7 @@ def _set_custom_task_factory(loop: asyncio.AbstractEventLoop):
loop.set_task_factory(factory)
-def run_sync_function_on_asyncio_thread(func: Callable, *, block: bool) -> None:
+def run_sync_function_on_asyncio_thread(func: Callable[[], Any], *, block: bool) -> None:
"""Run a non-async fn on the asyncio thread. Can be called from any thread.
If the current thread is already the asyncio thread, func is guaranteed
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.