Merge pull request #10903 from SomberNight/202608_fix_type_hints_optional
What changed, and why it matters
This commit only updates Python type hints to explicitly mark parameters and variables that can be None. It does not change any program logic, runtime behavior, or security checks. It is a code-quality and documentation improvement with no security impact.
No action required. This is a non-functional type-hint cleanup commit.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The merge commit adds missing ‘Optional’ (or ‘| None’) qualifiers across 51 files. The changes are purely annotations: default values remain the same, function bodies are unchanged, and no runtime behavior is altered. One minor non-annotation change in lnsweep.py removes a debug f-string from a keyword argument, and test_commands.py changes a boolean flag to an integer counter, but these are also not security-relevant. No cryptographic, network, or wallet logic was modified.
Changed components
electrum/address_synchronizer.pyelectrum/bip32.pyelectrum/bolt11.pyelectrum/channel_db.pyelectrum/crypto.pyelectrum/exchange_rate.pyelectrum/gui/qml/qebiometrics.pyelectrum/gui/qml/qebip39recovery.pyelectrum/gui/qml/qetxfinalizer.pyelectrum/gui/qml/qetypes.pyelectrum/gui/qml/qewallet.pyelectrum/gui/qt/__init__.pyelectrum/gui/qt/address_list.pyelectrum/gui/qt/main_window.pyelectrum/gui/qt/my_treeview.pyelectrum/gui/qt/qrreader/qtmultimedia/camera_dialog.pyelectrum/gui/qt/qrreader/qtmultimedia/validator.pyelectrum/gui/qt/util.pyelectrum/gui/qt/wizard/wallet.pyelectrum/gui/qt/wizard/wizard.pyelectrum/hw_wallet/plugin.pyelectrum/interface.pyelectrum/invoices.pyelectrum/keystore.pyelectrum/lnchannel.pyelectrum/lnhtlc.pyelectrum/lnonion.pyelectrum/lnpeer.pyelectrum/lnrouter.pyelectrum/lnsweep.pyelectrum/lnutil.pyelectrum/lnworker.pyelectrum/mnemonic.pyelectrum/network.pyelectrum/onion_message.pyelectrum/payment_identifier.pyelectrum/plugin.pyelectrum/plugins/nwc/nwcserver.pyelectrum/plugins/psbt_nostr/psbt_nostr.pyelectrum/plugins/psbt_nostr/qml.pyelectrum/plugins/watchtower/watchtower.pyelectrum/submarine_swaps.pyelectrum/trampoline.pyelectrum/transaction.pyelectrum/wallet.pytests/lnhelpers.pytests/test_commands.pytests/test_lnpeer.pytests/test_submarine_swaps.pytests/toyserver/toynetwork.pytests/toyserver/toyserver.pyInspect captured patch +144 / −142
### electrum/address_synchronizer.py
@@ -82,7 +82,7 @@ class AddressSynchronizer(Logger, EventListener):
synchronizer: Optional['Synchronizer']
verifier: Optional['SPV']
- def __init__(self, db: 'WalletDB', config: 'SimpleConfig', *, name: str = None):
+ def __init__(self, db: 'WalletDB', config: 'SimpleConfig', *, name: str | None = None):
self.db = db
self.config = config
self.name = name
@@ -172,7 +172,7 @@ def get_txin_address(self, txin: TxInput) -> Optional[str]:
return None
@with_lock
- def get_txin_value(self, txin: TxInput, *, address: str = None) -> Optional[int]:
+ def get_txin_value(self, txin: TxInput, *, address: str | None = None) -> Optional[int]:
if txin.value_sats() is not None:
return txin.value_sats()
prevout_hash = txin.prevout.txid.hex()
@@ -524,7 +524,7 @@ def _get_tx_sort_key(self, tx_hash: str) -> Tuple[int, int]:
return height, txpos
@classmethod
- def tx_height_to_sort_height(cls, height: int = None):
+ def tx_height_to_sort_height(cls, height: int | None = None):
"""Return a height-like value to be used for sorting txs."""
if height is not None:
if height > 0:
@@ -970,7 +970,7 @@ def get_utxos(
confirmed_funding_only: bool = False,
confirmed_spending_only: bool = False,
nonlocal_only: bool = False,
- block_height: int = None,
+ block_height: int | None = None,
) -> Sequence[PartialTxInput]:
if block_height is not None:
# caller wants the UTXOs we had at a given height; check other parameters
### electrum/bip32.py
@@ -429,9 +429,12 @@ def root_fp_and_der_prefix_from_xkey(xkey: str) -> Tuple[Optional[str], Optional
return root_fingerprint, derivation_prefix
-def is_xkey_consistent_with_key_origin_info(xkey: str, *,
- derivation_prefix: str = None,
- root_fingerprint: str = None) -> bool:
+def is_xkey_consistent_with_key_origin_info(
+ xkey: str,
+ *,
+ derivation_prefix: str | None = None,
+ root_fingerprint: str | None = None,
+) -> bool:
bip32node = BIP32Node.from_xkey(xkey)
int_path = None
if derivation_prefix is not None:
### electrum/bolt11.py
@@ -118,7 +118,7 @@ def tagged8(char: str, data8: Sequence[int]) -> Sequence[int]:
return tagged5(char, convertbits(data8, 8, 5))
-def int_to_data5(val: int, *, bit_len: int = None) -> Sequence[int]:
+def int_to_data5(val: int, *, bit_len: int | None = None) -> Sequence[int]:
"""Represent big-endian number with as many 0-31 values as it takes.
If `bit_len` is set, use exactly bit_len//5 values (left-padded with zeroes).
"""
### electrum/channel_db.py
@@ -495,7 +495,7 @@ def add_channel_announcements(self, msg_payloads, *, trusted=True):
self.update_counts()
- def add_verified_channel_info(self, msg: dict, *, capacity_sat: int = None) -> None:
+ def add_verified_channel_info(self, msg: dict, *, capacity_sat: int | None = None) -> None:
try:
channel_info = ChannelInfo.from_msg(msg)
except IncompatibleOrInsaneFeatures:
@@ -664,7 +664,7 @@ def _db_save_node_addresses(self, node_addresses: Sequence[LNPeerAddr]):
c.execute("INSERT INTO address (node_id, host, port, timestamp) VALUES (?,?,?,?)", (addr.pubkey, addr.host, addr.port, 0))
@classmethod
- def verify_channel_update(cls, payload, *, start_node: bytes = None) -> None:
+ def verify_channel_update(cls, payload, *, start_node: bytes | None = None) -> None:
short_channel_id = payload['short_channel_id']
short_channel_id = ShortChannelID(short_channel_id)
if constants.net.rev_genesis_bytes() != payload['chain_hash']:
@@ -760,7 +760,7 @@ def _get_channel_update_for_private_channel(
start_node_id: bytes,
short_channel_id: ShortChannelID,
*,
- now: int = None, # unix ts
+ now: int | None = None, # unix ts
) -> Optional[dict]:
if now is None:
now = int(time.time())
@@ -776,7 +776,7 @@ def add_channel_update_for_private_channel(
msg_payload: dict,
start_node_id: bytes,
*,
- cache_ttl: int = None, # seconds
+ cache_ttl: int | None = None, # seconds
) -> bool:
"""Returns True iff the channel update was successfully added and it was different than
what we had before (if any).
@@ -931,7 +931,7 @@ def get_policy_for_node(
*,
my_channels: Dict[ShortChannelID, 'Channel'] = None,
private_route_edges: Dict[ShortChannelID, 'RouteEdge'] = None,
- now: int = None, # unix ts
+ now: int | None = None, # unix ts
) -> Optional['Policy']:
channel_info = self.get_channel_info(short_channel_id)
if channel_info is not None: # publicly announced channel
### electrum/crypto.py
@@ -362,7 +362,7 @@ def chacha20_poly1305_encrypt(
*,
key: bytes,
nonce: bytes,
- associated_data: bytes = None,
+ associated_data: bytes | None = None,
data: bytes
) -> bytes:
assert isinstance(key, (bytes, bytearray))
@@ -387,7 +387,7 @@ def chacha20_poly1305_decrypt(
*,
key: bytes,
nonce: bytes,
- associated_data: bytes = None,
+ associated_data: bytes | None = None,
data: bytes
) -> bytes:
assert isinstance(key, (bytes, bytearray))
### electrum/exchange_rate.py
@@ -749,14 +749,14 @@ def exchange_rate(self) -> Decimal:
return Decimal('NaN')
return self.exchange.get_cached_spot_quote(self.ccy)
- def format_amount(self, btc_balance, *, timestamp: int = None) -> str:
+ def format_amount(self, btc_balance, *, timestamp: int | None = None) -> str:
if timestamp is None:
rate = self.exchange_rate()
else:
rate = self.timestamp_rate(timestamp)
return '' if rate.is_nan() else "%s" % self.value_str(btc_balance, rate)
- def format_amount_and_units(self, btc_balance, *, timestamp: int = None) -> str:
+ def format_amount_and_units(self, btc_balance, *, timestamp: int | None = None) -> str:
if timestamp is None:
rate = self.exchange_rate()
else:
### electrum/gui/qml/qebiometrics.py
@@ -117,7 +117,7 @@ def _disable_protected_failed(self):
@pyqtSlot()
@pyqtSlot(str)
- def unlock(self, auth_message: str = None):
+ def unlock(self, auth_message: str | None = None):
"""
Called when the user needs to authenticate.
Makes the AndroidKeyStore decrypt our encrypted wrap key, we then use the decrypted wrap key
@@ -128,7 +128,7 @@ def unlock(self, auth_message: str = None):
assert encrypted_wrap_key, "shouldn't unlock if biometric auth is disabled"
self._start_activity(BiometricAction.DECRYPT, data=encrypted_wrap_key, auth_message=auth_message)
- def _start_activity(self, action: BiometricAction, data: str, auth_message: str = None):
+ def _start_activity(self, action: BiometricAction, data: str, auth_message: str | None = None):
assert self._current_action is None, f"don't run concurrent activities: {self._current_action=} {action=}"
self._current_action = action
### electrum/gui/qml/qebip39recovery.py
@@ -71,7 +71,7 @@ def state(self, state: State):
@pyqtSlot(str, str)
@pyqtSlot(str, str, str)
- def startScan(self, wallet_type: str, seed: str, seed_extra_words: str = None):
+ def startScan(self, wallet_type: str, seed: str, seed_extra_words: str | None = None):
if not seed or not wallet_type:
return
### electrum/gui/qml/qetxfinalizer.py
@@ -585,7 +585,7 @@ def on_signed_tx(self, save: bool, tx: Transaction):
self._logger.error('Could not save tx')
self.finished.emit(True, saved, tx.is_complete())
- def on_sign_failed(self, msg: str = None):
+ def on_sign_failed(self, msg: str | None = None):
self._logger.debug('on_sign_failed')
self.signError.emit(msg)
@@ -941,7 +941,7 @@ def __init__(self, parent=None):
self._parent_tx = None
self._new_tx = None
self._parent_tx_size = 0
- self._parent_fee = 0
+ self._parent_fee = 0 # type: int | None
self._max_fee = 0
self._txid = ''
self._rbf = True
### electrum/gui/qml/qetypes.py
@@ -120,7 +120,7 @@ def __repr__(self):
class QEBytes(QObject):
- def __init__(self, data: bytes = None, *, parent=None):
+ def __init__(self, data: bytes | None = None, *, parent=None):
super().__init__(parent)
self.data = data
### electrum/gui/qml/qewallet.py
@@ -596,7 +596,7 @@ def on_sign_complete(self, broadcast, cb: Callable[[Transaction], None] = None,
self.broadcast(tx)
# this assumes a 2fa wallet, but there are no other tc_sign_wrapper hooks, so that's ok
- def on_sign_failed(self, cb: Callable[[], None] = None, error: str = None):
+ def on_sign_failed(self, cb: Callable[[], None] | None = None, error: str | None = None):
self.otpFailed.emit('error', error)
if cb:
cb()
@@ -658,7 +658,7 @@ def ln_auth_rejected(self):
self.paymentAuthRejected.emit()
@auth_protect(message=_('Pay lightning invoice?'), reject='ln_auth_rejected')
- def pay_lightning_invoice(self, invoice: 'Invoice', amount_msat: int = None):
+ def pay_lightning_invoice(self, invoice: 'Invoice', amount_msat: int | None = None):
# at this point, the user confirmed the payment, potentially with an override amount.
# we save the invoice with the override amount if there was no amount defined in the invoice.
# (this is similar to what the desktop client does)
### electrum/gui/qt/__init__.py
@@ -612,7 +612,7 @@ def version_info(cls):
ret["pyqt.path"] = ", ".join(PyQt6.__path__ or [])
return ret
- def do_copy(self, text: str, *, title: str = None) -> None:
+ def do_copy(self, text: str, *, title: str | None = None) -> None:
self.app.clipboard().setText(text)
message = _("Text copied to Clipboard") if title is None else _("{} copied to Clipboard").format(title)
# tooltip cannot be displayed immediately when called from a menu; wait 200ms
### electrum/gui/qt/address_list.py
@@ -350,7 +350,7 @@ def create_menu(self, position):
run_hook('receive_menu', menu, addrs, self.wallet)
self.open_menu(menu, position)
- def place_text_on_clipboard(self, text: str, *, title: str = None) -> None:
+ def place_text_on_clipboard(self, text: str, *, title: str | None = None) -> None:
if is_address(text):
try:
self.wallet.check_address_for_corruption(text)
### electrum/gui/qt/main_window.py
@@ -184,7 +184,7 @@ def __init__(self, gui_object: 'ElectrumGui', wallet: Abstract_Wallet):
Exception_Hook.maybe_setup(config=self.config, wallet=self.wallet)
- self.network = gui_object.daemon.network # type: Network
+ self.network = gui_object.daemon.network # type: Network | None
self.fx = gui_object.daemon.fx # type: FxThread
self.contacts = wallet.contacts
self.tray = gui_object.tray
@@ -973,7 +973,7 @@ def format_amount(
add_thousands_sep=add_thousands_sep,
)
- def format_amount_and_units(self, amount_sat, *, timestamp: int = None) -> str:
+ def format_amount_and_units(self, amount_sat, *, timestamp: int | None = None) -> str:
"""Returns string with both bitcoin and fiat amounts, in desired units.
E.g. 500_000 -> '0.005 BTC (191.42 EUR)'
"""
@@ -1205,7 +1205,7 @@ def create_receive_tab(self):
from .receive_tab import ReceiveTab
return ReceiveTab(self)
- def do_copy(self, text: str, *, title: str = None) -> None:
+ def do_copy(self, text: str, *, title: str | None = None) -> None:
self.gui_object.do_copy(text, title=title)
def show_tooltip_after_delay(self, message):
@@ -2404,7 +2404,7 @@ def do_process_from_file(self):
if tx:
self.show_transaction(tx)
- def do_process_from_txid(self, *, parent: QWidget = None, txid: str = None):
+ def do_process_from_txid(self, *, parent: QWidget = None, txid: str | None = None):
if parent is None:
parent = self
from electrum import transaction
### electrum/gui/qt/my_treeview.py
@@ -481,7 +481,7 @@ def add_copy_menu(self, menu: QMenu, idx) -> QMenu:
self.place_text_on_clipboard(text, title=title))
return cc
- def place_text_on_clipboard(self, text: str, *, title: str = None) -> None:
+ def place_text_on_clipboard(self, text: str, *, title: str | None = None) -> None:
self.main_window.do_copy(text, title=title)
def showEvent(self, e: 'QShowEvent'):
### electrum/gui/qt/qrreader/qtmultimedia/camera_dialog.py
@@ -87,7 +87,7 @@ def __init__(self, parent: Optional[QWidget], *, config: SimpleConfig):
self.last_qr_scan_ts: float = 0.0
self.camera: QCamera = None
self.media_capture_session: QMediaCaptureSession = None
- self._error_message: str = None
+ self._error_message: str | None = None
self._ok_done: bool = False
self.camera_sc_conn = None
self.resolution: QSize = None
### electrum/gui/qt/qrreader/qtmultimedia/validator.py
@@ -43,10 +43,10 @@ class QrReaderValidatorResult():
def __init__(self):
self.accepted: bool = False
- self.message: str = None
- self.message_color: QColor = None
+ self.message: str | None = None
+ self.message_color: QColor | None = None
- self.simple_result : str = None
+ self.simple_result : str | None = None
self.result_usable: Dict[QrCodeResult, bool] = {}
self.result_colors: Dict[QrCodeResult, QColor] = {}
### electrum/gui/qt/util.py
@@ -146,7 +146,7 @@ def setVisible(self, visible):
class HelpMixin:
- def __init__(self, help_text: str, *, help_title: str = None):
+ def __init__(self, help_text: str, *, help_title: str | None = None):
assert isinstance(self, QWidget), "HelpMixin must be a QWidget instance!"
self.help_text = help_text
self._help_title = help_title or _('Help')
@@ -641,7 +641,7 @@ def __init__(self):
self.setLineWidth(1)
-def address_field(addresses, *, btn_text: str = None):
+def address_field(addresses, *, btn_text: str | None = None):
if btn_text is None:
btn_text = _('Get wallet address')
hbox = QHBoxLayout()
@@ -1253,8 +1253,8 @@ def getSaveFileName(
title,
filename,
filter="",
- default_extension: str = None,
- default_filter: str = None,
+ default_extension: str | None = None,
+ default_filter: str | None = None,
config: 'SimpleConfig',
) -> Optional[str]:
"""Custom wrapper for getSaveFileName that remembers the path selected by the user."""
### electrum/gui/qt/wizard/wallet.py
@@ -145,7 +145,7 @@ def is_single_password(self):
# not supported on desktop
return False
- def create_storage(self, single_password: str = None):
+ def create_storage(self, single_password: str | None = None):
self._logger.info('Creating wallet from wizard data')
data = self.get_wizard_data()
@@ -1019,7 +1019,7 @@ def apply(self):
class SeedExtensionEdit(QWidget):
- def __init__(self, parent, *, message: str = None, warning: str = None, warn_issue4566: bool = False):
+ def __init__(self, parent, *, message: str | None = None, warning: str | None = None, warn_issue4566: bool = False):
super().__init__(parent)
self.warn_issue4566 = warn_issue4566
### electrum/gui/qt/wizard/wizard.py
@@ -261,7 +261,7 @@ def is_finalized(self, wizard_data: dict) -> bool:
class WizardComponent(AbstractQWidget):
updated = pyqtSignal(object)
- def __init__(self, parent: QWidget, wizard: QEAbstractWizard, *, title: str = None, layout: QLayout = None):
+ def __init__(self, parent: QWidget, wizard: QEAbstractWizard, *, title: str | None = None, layout: QLayout | None = None):
super().__init__(parent)
self.setLayout(layout if layout else QVBoxLayout(self))
self.wizard_data = {}
### electrum/hw_wallet/plugin.py
@@ -363,7 +363,7 @@ def trezor_validate_op_return_output_and_get_data(output: TxOutput) -> bytes:
return script[2:]
-def validate_op_return_output(output: TxOutput, *, max_size: int = None) -> None:
+def validate_op_return_output(output: TxOutput, *, max_size: int | None = None) -> None:
script = output.scriptpubkey
if script[0] != opcodes.OP_RETURN:
raise UserFacingException(_("Only OP_RETURN scripts are supported."))
### electrum/interface.py
@@ -268,7 +268,7 @@ def default_framer(self):
assert max_size > 500_000, f"{max_size=} (< 500_000) is too small"
return NewlineFramer(max_size=max_size)
- async def close(self, *, force_after: int = None):
+ async def close(self, *, force_after: int | None = None):
"""Closes the connection and waits for it to be closed.
We try to flush buffered data to the wire, which can take some time.
"""
@@ -498,7 +498,7 @@ async def close(self, *args, **kwargs):
class ServerAddr:
- def __init__(self, host: str, port: Union[int, str], *, protocol: str = None):
+ def __init__(self, host: str, port: Union[int, str], *, protocol: str | None = None):
assert isinstance(host, str), repr(host)
if protocol is None:
protocol = 's'
@@ -1130,7 +1130,7 @@ async def request_fee_estimates(self):
self.network.update_fee_estimates()
await asyncio.sleep(60)
- async def close(self, *, force_after: int = None):
+ async def close(self, *, force_after: int | None = None):
"""Closes the connection and waits for it to be closed.
We try to flush buffered data to the wire, which can take some time.
"""
### electrum/invoices.py
@@ -257,7 +257,7 @@ def as_dict(self, status):
@attr.s
class Invoice(BaseInvoice):
lightning_invoice = attr.ib(type=str, kw_only=True) # type: Optional[str]
- __lnaddr = None
+ __lnaddr = None # type: BOLT11Addr | None
_broadcasting_status = None # can be None or PR_BROADCASTING or PR_BROADCAST
def is_lightning(self):
### electrum/keystore.py
@@ -522,7 +522,7 @@ def test_der_suffix_against_pubkey(der_suffix: Sequence[int], pubkey: bytes) ->
class Xpub(MasterPublicKeyMixin):
- def __init__(self, *, derivation_prefix: str = None, root_fingerprint: str = None):
+ def __init__(self, *, derivation_prefix: str | None = None, root_fingerprint: str | None = None):
MasterPublicKeyMixin.__init__(self)
self.xpub = None
self.xpub_receive = None
@@ -616,7 +616,7 @@ def add_key_origin_from_root_node(self, *, derivation_prefix: str, root_node: BI
self.add_key_origin(derivation_prefix=derivation_prefix,
root_fingerprint=root_node.calc_fingerprint_of_this_node().hex().lower())
- def add_key_origin(self, *, derivation_prefix: str = None, root_fingerprint: str = None) -> None:
+ def add_key_origin(self, *, derivation_prefix: str | None = None, root_fingerprint: str | None = None) -> None:
assert self.xpub
if not (root_fingerprint is None or (is_hex_str(root_fingerprint) and len(root_fingerprint) == 8)):
raise Exception("root fp must be 8 hex characters")
@@ -851,7 +851,7 @@ def get_private_key(self, sequence: Sequence[int], password):
pk = self._get_private_key_from_stretched_exponent(for_change, n, secexp)
return pk, False
- def _check_seed(self, hex_seed: str, *, secexp: int = None) -> None:
+ def _check_seed(self, hex_seed: str, *, secexp: int | None = None) -> None:
if secexp is None:
secexp = self.stretch_key(hex_seed)
master_private_key = ecc.ECPrivkey.from_secret_scalar(secexp)
### electrum/lnchannel.py
@@ -500,15 +500,15 @@ def get_oldest_unrevoked_ctn(self, subject: HTLCOwner) -> int:
pass
@abstractmethod
- def included_htlcs(self, subject: HTLCOwner, direction: Direction, ctn: int = None) -> Sequence[UpdateAddHtlc]:
+ def included_htlcs(self, subject: HTLCOwner, direction: Direction, ctn: int | None = None) -> Sequence[UpdateAddHtlc]:
pass
@abstractmethod
def funding_txn_minimum_depth(self) -> int:
pass
@abstractmethod
- def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int = None) -> int:
+ def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int | None = None) -> int:
"""This balance (in msat) only considers HTLCs that have been settled by ctn.
It disregards reserve, fees, and pending HTLCs (in both directions).
"""
@@ -517,7 +517,7 @@ def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int = Non
@abstractmethod
def balance_minus_outgoing_htlcs(self, whose: HTLCOwner, *,
ctx_owner: HTLCOwner = HTLCOwner.LOCAL,
- ctn: int = None) -> int:
+ ctn: int | None = None) -> int:
"""This balance (in msat), which includes the value of
pending outgoing HTLCs, is used in the UI.
"""
@@ -710,10 +710,10 @@ def funding_txn_minimum_depth(self):
def is_funding_tx_mined(self, funding_height):
return funding_height.conf > 1
- def balance_minus_outgoing_htlcs(self, whose: HTLCOwner, *, ctx_owner: HTLCOwner = HTLCOwner.LOCAL, ctn: int = None):
+ def balance_minus_outgoing_htlcs(self, whose: HTLCOwner, *, ctx_owner: HTLCOwner = HTLCOwner.LOCAL, ctn: int | None = None):
return 0
- def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int = None) -> int:
+ def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int | None = None) -> int:
return 0
def is_frozen_for_sending(self) -> bool:
@@ -1240,7 +1240,7 @@ def add_htlc(self, htlc: UpdateAddHtlc) -> UpdateAddHtlc:
self.logger.info("add_htlc")
return htlc
- def receive_htlc(self, htlc: UpdateAddHtlc, onion_packet:bytes = None) -> UpdateAddHtlc:
+ def receive_htlc(self, htlc: UpdateAddHtlc, onion_packet: bytes | None = None) -> UpdateAddHtlc:
"""Adds a new REMOTE HTLC to the channel.
Action must be initiated by REMOTE.
"""
@@ -1528,7 +1528,7 @@ def extract_preimage_from_htlc_txin(self, txin: TxInput, *, is_deeply_mined: boo
error_bytes=None,
failure_message=failure)
- def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int = None) -> int:
+ def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int | None = None) -> int:
assert type(whose) is HTLCOwner
initial = self.config[whose].initial_msat
return self.hm.get_balance_msat(whose=whose,
@@ -1537,7 +1537,7 @@ def balance(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int = Non
initial_balance_msat=initial)
def balance_minus_outgoing_htlcs(self, whose: HTLCOwner, *, ctx_owner: HTLCOwner = HTLCOwner.LOCAL,
- ctn: int = None) -> int:
+ ctn: int | None = None) -> int:
assert type(whose) is HTLCOwner
if ctn is None:
ctn = self.get_next_ctn(ctx_owner)
@@ -1546,7 +1546,7 @@ def balance_minus_outgoing_htlcs(self, whose: HTLCOwner, *, ctx_owner: HTLCOwner
balance_in_htlcs = self.balance_tied_up_in_htlcs_by_direction(ctx_owner, ctn=ctn, direction=direction)
return committed_balance - balance_in_htlcs
- def balance_tied_up_in_htlcs_by_direction(self, ctx_owner: HTLCOwner = LOCAL, *, ctn: int = None,
+ def balance_tied_up_in_htlcs_by_direction(self, ctx_owner: HTLCOwner = LOCAL, *, ctn: int | None = None,
direction: Direction):
# in msat
if ctn is None:
@@ -1641,8 +1641,8 @@ def consider_ctx(*, ctx_owner: HTLCOwner, is_htlc_dust: bool) -> int:
return max_send_msat
- def included_htlcs(self, subject: HTLCOwner, direction: Direction, ctn: int = None, *,
- feerate: int = None) -> List[UpdateAddHtlc]:
+ def included_htlcs(self, subject: HTLCOwner, direction: Direction, ctn: int | None = None, *,
+ feerate: int | None = None) -> List[UpdateAddHtlc]:
"""Returns list of non-dust HTLCs for subject's commitment tx at ctn,
filtered by direction (of HTLCs).
"""
### electrum/lnhtlc.py
@@ -407,7 +407,7 @@ def _is_htlc_irrevocably_removed_yet(
@with_lock
def htlcs_by_direction(self, subject: HTLCOwner, direction: Direction,
- ctn: int = None) -> Dict[int, UpdateAddHtlc]:
+ ctn: int | None = None) -> Dict[int, UpdateAddHtlc]:
"""Return the dict of received or sent (depending on direction) HTLCs
in subject's ctx at ctn, keyed by htlc_id.
@@ -432,7 +432,7 @@ def htlcs_by_direction(self, subject: HTLCOwner, direction: Direction,
return d
@with_lock
- def htlcs(self, subject: HTLCOwner, ctn: int = None) -> Sequence[Tuple[Direction, UpdateAddHtlc]]:
+ def htlcs(self, subject: HTLCOwner, ctn: int | None = None) -> Sequence[Tuple[Direction, UpdateAddHtlc]]:
"""Return the list of HTLCs in subject's ctx at ctn."""
assert type(subject) is HTLCOwner
if ctn is None:
@@ -475,7 +475,7 @@ def was_htlc_failed(self, *, htlc_id: int, htlc_proposer: HTLCOwner) -> bool:
@with_lock
def all_settled_htlcs_ever_by_direction(self, subject: HTLCOwner, direction: Direction,
- ctn: int = None) -> Sequence[UpdateAddHtlc]:
+ ctn: int | None = None) -> Sequence[UpdateAddHtlc]:
"""Return the list of all HTLCs that have been ever settled in subject's
ctx up to ctn, filtered to only "direction".
"""
@@ -492,7 +492,7 @@ def all_settled_htlcs_ever_by_direction(self, subject: HTLCOwner, direction: Dir
return d
@with_lock
- def all_settled_htlcs_ever(self, subject: HTLCOwner, ctn: int = None) -> Sequence[Tuple[Direction, UpdateAddHtlc]]:
+ def all_settled_htlcs_ever(self, subject: HTLCOwner, ctn: int | None = None) -> Sequence[Tuple[Direction, UpdateAddHtlc]]:
"""Return the list of all HTLCs that have been ever settled in subject's
ctx up to ctn.
"""
@@ -510,7 +510,7 @@ def all_htlcs_ever(self) -> Sequence[Tuple[Direction, UpdateAddHtlc]]:
return sent + received
@with_lock
- def get_balance_msat(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int = None,
+ def get_balance_msat(self, whose: HTLCOwner, *, ctx_owner=HTLCOwner.LOCAL, ctn: int | None = None,
initial_balance_msat: int) -> int:
"""Returns the balance of 'whose' in 'ctx' at 'ctn'.
Only HTLCs that have been settled by that ctn are counted.
### electrum/lnonion.py
@@ -389,7 +389,7 @@ class ProcessedOnionPacket(NamedTuple):
are_we_final: bool
hop_data: OnionHopsDataSingle
next_packet: OnionPacket
- trampoline_onion_packet: OnionPacket
+ trampoline_onion_packet: OnionPacket | None
@property
def amt_to_forward(self) -> Optional[int]:
### electrum/lnpeer.py
@@ -330,7 +330,7 @@ def on_error(self, payload):
return
raise GracefulDisconnect
- def send_warning(self, channel_id: bytes, message: str = None, *, close_connection=False):
+ def send_warning(self, channel_id: bytes, message: str | None = None, *, close_connection=False):
"""Sends a warning and disconnects if close_connection.
Note:
@@ -349,7 +349,7 @@ def send_warning(self, channel_id: bytes, message: str = None, *, close_connecti
if close_connection:
raise GracefulDisconnect
- def send_error(self, channel_id: bytes, message: str = None, *, force_close_channel=False):
+ def send_error(self, channel_id: bytes, message: str | None = None, *, force_close_channel=False):
"""Sends an error message and force closes the channel.
Note:
@@ -993,7 +993,7 @@ async def channel_establishment_flow(
public: bool,
zeroconf: bool = False,
temp_channel_id: bytes,
- opening_fee: int = None,
+ opening_fee: int | None = None,
) -> Tuple[Channel, 'PartialTransaction']:
"""Implements the channel opening flow.
@@ -2200,7 +2200,7 @@ def _check_unfulfilled_htlc(
chan: Channel,
htlc: UpdateAddHtlc,
processed_onion: ProcessedOnionPacket,
- outer_onion_payment_secret: bytes = None, # used to group trampoline htlcs for forwarding
+ outer_onion_payment_secret: bytes | None = None, # used to group trampoline htlcs for forwarding
) -> str:
"""
Does additional checks on the incoming htlc and return the payment key if the tests pass,
### electrum/lnrouter.py
@@ -447,7 +447,7 @@ def add_edge_to_blacklist(
self,
short_channel_id: ShortChannelID,
*,
- now: int = None,
+ now: int | None = None,
duration: int = 3600, # seconds
) -> None:
if now is None:
### electrum/lnsweep.py
@@ -846,7 +846,7 @@ def tx_our_ctx_htlctx(
commit=ctx,
htlc=htlc,
ctx_output_idx=ctx_output_idx,
- name=f'our_ctx_{ctx_output_idx}_htlc_tx_{htlc.payment_hash.hex()}')
+ )
# sign HTLC output
remote_htlc_sig = chan.get_remote_htlc_sig_for_htlc(htlc_relative_idx=htlc_relative_idx)
@@ -934,7 +934,7 @@ def sweep_ctx_anchor(*, ctx: Transaction, multisig_key: Keypair) -> Optional[Par
def sweep_ctx_to_local(
*, ctx: Transaction, output_idx: int, witness_script: bytes,
privkey: bytes, is_revocation: bool,
- to_self_delay: int = None) -> Optional[PartialTxInput]:
+ to_self_delay: int | None = None) -> Optional[PartialTxInput]:
"""Create a txin that sweeps the 'to_local' output of a commitment
transaction into our wallet.
@@ -963,7 +963,7 @@ def sweep_htlctx_output(
htlctx_witness_script: bytes,
privkey: bytes,
is_revocation: bool,
- to_self_delay: int = None,
+ to_self_delay: int | None = None,
) -> Optional[PartialTxInput]:
"""Create a txn that sweeps the output of a first stage htlc tx
(i.e. sweeps from an HTLC-Timeout or an HTLC-Success tx).
### electrum/lnutil.py
@@ -1058,7 +1058,6 @@ def make_htlc_tx_with_open_channel(
commit: Transaction,
ctx_output_idx: int,
htlc: 'UpdateAddHtlc',
- name: str = None
) -> Tuple[bytes, PartialTransaction]:
amount_msat, cltv_abs, payment_hash = htlc.amount_msat, htlc.cltv_abs, htlc.payment_hash
for_us = subject == LOCAL
@@ -2053,7 +2052,7 @@ class ReceivedMPPStatus(NamedTuple):
# payment key of the final mpp set (derived from inner trampoline onion payment secret)
# to which the separate trampoline sets htlcs get added once they are complete.
# https://github.com/lightning/bolts/pull/829/commits/bc7a1a0bc97b2293e7f43dd8a06529e5fdcf7cd2
- parent_set_key: str = None
+ parent_set_key: str | None = None
def get_first_htlc_timestamp(self) -> Optional[int]:
return min([mpp_htlc.htlc.timestamp for mpp_htlc in self.htlcs], default=None)
### electrum/lnworker.py
@@ -1706,7 +1706,7 @@ def make_local_config_for_new_channel(
channel_type: ChannelType,
multisig_funding_keypair: Optional[Keypair], # if None, will get derived from channel_seed
peer_features: LnFeatures,
- channel_seed: bytes = None,
+ channel_seed: bytes | None = None,
) -> LocalConfig:
if channel_seed is None:
channel_seed = os.urandom(32)
@@ -1840,7 +1840,7 @@ def open_channel(
funding_sat: int,
push_amt_sat: int,
public: bool = False,
- password: str = None,
+ password: str | None = None,
) -> Tuple[Channel, PartialTransaction]:
fut = asyncio.run_coroutine_threadsafe(self.lnpeermgr.add_peer(connect_str), get_asyncio_loop())
@@ -1887,8 +1887,8 @@ def can_pay_invoice(self, invoice: Invoice) -> bool:
@log_exceptions
async def pay_invoice(
self, invoice: Invoice, *,
- amount_msat: int = None, # to overwrite amt in invoice
- attempts: int = None, # used only in unit tests
+ amount_msat: int | None = None, # to overwrite amt in invoice
+ attempts: int | None = None, # used only in unit tests
full_path: LNPaymentPath = None,
channels: Optional[Sequence[Channel]] = None, # my own direct channels
budget: Optional[PaymentFeeBudget] = None, # to limit max fee
@@ -1974,12 +1974,12 @@ async def pay_to_node(
min_final_cltv_delta: int,
r_tags,
invoice_features: int,
- attempts: int = None,
+ attempts: int | None = None,
full_path: LNPaymentPath = None,
fwd_trampoline_onion: OnionPacket = None,
budget: PaymentFeeBudget,
channels: Optional[Sequence[Channel]] = None,
- fw_payment_key: str = None, # for forwarding
+ fw_payment_key: str | None = None, # for forwarding
) -> None:
"""
Can raise PaymentFailure, ChannelDBNotLoaded,
@@ -2132,7 +2132,7 @@ async def pay_to_route(
sent_htlc_info: SentHtlcInfo,
min_final_cltv_delta: int,
trampoline_onion: Optional[OnionPacket] = None,
- fw_payment_key: str = None,
+ fw_payment_key: str | None = None,
) -> None:
"""Sends a single HTLC."""
shi = sent_htlc_info
@@ -2300,7 +2300,7 @@ def _decode_channel_update_msg(cls, chan_upd_msg: bytes) -> Optional[Dict[str, A
except Exception:
return None
- def _check_bolt11_invoice(self, bolt11_invoice: str, *, amount_msat: int = None, max_min_final_cltv_delta=NBLOCK_CLTV_DELTA_TOO_FAR_INTO_FUTURE) -> BOLT11Addr:
+ def _check_bolt11_invoice(self, bolt11_invoice: str, *, amount_msat: int | None = None, max_min_final_cltv_delta=NBLOCK_CLTV_DELTA_TOO_FAR_INTO_FUTURE) -> BOLT11Addr:
"""Parses and validates a bolt11 invoice str into a BOLT11Addr.
Includes pre-payment checks external to the parser.
"""
### electrum/mnemonic.py
@@ -158,7 +158,7 @@ class Mnemonic(Logger):
# Seed derivation does not follow BIP39
# Mnemonic phrase uses a hash based checksum, instead of a wordlist-dependent checksum
- def __init__(self, lang: str = None):
+ def __init__(self, lang: str | None = None):
Logger.__init__(self)
lang = lang or 'en'
self.logger.info(f'language {lang}')
@@ -198,7 +198,7 @@ def mnemonic_decode(self, seed: str) -> int:
i = i*n + k
return i
- def make_seed(self, *, seed_type: str = None, num_bits: int = None) -> str:
+ def make_seed(self, *, seed_type: str | None = None, num_bits: int | None = None) -> str:
from .keystore import bip39_is_checksum_valid
if seed_type is None:
seed_type = 'segwit'
### electrum/network.py
@@ -185,7 +185,7 @@ def set_defaults(self):
def serialize_proxy_cfgstr(self):
return ':'.join([self.mode, self.host, self.port])
- def deserialize_proxy_cfgstr(self, s: Optional[str], user: str = None, password: str = None) -> None:
+ def deserialize_proxy_cfgstr(self, s: str | None, user: str | None = None, password: str | None = None) -> None:
if s is None or (isinstance(s, str) and s.lower() == 'none'):
self.set_defaults()
self.user = user
@@ -1314,9 +1314,9 @@ async def maintain_main_interface():
@classmethod
async def async_send_http_on_proxy(
cls, method: str, url: str, *,
- params: dict = None,
- body: bytes = None,
- json: dict = None,
+ params: dict | None = None,
+ body: bytes | None = None,
+ json: dict | None = None,
headers=None,
on_finish=None,
timeout=None,
@@ -1366,7 +1366,7 @@ async def send_multiple_requests(
method: str,
params: Sequence,
*,
- timeout: int = None,
+ timeout: int | None = None,
):
if timeout is None:
timeout = self.get_network_timeout_seconds(NetworkTimeout.Urgent)
### electrum/onion_message.py
@@ -265,7 +265,7 @@ def send_onion_message_to(
lnwallet: 'LNWallet',
node_id_or_blinded_path: bytes,
destination_payload: dict,
- session_key: bytes = None
+ session_key: bytes | None = None
) -> None:
if session_key is None:
session_key = os.urandom(32)
### electrum/payment_identifier.py
@@ -370,7 +370,7 @@ def finalize(
self,
*,
amount_sat: int = 0,
- comment: str = None,
+ comment: str | None = None,
on_finished: Callable[['PaymentIdentifier'], None] = None,
):
assert self._state == PaymentIdentifierState.LNURLP_FINALIZE
@@ -381,9 +381,9 @@ def finalize(
async def _do_finalize(
self,
*,
- amount_sat: int = None,
- comment: str = None,
- on_finished: Callable[['PaymentIdentifier'], None] = None,
+ amount_sat: int | None = None,
+ comment: str | None = None,
+ on_finished: Callable[['PaymentIdentifier'], None] | None = None,
):
from .invoices import Invoice
try:
@@ -617,7 +617,7 @@ def invoice_from_payment_identifier(
pi: 'PaymentIdentifier',
wallet: 'Abstract_Wallet',
amount_sat: Union[int, str],
- message: str = None
+ message: str | None = None
) -> Optional[Invoice]:
assert pi.state in [PaymentIdentifierState.AVAILABLE,]
assert pi.is_onchain() if amount_sat == '!' else True # MAX should only be allowed if pi has onchain destination
### electrum/plugin.py
@@ -78,7 +78,7 @@ class Plugins(DaemonThread):
keyfile_windows = r'HKEY_LOCAL_MACHINE\SOFTWARE\Electrum\PluginsKey'
@profiler
- def __init__(self, config: SimpleConfig, gui_name: str = None, cmd_only: bool = False):
+ def __init__(self, config: SimpleConfig, gui_name: str | None = None, cmd_only: bool = False):
self.config = config
self.cmd_only = cmd_only # type: bool
self.internal_plugin_metadata = {}
### electrum/plugins/nwc/nwcserver.py
@@ -380,7 +380,7 @@ async def _handle_single_request(self, event: nEvent) -> None:
if task:
await self.taskgroup.spawn(self.run_request_task(task, request_event=event, request_method=method))
- async def run_request_task(self, task: Awaitable, *, request_event: nEvent, request_method: str = None) -> None:
+ async def run_request_task(self, task: Awaitable, *, request_event: nEvent, request_method: str | None = None) -> None:
"""Catches request handling exceptions and send an error response"""
try:
await task
@@ -397,7 +397,7 @@ async def send_error(
error_type: str,
error_msg: str = "",
*,
- error_restype: str = None,
+ error_restype: str | None = None,
) -> None:
"""Sends an error as response to the passed nEvent, containing the error type and message"""
to_pubkey_hex = causing_event.pubkey
### electrum/plugins/psbt_nostr/psbt_nostr.py
@@ -254,7 +254,7 @@ def mark_pending_event_rcvd(self, event_id):
self.known_events[event_id] = now()
run_sync_function_on_asyncio_thread(self.pending.set, block=False)
- def prepare_messages(self, tx: Union[Transaction, PartialTransaction], label: str = None) -> List[Tuple[str, dict]]:
+ def prepare_messages(self, tx: Union[Transaction, PartialTransaction], label: str | None = None) -> List[Tuple[str, dict]]:
messages = []
for xpub, pubkey in self.cosigner_list:
if not self.cosigner_can_sign(tx, xpub):
@@ -265,7 +265,7 @@ def prepare_messages(self, tx: Union[Transaction, PartialTransaction], label: st
messages.append((pubkey, payload))
return messages
- def send_psbt(self, tx: Union[Transaction, PartialTransaction], label: str):
+ def send_psbt(self, tx: Union[Transaction, PartialTransaction], label: str | None):
self.do_send(self.prepare_messages(tx, label), tx.txid())
def do_send(self, messages: List[Tuple[str, dict]], txid: Optional[str] = None):
@@ -278,9 +278,9 @@ def add_transaction_to_wallet(
self,
tx: Union['Transaction', 'PartialTransaction'],
*,
- label: str = None,
- on_failure: Callable[[str], None] = None,
- on_success: Callable[[], None] = None
+ label: str | None = None,
+ on_failure: Callable[[str], None] | None = None,
+ on_success: Callable[[], None] | None = None
) -> None:
assert tx.txid(), "Shouldn't allow to save tx without txid"
try:
### electrum/plugins/psbt_nostr/qml.py
@@ -65,7 +65,7 @@ def canSendPsbt(self, wallet: 'QEWallet', tx: str) -> bool:
@pyqtSlot(QEWallet, str)
@pyqtSlot(QEWallet, str, str)
- def sendPsbt(self, wallet: 'QEWallet', tx: str, label: str = None):
+ def sendPsbt(self, wallet: 'QEWallet', tx: str, label: str | None = None):
cosigner_wallet = self._plugin.cosigner_wallets.get(wallet.wallet)
if not cosigner_wallet:
return
### electrum/plugins/watchtower/watchtower.py
@@ -152,7 +152,7 @@ async def check_onchain_situation(self, address, funding_outpoint):
if not keep_watching:
await self.unwatch_channel(address, funding_outpoint)
- def inspect_tx_candidate(self, outpoint, n: int) -> Dict[str, str]:
+ def inspect_tx_candidate(self, outpoint: str, n: int) -> Dict[str, str | None]:
"""
returns a dict of spenders for a transaction of interest.
subscribes to addresses as a side effect.
### electrum/submarine_swaps.py
@@ -253,11 +253,11 @@ class SwapManager(Logger):
def __init__(self, *, wallet: 'Abstract_Wallet', lnworker: 'LNWallet'):
Logger.__init__(self)
- self.mining_fee = None
+ self.mining_fee = None # type: int | None
self.percentage = None # type: Optional[Decimal]
- self._min_amount = None
- self._max_forward = None
- self._max_reverse = None
+ self._min_amount = None # type: int | None
+ self._max_forward = None # type: int | None
+ self._max_reverse = None # type: int | None
self.wallet = wallet
self.config = wallet.config
@@ -746,7 +746,7 @@ async def hold_invoice_callback(self, payment_hash: bytes) -> None:
output = self.create_funding_output(swap)
self.wallet.txbatcher.add_payment_output('swaps', output)
- def create_normal_swap(self, *, lightning_amount_sat: int, payment_hash: bytes, their_pubkey: bytes = None):
+ def create_normal_swap(self, *, lightning_amount_sat: int, payment_hash: bytes, their_pubkey: bytes | None = None):
""" server method """
assert lightning_amount_sat
if payment_hash.hex() in self._swaps:
### electrum/trampoline.py
@@ -195,9 +195,9 @@ def is_legacy_relay(invoice_features, r_tags) -> Tuple[bool, Set[bytes]]:
def _extend_trampoline_route(
route: List[TrampolineEdge],
*,
- start_node: bytes = None,
+ start_node: bytes | None = None,
end_node: bytes,
- fee_info: tuple = None,
+ fee_info: tuple | None = None,
):
"""Extends the route and modifies it in place."""
if start_node is None:
### electrum/transaction.py
@@ -327,9 +327,9 @@ class TxInput:
def __init__(self, *,
prevout: TxOutpoint,
- script_sig: bytes = None,
+ script_sig: bytes | None = None,
nsequence: int = 0xffffffff - 1,
- witness: bytes = None,
+ witness: bytes | None = None,
is_coinbase_output: bool = False):
self.prevout = prevout
self.script_sig = script_sig
@@ -435,7 +435,7 @@ def to_json(self):
d['witness'] = [x.hex() for x in self.witness_elements()]
return d
- def serialize_to_network(self, *, script_sig: bytes = None) -> bytes:
+ def serialize_to_network(self, *, script_sig: bytes | None = None) -> bytes:
if script_sig is None:
script_sig = self.script_sig
# Prev hash and index
@@ -907,8 +907,8 @@ def __init__(self, raw):
self._cached_network_ser = raw.hex()
else:
raise Exception(f"cannot initialize transaction from {raw}")
- self._inputs = None # type: List[TxInput]
- self._outputs = None # type: List[TxOutput]
+ self._inputs = None # type: List[TxInput] | None
+ self._outputs = None # type: List[TxOutput] | None
self._locktime = 0
self._version = 2
@@ -1045,7 +1045,7 @@ def serialize_preimage(
txin_index: int,
*,
sighash: Optional[int] = None,
- sighash_cache: SighashCache = None,
+ sighash_cache: SighashCache | None = None,
) -> bytes:
nVersion = int.to_bytes(self.version, length=4, byteorder="little", signed=True)
nLocktime = int.to_bytes(self.locktime, length=4, byteorder="little", signed=False)
@@ -1144,7 +1144,7 @@ def verify_sig_for_txin(
txin_index: int,
pubkey_bytes: bytes,
sig: bytes,
- sighash_cache: SighashCache = None,
+ sighash_cache: SighashCache | None = None,
) -> bool:
txin = self.inputs()[txin_index]
if txin.is_taproot():
@@ -2059,7 +2059,7 @@ def is_taproot(self) -> Optional[bool]:
def already_has_some_signatures(self) -> bool:
"""Returns whether progress has been made towards completing this input."""
- return (self.sigs_ecdsa
+ return (bool(self.sigs_ecdsa)
or self.tap_key_sig is not None
or self.script_sig is not None
or self.witness is not None)
@@ -2336,8 +2336,8 @@ def from_io(
inputs: Sequence[PartialTxInput],
outputs: Sequence[PartialTxOutput],
*,
- locktime: int = None,
- version: int = None,
+ locktime: int | None = None,
+ version: int | None = None,
BIP69_sort: bool = True
) -> 'PartialTransaction':
self = cls()
@@ -2495,7 +2495,7 @@ def sign_txin(
txin_index: int,
privkey_bytes: bytes,
*,
- sighash_cache: SighashCache = None,
+ sighash_cache: SighashCache | None = None,
) -> bytes:
txin = self.inputs()[txin_index]
txin.validate_data(for_signing=True)
### electrum/wallet.py
@@ -273,8 +273,8 @@ def __init__(
self,
*,
risk_level: TxSighashRiskLevel = TxSighashRiskLevel.SAFE,
- short_message: str = None,
- messages: List[str] = None,
+ short_message: str | None = None,
+ messages: List[str] | None = None,
):
self.risk_level = risk_level
self.short_message = short_message
@@ -748,7 +748,7 @@ def _set_label(self, key: str, value: Optional[str]) -> None:
else:
self._labels[key] = value
- def set_label(self, name: str, text: str = None) -> bool:
+ def set_label(self, name: str, text: str | None = None) -> bool:
if not name:
return False
changed = False
@@ -1984,7 +1984,7 @@ def make_unsigned_transaction(
outputs: List[PartialTxOutput],
inputs: Optional[List[PartialTxInput]] = None,
fee_policy: FeePolicy,
- change_addr: str = None,
+ change_addr: str | None = None,
is_sweep: bool = False, # used by Wallet_2fa subclass
rbf: bool = True,
BIP69_sort: Optional[bool] = True,
@@ -2654,7 +2654,7 @@ def _add_input_utxo_info(
self,
txin: PartialTxInput,
*,
- address: str = None,
+ address: str | None = None,
) -> None:
# - We prefer to include UTXO (full tx), even for segwit inputs (see #6198).
# - For witness v0 inputs, we include *both* UTXO and WITNESS_UTXO. UTXO is a strict superset,
### tests/lnhelpers.py
@@ -351,11 +351,11 @@ def prepare_invoice(
*,
amount_msat=100_000_000,
include_routing_hints=False,
- payment_preimage: bytes = None,
- payment_hash: bytes = None,
+ payment_preimage: bytes | None = None,
+ payment_hash: bytes | None = None,
invoice_features: LnFeatures = None,
- min_final_cltv_delta: int = None,
- expiry: int = None,
+ min_final_cltv_delta: int | None = None,
+ expiry: int | None = None,
) -> Tuple[BOLT11Addr, Invoice]:
amount_btc = amount_msat/Decimal(COIN*1000)
if payment_preimage is None and not payment_hash:
### tests/test_commands.py
@@ -748,7 +748,7 @@ async def test_add_peer(self, *mock_args):
mock_peer = mock.Mock()
mock_peer.initialized = asyncio.Future()
connection_string = "test_node_id@127.0.0.1:9735"
- called = False
+ called = 0
async def lnpeermgr_add_peer(*args, **kwargs):
assert args[0] == connection_string
nonlocal called
### tests/test_lnpeer.py
@@ -412,7 +412,7 @@ async def f(
*,
ctn_delta: int = 0,
revnum_delta: int = 0,
- last_rev_secret: bytes = None,
+ last_rev_secret: bytes | None = None,
) -> tuple[Channel, Channel]:
alice_lnwallet, bob_lnwallet = self.prepare_lnwallets(self.GRAPH_DEFINITIONS['single_chan']).values()
alice_channel, bob_channel = create_test_channels(alice_lnwallet=alice_lnwallet, bob_lnwallet=bob_lnwallet)
@@ -2668,7 +2668,7 @@ async def test_forwarder_fails_for_inconsistent_trampoline_onions(self):
"""
# store a modified trampoline onion to be injected into lnworker.new_onion_packet later when sending the htlcs
- modified_trampoline_onion = None
+ modified_trampoline_onion = None # type: OnionPacket | None
def modified_new_onion_packet_trampoline(payment_path_pubkeys, session_key, hops_data: List[OnionHopsDataSingle], **kwargs):
nonlocal modified_trampoline_onion
assert modified_trampoline_onion is None, "this mock should get called only once"
### tests/test_submarine_swaps.py
@@ -57,7 +57,7 @@ class TestSwapClaim(ToyServerTestCase):
test, and build its claim tx with the same code the server would use.
"""
- def create_config(self, name: str = None) -> SimpleConfig:
+ def create_config(self, name: str | None = None) -> SimpleConfig:
config = super().create_config(name)
config.FEE_POLICY_SWAPS = 'feerate:5000'
return config
### tests/toyserver/toynetwork.py
@@ -33,7 +33,7 @@ def __init__(self, *, config: SimpleConfig):
self.interface = None # type: Interface | None
self.relay_fee = None # type: int | None # sat/kbyte, set from the server on connect
- async def connect(self, server: ToyServer, *, client_name: str = None) -> Interface:
+ async def connect(self, server: ToyServer, *, client_name: str | None = None) -> Interface:
"""connect to server, and wait until we have synced its headers"""
assert self.interface is None, "already connected"
interface = Interface(network=self, server=ServerAddr(host="127.0.0.1", port=server.server_port, protocol="t"))
### tests/toyserver/toyserver.py
@@ -62,7 +62,7 @@ def recurse(node: str):
@dataclass(kw_only=True, slots=True, frozen=True)
class FakeBlock:
header: bytes
- txids: Sequence[str] = None # FIXME needs OrderedSet with index-based lookup? >.<
+ txids: Sequence[str] | None = None # FIXME needs OrderedSet with index-based lookup? >.<
def __post_init__(self):
if self.txids is None: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.