• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

safe-global / safe-cli / 11343896716

15 Oct 2024 10:09AM CUT coverage: 88.64%. Remained the same
11343896716

push

github

Uxio0
Bump flake8 from 7.1.0 to 7.1.1

Bumps [flake8](https://github.com/pycqa/flake8) from 7.1.0 to 7.1.1.
- [Commits](https://github.com/pycqa/flake8/compare/7.1.0...7.1.1)

---
updated-dependencies:
- dependency-name: flake8
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

221 of 262 branches covered (84.35%)

Branch coverage included in aggregate %.

2869 of 3224 relevant lines covered (88.99%)

3.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.61
/src/safe_cli/main.py
1
#!/bin/env python3
2
import json
4✔
3
import os
4✔
4
import sys
4✔
5
from pathlib import Path
4✔
6
from typing import Annotated, List
4✔
7

8
import typer
4✔
9
from art import text2art
4✔
10
from eth_typing import ChecksumAddress
4✔
11
from hexbytes import HexBytes
4✔
12
from prompt_toolkit import HTML, print_formatted_text
4✔
13
from typer.main import get_command, get_command_name
4✔
14
from web3 import Web3
4✔
15

16
from . import VERSION
4✔
17
from .argparse_validators import check_hex_str
4✔
18
from .operators import SafeOperator
4✔
19
from .safe_cli import SafeCli
4✔
20
from .tx_builder.tx_builder_file_decoder import convert_to_proposed_transactions
4✔
21
from .typer_validators import (
4✔
22
    ChecksumAddressParser,
23
    HexBytesParser,
24
    check_ethereum_address,
25
    check_private_keys,
26
)
27
from .utils import get_safe_from_owner
4✔
28

29
app = typer.Typer(name="Safe CLI")
4✔
30

31

32
def _build_safe_operator_and_load_keys(
4✔
33
    safe_address: ChecksumAddress,
34
    node_url: str,
35
    private_keys: List[str],
36
    interactive: bool,
37
) -> SafeOperator:
38
    safe_operator = SafeOperator(safe_address, node_url, interactive=interactive)
4✔
39
    safe_operator.load_cli_owners(private_keys)
4✔
40
    return safe_operator
4✔
41

42

43
def _check_interactive_mode(interactive_mode: bool) -> bool:
4✔
44
    if not interactive_mode:
4✔
45
        return False
4✔
46

47
    # --non-interactive arg > env var.
48
    env_var = os.getenv("SAFE_CLI_INTERACTIVE")
4✔
49
    if env_var:
4✔
50
        return env_var.lower() in ("true", "1", "yes")
×
51

52
    return True
4✔
53

54

55
# Common Options
56
safe_address_option = Annotated[
4✔
57
    ChecksumAddress,
58
    typer.Argument(
59
        help="The address of the Safe.",
60
        callback=check_ethereum_address,
61
        click_type=ChecksumAddressParser(),
62
        show_default=False,
63
    ),
64
]
65
node_url_option = Annotated[
4✔
66
    str, typer.Argument(help="Ethereum node url.", show_default=False)
67
]
68
to_option = Annotated[
4✔
69
    ChecksumAddress,
70
    typer.Argument(
71
        help="The address of destination.",
72
        callback=check_ethereum_address,
73
        click_type=ChecksumAddressParser(),
74
        show_default=False,
75
    ),
76
]
77
interactive_option = Annotated[
4✔
78
    bool,
79
    typer.Option(
80
        "--interactive/--non-interactive",
81
        help=(
82
            "Enable interactive mode to allow user input during execution. "
83
            "Use --non-interactive to disable prompts and run unattended. "
84
            "This is useful for scripting and automation where no user intervention is required."
85
        ),
86
        rich_help_panel="Optional Arguments",
87
        callback=_check_interactive_mode,
88
    ),
89
]
90

91

92
@app.command()
4✔
93
def send_ether(
4✔
94
    safe_address: safe_address_option,
95
    node_url: node_url_option,
96
    to: to_option,
97
    value: Annotated[
98
        int, typer.Argument(help="Amount of ether in wei to send.", show_default=False)
99
    ],
100
    private_key: Annotated[
101
        List[str],
102
        typer.Option(
103
            help="List of private keys of signers.",
104
            rich_help_panel="Optional Arguments",
105
            show_default=False,
106
            callback=check_private_keys,
107
        ),
108
    ] = None,
109
    safe_nonce: Annotated[
110
        int,
111
        typer.Option(
112
            help="Force nonce for tx_sender",
113
            rich_help_panel="Optional Arguments",
114
            show_default=False,
115
        ),
116
    ] = None,
117
    interactive: interactive_option = True,
118
):
119
    safe_operator = _build_safe_operator_and_load_keys(
4✔
120
        safe_address, node_url, private_key, interactive
121
    )
122
    safe_operator.send_ether(to, value, safe_nonce=safe_nonce)
4✔
123

124

125
@app.command()
4✔
126
def send_erc20(
4✔
127
    safe_address: safe_address_option,
128
    node_url: node_url_option,
129
    to: to_option,
130
    token_address: Annotated[
131
        ChecksumAddress,
132
        typer.Argument(
133
            help="Erc20 token address.",
134
            callback=check_ethereum_address,
135
            click_type=ChecksumAddressParser(),
136
            show_default=False,
137
        ),
138
    ],
139
    amount: Annotated[
140
        int,
141
        typer.Argument(
142
            help="Amount of erc20 tokens in wei to send.", show_default=False
143
        ),
144
    ],
145
    private_key: Annotated[
146
        List[str],
147
        typer.Option(
148
            help="List of private keys of signers.",
149
            rich_help_panel="Optional Arguments",
150
            show_default=False,
151
            callback=check_private_keys,
152
        ),
153
    ] = None,
154
    safe_nonce: Annotated[
155
        int,
156
        typer.Option(
157
            help="Force nonce for tx_sender",
158
            rich_help_panel="Optional Arguments",
159
            show_default=False,
160
        ),
161
    ] = None,
162
    interactive: interactive_option = True,
163
):
164
    safe_operator = _build_safe_operator_and_load_keys(
4✔
165
        safe_address, node_url, private_key, interactive
166
    )
167
    safe_operator.send_erc20(to, token_address, amount, safe_nonce=safe_nonce)
4✔
168

169

170
@app.command()
4✔
171
def send_erc721(
4✔
172
    safe_address: safe_address_option,
173
    node_url: node_url_option,
174
    to: to_option,
175
    token_address: Annotated[
176
        ChecksumAddress,
177
        typer.Argument(
178
            help="Erc721 token address.",
179
            callback=check_ethereum_address,
180
            click_type=ChecksumAddressParser(),
181
            show_default=False,
182
        ),
183
    ],
184
    token_id: Annotated[
185
        int, typer.Argument(help="Erc721 token id.", show_default=False)
186
    ],
187
    private_key: Annotated[
188
        List[str],
189
        typer.Option(
190
            help="List of private keys of signers.",
191
            rich_help_panel="Optional Arguments",
192
            show_default=False,
193
            callback=check_private_keys,
194
        ),
195
    ] = None,
196
    safe_nonce: Annotated[
197
        int,
198
        typer.Option(
199
            help="Force nonce for tx_sender",
200
            rich_help_panel="Optional Arguments",
201
            show_default=False,
202
        ),
203
    ] = None,
204
    interactive: interactive_option = True,
205
):
206
    safe_operator = _build_safe_operator_and_load_keys(
4✔
207
        safe_address, node_url, private_key, interactive
208
    )
209
    safe_operator.send_erc721(to, token_address, token_id, safe_nonce=safe_nonce)
4✔
210

211

212
@app.command()
4✔
213
def send_custom(
4✔
214
    safe_address: safe_address_option,
215
    node_url: node_url_option,
216
    to: to_option,
217
    value: Annotated[int, typer.Argument(help="Value to send.", show_default=False)],
218
    data: Annotated[
219
        HexBytes,
220
        typer.Argument(
221
            help="HexBytes data to send.",
222
            callback=check_hex_str,
223
            click_type=HexBytesParser(),
224
            show_default=False,
225
        ),
226
    ],
227
    private_key: Annotated[
228
        List[str],
229
        typer.Option(
230
            help="List of private keys of signers.",
231
            rich_help_panel="Optional Arguments",
232
            show_default=False,
233
            callback=check_private_keys,
234
        ),
235
    ] = None,
236
    safe_nonce: Annotated[
237
        int,
238
        typer.Option(
239
            help="Force nonce for tx_sender",
240
            rich_help_panel="Optional Arguments",
241
            show_default=False,
242
        ),
243
    ] = None,
244
    delegate: Annotated[
245
        bool,
246
        typer.Option(
247
            help="Use DELEGATE_CALL. By default use CALL",
248
            rich_help_panel="Optional Arguments",
249
        ),
250
    ] = False,
251
    interactive: interactive_option = True,
252
):
253
    safe_operator = _build_safe_operator_and_load_keys(
4✔
254
        safe_address, node_url, private_key, interactive
255
    )
256
    safe_operator.send_custom(
4✔
257
        to, value, data, safe_nonce=safe_nonce, delegate_call=delegate
258
    )
259

260

261
@app.command()
4✔
262
def tx_builder(
4✔
263
    safe_address: safe_address_option,
264
    node_url: node_url_option,
265
    file_path: Annotated[
266
        Path,
267
        typer.Argument(
268
            exists=True,
269
            file_okay=True,
270
            dir_okay=False,
271
            writable=False,
272
            readable=True,
273
            resolve_path=True,
274
            help="File path with tx_builder data.",
275
            show_default=False,
276
        ),
277
    ],
278
    private_key: Annotated[
279
        List[str],
280
        typer.Option(
281
            help="List of private keys of signers.",
282
            rich_help_panel="Optional Arguments",
283
            show_default=False,
284
            callback=check_private_keys,
285
        ),
286
    ] = None,
287
    interactive: interactive_option = True,
288
):
289
    safe_operator = _build_safe_operator_and_load_keys(
4✔
290
        safe_address, node_url, private_key, interactive
291
    )
292
    data = json.loads(file_path.read_text())
4✔
293
    safe_txs = [
4✔
294
        safe_operator.prepare_safe_transaction(tx.to, tx.value, tx.data)
295
        for tx in convert_to_proposed_transactions(data)
296
    ]
297

298
    if len(safe_txs) == 0:
4✔
299
        raise typer.BadParameter("No transactions found.")
4✔
300

301
    if len(safe_txs) == 1:
4✔
302
        safe_operator.execute_safe_transaction(safe_txs[0])
4✔
303
    else:
304
        multisend_tx = safe_operator.batch_safe_txs(safe_operator.get_nonce(), safe_txs)
4✔
305
        if multisend_tx is not None:
×
306
            safe_operator.execute_safe_transaction(multisend_tx)
×
307

308

309
@app.command()
4✔
310
def version():
4✔
311
    print(f"Safe Cli v{VERSION}")
4✔
312

313

314
@app.command(
4✔
315
    hidden=True,
316
    name="attended-mode",
317
    help="""
318
        safe-cli [--history] [--get-safes-from-owner] address node_url\n
319
        Examples:\n
320
            safe-cli 0x0000000000000000000000000000000000000000 https://sepolia.drpc.org\n
321
            safe-cli --get-safes-from-owner 0x0000000000000000000000000000000000000000 https://sepolia.drpc.org\n\n\n\n
322
            safe-cli --history 0x0000000000000000000000000000000000000000 https://sepolia.drpc.org\n
323
            safe-cli --history --get-safes-from-owner 0x0000000000000000000000000000000000000000 https://sepolia.drpc.org\n\n\n\n
324
            safe-cli send-ether 0xsafeaddress https://sepolia.drpc.org 0xtoaddress wei-amount --private-key key1 --private-key key1 --private-key keyN [--non-interactive]\n
325
            safe-cli send-erc721 0xsafeaddress https://sepolia.drpc.org 0xtoaddress 0xtokenaddres id --private-key key1 --private-key key2 --private-key keyN [--non-interactive]\n
326
            safe-cli send-erc20 0xsafeaddress https://sepolia.drpc.org 0xtoaddress 0xtokenaddres wei-amount --private-key key1 --private-key key2 --private-key keyN [--non-interactive]\n
327
            safe-cli send-custom 0xsafeaddress https://sepolia.drpc.org 0xtoaddress value 0xtxdata --private-key key1 --private-key key2 --private-key keyN [--non-interactive]\n\n\n\n
328
            safe-cli tx-builder 0xsafeaddress https://sepolia.drpc.org  ./path/to/exported/tx-builder/file.json --private-key key1 --private-key keyN [--non-interactive]
329
    """,
330
    epilog="Commands available in unattended mode:\n\n\n\n"
331
    + "\n\n".join(
332
        [
333
            f"  {get_command_name(command)}"
334
            for command in get_command(app).commands.keys()
335
        ]
336
    )
337
    + "\n\n\n\nUse the --help option of each command to see the usage options.",
338
)
339
def default_attended_mode(
4✔
340
    address: Annotated[
341
        ChecksumAddress,
342
        typer.Argument(
343
            help="The address of the Safe, or an owner address if --get-safes-from-owner is specified.",
344
            callback=check_ethereum_address,
345
            click_type=ChecksumAddressParser(),
346
            show_default=False,
347
        ),
348
    ],
349
    node_url: node_url_option,
350
    history: Annotated[
351
        bool,
352
        typer.Option(
353
            help="Enable history. By default it's disabled due to security reasons",
354
            rich_help_panel="Optional Arguments",
355
        ),
356
    ] = False,
357
    get_safes_from_owner: Annotated[
358
        bool,
359
        typer.Option(
360
            help="Indicates that address is an owner (Safe Transaction Service is required for this feature)",
361
            rich_help_panel="Optional Arguments",
362
        ),
363
    ] = False,
364
) -> None:
365
    print_formatted_text(text2art("Safe CLI"))  # Print fancy text
4✔
366
    print_formatted_text(HTML(f"<b>Version: {VERSION}</b>"))
4✔
367

368
    if get_safes_from_owner:
4✔
369
        safe_address_listed = get_safe_from_owner(address, node_url)
4✔
370
        safe_cli = SafeCli(safe_address_listed, node_url, history)
×
371
    else:
372
        safe_cli = SafeCli(address, node_url, history)
4✔
373
    safe_cli.print_startup_info()
4✔
374
    safe_cli.loop()
4✔
375

376

377
def _is_safe_cli_default_command(arguments: List[str]) -> bool:
4✔
378
    # safe-cli
379
    if len(sys.argv) == 1:
×
380
        return True
×
381

382
    if sys.argv[1] == "--help":
×
383
        return True
×
384

385
    # Only added if is not a valid command, and it is an address. safe-cli 0xaddress http://url
386
    if sys.argv[1] not in [
×
387
        get_command_name(key) for key in get_command(app).commands.keys()
388
    ] and Web3.is_checksum_address(sys.argv[1]):
389
        return True
×
390

391
    return False
×
392

393

394
def main():
4✔
395
    # By default, the attended mode is initialised. Otherwise, the required command must be specified.
396
    if _is_safe_cli_default_command(sys.argv):
×
397
        sys.argv.insert(1, "attended-mode")
×
398
    app()
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc