type hints: (trivial) add missing "Optional" qualifiers
What changed, and why it matters
This commit only fixes type hints in the source code. It changes annotations such as `name: str = None` to `name: str | None = None` so that developer tools correctly understand that these variables may be empty. No actual program behavior, logic, or security checks are changed.
No security action required. Treat as a normal code-quality/type-hint maintenance commit.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit is a large but purely cosmetic type-hint correction across 46 files. It replaces incorrect annotations where a parameter or attribute with a default of None was typed as non-optional (e.g., str, int, bytes) with union types (str | None, int | None, etc.). The runtime code is unchanged; there are no functional modifications, no added validations, and no security fixes.
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/keystore.pyelectrum/lnchannel.pyelectrum/lnhtlc.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/submarine_swaps.pyelectrum/trampoline.pyelectrum/transaction.pyelectrum/wallet.pytests/lnhelpers.pytests/test_lnpeer.pytests/test_submarine_swaps.pytests/toyserver/toynetwork.pyInspect captured patch +127 / −124
### 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)
### 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
@@ -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/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/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
@@ -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
@@ -2052,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/submarine_swaps.py
@@ -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
@@ -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():
@@ -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()
### 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_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)
### 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"))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.