Use "with suppress()" to suppress python exceptions

Using
```Python
with suppress(OSError):
    os.remove('somefile.tmp')
```
instead of
```Python
try:
    os.remove('somefile.tmp')
except OSError:
    pass
```
makes the code more compact and more readable IMO.

This pattern was recommended by Raymond Hettinger, a Python Core
Developer in his talk "Transforming Code into Beautiful, Idiomatic Python" at https://www.youtube.com/watch?v=OSGv2VnC0go. The transcript is available at https://github.com/JeffPaine/beautiful_idiomatic_python
This commit is contained in:
Luflosi 2019-06-03 11:50:07 +02:00
parent d6e750727f
commit 2b095f720e
No known key found for this signature in database
GPG Key ID: 14140F703B7D8362
22 changed files with 68 additions and 138 deletions

View File

@ -3,6 +3,7 @@
# License: GPL v3 Copyright: 2018, Kovid Goyal <kovid at kovidgoyal.net>
import os
from contextlib import suppress
from kitty.cli import parse_args
from kitty.constants import cache_dir
@ -27,10 +28,8 @@ class HistoryCompleter:
self.history_path = None
if name:
ddir = os.path.join(cache_dir(), 'ask')
try:
with suppress(FileExistsError):
os.makedirs(ddir)
except FileExistsError:
pass
self.history_path = os.path.join(ddir, name)
def complete(self, text, state):
@ -50,10 +49,8 @@ class HistoryCompleter:
def __enter__(self):
if self.history_path:
try:
with suppress(Exception):
readline.read_history_file(self.history_path)
except Exception:
pass
readline.set_completer(self.complete)
return self
@ -106,10 +103,8 @@ def main(args):
print(styled(args.message, bold=True))
prompt = '> '
try:
with suppress(KeyboardInterrupt, EOFError):
ans['response'] = input(prompt)
except (KeyboardInterrupt, EOFError):
pass
return ans

View File

@ -7,6 +7,7 @@ import re
from functools import lru_cache
from hashlib import md5
from mimetypes import guess_type
from contextlib import suppress
path_name_map = {}
@ -140,10 +141,8 @@ def is_image(path):
def data_for_path(path):
ans = raw_data_for_path(path)
if not is_image(path) and not os.path.samefile(path, os.devnull):
try:
with suppress(UnicodeDecodeError):
ans = ans.decode('utf-8')
except UnicodeDecodeError:
pass
return ans

View File

@ -9,6 +9,7 @@ import warnings
from collections import defaultdict
from functools import partial
from gettext import gettext as _
from contextlib import suppress
from kitty.cli import CONFIG_HELP, parse_args
from kitty.constants import appname
@ -511,10 +512,8 @@ usage = 'file_or_directory_left file_or_directory_right'
def terminate_processes(processes):
for pid in processes:
try:
with suppress(Exception):
os.kill(pid, signal.SIGKILL)
except Exception:
pass
def main(args):

View File

@ -8,6 +8,7 @@ import sys
from base64 import standard_b64encode
from collections import defaultdict, deque
from itertools import count
from contextlib import suppress
from kitty.utils import fit_image
@ -172,10 +173,8 @@ class ImageManager:
if in_flight:
pl = in_flight.popleft()
if payload.startswith('ENOENT:'):
try:
with suppress(Exception):
self.resend_image(image_id, pl)
except Exception:
pass
if not in_flight:
self.placements_in_flight.pop(image_id, None)

View File

@ -7,6 +7,7 @@ import string
import subprocess
from functools import lru_cache
from gettext import gettext as _
from contextlib import suppress
from kitty.config import cached_values_for
from kitty.constants import config_dir
@ -293,24 +294,20 @@ class UnicodeInput(Handler):
self.update_codepoints()
self.current_char = None
if self.mode is HEX:
try:
with suppress(Exception):
if self.line_edit.current_input.startswith(INDEX_CHAR) and len(self.line_edit.current_input) > 1:
self.current_char = chr(self.table.codepoint_at_hint(self.line_edit.current_input[1:]))
else:
code = int(self.line_edit.current_input, 16)
self.current_char = chr(code)
except Exception:
pass
elif self.mode is NAME:
cc = self.table.current_codepoint
if cc:
self.current_char = chr(cc)
else:
try:
with suppress(Exception):
if self.line_edit.current_input:
self.current_char = chr(self.table.codepoint_at_hint(self.line_edit.current_input.lstrip(INDEX_CHAR)))
except Exception:
pass
if self.current_char is not None:
code = ord(self.current_char)
if not codepoint_ok(code):
@ -483,10 +480,8 @@ def main(args):
handler = UnicodeInput(cached_values)
loop.loop(handler)
if handler.current_char and loop.return_code == 0:
try:
with suppress(Exception):
handler.recent.remove(ord(handler.current_char))
except Exception:
pass
recent = [ord(handler.current_char)] + handler.recent
cached_values['recent'] = recent[:len(DEFAULT_SET)]
return handler.current_char

View File

@ -9,6 +9,7 @@ import re
from functools import partial
from gettext import gettext as _
from weakref import WeakValueDictionary
from contextlib import suppress
from .child import cached_process_data
from .cli import create_opts, parse_args
@ -639,16 +640,12 @@ class Boss:
def notify_on_os_window_death(self, address):
import socket
s = socket.socket(family=socket.AF_UNIX)
try:
with suppress(Exception):
s.connect(address)
s.sendall(b'c')
try:
with suppress(EnvironmentError):
s.shutdown(socket.SHUT_RDWR)
except EnvironmentError:
pass
s.close()
except Exception:
pass
def display_scrollback(self, window, data, cmd):
tab = self.active_tab
@ -1032,18 +1029,14 @@ class Boss:
def safe_delete_temp_file(self, path):
if is_path_in_temp_dir(path):
try:
with suppress(FileNotFoundError):
os.remove(path)
except FileNotFoundError:
pass
def set_update_check_process(self, process=None):
if self.update_check_process is not None:
try:
with suppress(Exception):
if self.update_check_process.poll() is None:
self.update_check_process.kill()
except Exception:
pass
self.update_check_process = process
def on_monitored_pid_death(self, pid, exit_status):

View File

@ -6,6 +6,7 @@ import fcntl
import os
from collections import defaultdict
from contextlib import contextmanager
from contextlib import suppress
import kitty.fast_data_types as fast_data_types
@ -220,14 +221,10 @@ class Child:
def process_desc(pid):
ans = {'pid': pid}
try:
with suppress(Exception):
ans['cmdline'] = cmdline_of_process(pid)
except Exception:
pass
try:
with suppress(Exception):
ans['cwd'] = cwd_of_process(pid) or None
except Exception:
pass
return ans
return list(map(process_desc, foreground_processes))
@ -250,25 +247,19 @@ class Child:
@property
def current_cwd(self):
try:
with suppress(Exception):
return cwd_of_process(self.pid)
except Exception:
pass
@property
def pid_for_cwd(self):
try:
with suppress(Exception):
pgrp = os.tcgetpgrp(self.child_fd)
foreground_processes = processes_in_group(pgrp) if pgrp >= 0 else []
if len(foreground_processes) == 1:
return foreground_processes[0]
except Exception:
pass
return self.pid
@property
def foreground_cwd(self):
try:
with suppress(Exception):
return cwd_of_process(self.pid_for_cwd) or None
except Exception:
pass

View File

@ -9,6 +9,7 @@
# will replay the commands and pause at the end waiting for user to press enter
import sys
from contextlib import suppress
CSI = '\033['
@ -158,7 +159,5 @@ def replay(raw):
def main(path):
raw = open(path).read()
replay(raw)
try:
with suppress(EOFError, KeyboardInterrupt):
input()
except (EOFError, KeyboardInterrupt):
pass

View File

@ -5,6 +5,7 @@
import json
import os
import sys
from contextlib import suppress
from .cli import parse_args
from .config import parse_config, parse_send_text_bytes
@ -904,7 +905,5 @@ def parse_subcommand_cli(func, args):
def display_subcommand_help(func):
try:
with suppress(SystemExit):
parse_args(['--help'], (func.options_spec or '\n').format, func.argspec, func.desc, func.name)
except SystemExit:
pass

View File

@ -8,6 +8,7 @@ import re
import sys
from collections import namedtuple
from contextlib import contextmanager
from contextlib import suppress
from functools import partial
from . import fast_data_types as defines
@ -47,10 +48,8 @@ def parse_shortcut(sc):
is_native = False
if key is None:
if parts[-1].startswith('0x'):
try:
with suppress(Exception):
key = int(parts[-1], 16)
except Exception:
pass
else:
key = defines.key_for_native_key_name(parts[-1])
is_native = key is not None
@ -578,10 +577,8 @@ def commented_out_default_config():
def prepare_config_file_for_editing():
if not os.path.exists(defconf):
d = os.path.dirname(defconf)
try:
with suppress(FileExistsError):
os.makedirs(d)
except FileExistsError:
pass
with open(defconf, 'w', encoding='utf-8') as f:
f.write(commented_out_default_config())
return defconf

View File

@ -6,6 +6,7 @@ import os
import pwd
import sys
from collections import namedtuple
from contextlib import suppress
appname = 'kitty'
version = (0, 14, 1)
@ -70,10 +71,8 @@ def _get_config_dir():
def cleanup():
import shutil
try:
with suppress(Exception):
shutil.rmtree(ans)
except Exception:
pass
atexit.register(cleanup)
return ans
@ -124,10 +123,8 @@ beam_cursor_data_file = os.path.join(base_dir, 'logo', 'beam-cursor.png')
try:
shell_path = pwd.getpwuid(os.geteuid()).pw_shell or '/bin/sh'
except KeyError:
try:
with suppress(Exception):
print('Failed to read login shell via getpwuid() for current user, falling back to /bin/sh', file=sys.stderr)
except Exception:
pass
shell_path = '/bin/sh'

View File

@ -6,6 +6,7 @@ import locale
import os
import sys
from contextlib import contextmanager
from contextlib import suppress
from .borders import load_borders_program
from .boss import Boss
@ -66,19 +67,15 @@ def talk_to_instance(args):
data = json.dumps(data, ensure_ascii=False).encode('utf-8')
single_instance.socket.sendall(data)
try:
with suppress(EnvironmentError):
single_instance.socket.shutdown(socket.SHUT_RDWR)
except EnvironmentError:
pass
single_instance.socket.close()
if args.wait_for_single_instance_window_close:
conn = notify_socket.accept()[0]
conn.recv(1)
try:
with suppress(EnvironmentError):
conn.shutdown(socket.SHUT_RDWR)
except EnvironmentError:
pass
conn.close()
@ -212,10 +209,8 @@ def setup_environment(opts, args):
def _main():
try:
with suppress(AttributeError): # python compiled without threading
sys.setswitchinterval(1000.0) # we have only a single python thread
except AttributeError:
pass # python compiled without threading
if is_macos:
ensure_macos_locale()
try:

View File

@ -8,6 +8,7 @@ import re
import sys
import types
from functools import partial
from contextlib import suppress
from .cli import emph, parse_args
from .cmds import cmap, parse_subcommand_cli
@ -67,10 +68,8 @@ class SocketIO:
def __exit__(self, *a):
import socket
try:
with suppress(EnvironmentError): # on some OSes such as macOS the socket is already closed at this point
self.socket.shutdown(socket.SHUT_RDWR)
except EnvironmentError:
pass # on some OSes such as macOS the socket is already closed at this point
self.socket.close()
def send(self, data):

5
kitty/rgb.py generated
View File

@ -4,6 +4,7 @@
import re
from collections import namedtuple
from contextlib import suppress
Color = namedtuple('Color', 'red green blue')
@ -62,13 +63,11 @@ def to_color(raw, validate=False):
if ans is not None:
return ans
val = None
try:
with suppress(Exception):
if raw.startswith('#'):
val = parse_sharp(raw[1:])
elif raw[:4].lower() == 'rgb:':
val = parse_rgb(raw[4:])
except Exception:
pass
if val is None and validate:
raise ValueError('Invalid color name: {}'.format(raw))
return val

View File

@ -8,6 +8,7 @@ import shlex
import sys
import traceback
from functools import lru_cache
from contextlib import suppress
from .cli import (
emph, green, italic, parse_option_spec, print_help_for_seq, title
@ -68,10 +69,8 @@ class Completer:
def __init__(self):
self.matches = []
ddir = cache_dir()
try:
with suppress(FileExistsError):
os.makedirs(ddir)
except FileExistsError:
pass
self.history_path = os.path.join(ddir, 'shell.history')
def complete(self, text, state):
@ -86,10 +85,8 @@ class Completer:
return self.matches[state]
def __enter__(self):
try:
with suppress(Exception):
readline.read_history_file(self.history_path)
except Exception:
pass
readline.set_completer(self.complete)
delims = readline.get_completer_delims()
readline.set_completer_delims(delims.replace('-', ''))

View File

@ -5,6 +5,7 @@
import weakref
from collections import deque, namedtuple
from functools import partial
from contextlib import suppress
from .borders import Borders
from .child import Child
@ -28,10 +29,8 @@ def SpecialWindow(cmd, stdin=None, override_title=None, cwd_from=None, cwd=None,
def add_active_id_to_history(items, item_id, maxlen=64):
try:
with suppress(ValueError):
items.remove(item_id)
except ValueError:
pass
items.append(item_id)
if len(items) > maxlen:
items.popleft()

View File

@ -7,6 +7,7 @@ import subprocess
import time
from collections import namedtuple
from urllib.request import urlopen
from contextlib import suppress
from .config import atomic_save
from .constants import cache_dir, get_boss, kitty_exe, version
@ -56,7 +57,7 @@ def parse_line(line):
def read_cache():
notified_versions = {}
try:
with suppress(FileNotFoundError):
with open(version_notification_log()) as f:
for line in f:
try:
@ -64,8 +65,6 @@ def read_cache():
except Exception:
continue
notified_versions[n.version] = n
except FileNotFoundError:
pass
return notified_versions
@ -100,10 +99,8 @@ def run_worker():
import time
import random
time.sleep(random.randint(1000, 4000) / 1000)
try:
with suppress(BrokenPipeError): # happens if parent process is killed before us
print(get_released_version())
except BrokenPipeError:
pass # happens if parent process is killed before us
def update_check(timer_id=None):

View File

@ -11,6 +11,7 @@ import re
import string
import sys
from time import monotonic
from contextlib import suppress
from .constants import (
appname, is_macos, is_wayland, supports_primary_selection
@ -28,19 +29,15 @@ def load_shaders(name):
def safe_print(*a, **k):
try:
with suppress(Exception):
print(*a, **k)
except Exception:
pass
def log_error(*a, **k):
from .fast_data_types import log_error_string
try:
with suppress(Exception):
msg = k.get('sep', ' ').join(map(str, a)) + k.get('end', '')
log_error_string(msg.replace('\0', ''))
except Exception:
pass
def ceil_int(x):
@ -258,15 +255,11 @@ class startup_notification_handler:
def remove_socket_file(s, path=None):
try:
with suppress(EnvironmentError):
s.close()
except EnvironmentError:
pass
if path:
try:
with suppress(EnvironmentError):
os.unlink(path)
except EnvironmentError:
pass
def unix_socket_paths(name, ext='.lock'):

View File

@ -6,6 +6,7 @@ import os
import sys
import zlib
from base64 import standard_b64encode
from contextlib import suppress
write = getattr(sys.stdout, 'buffer', sys.stdout).write
@ -62,10 +63,8 @@ def main():
move_cursor(0, 21)
print('Photo...')
display_png_file(photo)
try:
with suppress(EOFError, KeyboardInterrupt):
input()
except (EOFError, KeyboardInterrupt):
pass
if __name__ == '__main__':

View File

@ -15,6 +15,7 @@ import subprocess
import sys
import tempfile
import time
from contextlib import suppress
import requests
@ -112,10 +113,8 @@ def run_sdist(args):
shutil.copytree(os.path.join(docs_dir, '_build', x), os.path.join(dest, x))
dest = os.path.abspath(os.path.join('build', f'kitty-{version}.tar'))
subprocess.check_call(['tar', '-cf', dest, os.path.basename(base)], cwd=tdir)
try:
with suppress(FileNotFoundError):
os.remove(dest + '.xz')
except FileNotFoundError:
pass
subprocess.check_call(['xz', '-9', dest])

View File

@ -15,6 +15,7 @@ import subprocess
import sys
import sysconfig
import time
from contextlib import suppress
base = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(base, 'glfw'))
@ -279,10 +280,8 @@ def kitty_env():
if '-lz' not in ans.ldpaths:
ans.ldpaths.append('-lz')
try:
with suppress(FileExistsError):
os.mkdir(build_dir)
except FileExistsError:
pass
return ans
@ -424,10 +423,8 @@ def compile_c_extension(kenv, module, incremental, compilation_database, all_key
try:
run_tool([kenv.cc] + linker_cflags + kenv.ldflags + objects + kenv.ldpaths + ['-o', dest], desc='Linking {} ...'.format(emphasis(module)))
except Exception:
try:
with suppress(EnvironmentError):
os.remove(dest)
except EnvironmentError:
pass
else:
os.rename(dest, real_dest)
@ -587,10 +584,8 @@ def build_launcher(args, launcher_dir='.', for_bundle=False, sh_launcher=False,
def copy_man_pages(ddir):
mandir = os.path.join(ddir, 'share', 'man')
safe_makedirs(mandir)
try:
with suppress(FileNotFoundError):
shutil.rmtree(os.path.join(mandir, 'man1'))
except FileNotFoundError:
pass
src = os.path.join(base, 'docs/_build/man')
if not os.path.exists(src):
raise SystemExit('''\
@ -604,10 +599,8 @@ make && make docs
def copy_html_docs(ddir):
htmldir = os.path.join(ddir, 'share', 'doc', appname, 'html')
safe_makedirs(os.path.dirname(htmldir))
try:
with suppress(FileNotFoundError):
shutil.rmtree(htmldir)
except FileNotFoundError:
pass
src = os.path.join(base, 'docs/_build/html')
if not os.path.exists(src):
raise SystemExit('''\

View File

@ -8,6 +8,7 @@ import shutil
import subprocess
import sys
import tempfile
from contextlib import suppress
if False:
tarball = sys.argv[-1]
@ -16,14 +17,10 @@ if False:
os.makedirs(dest)
os.chdir(dest)
dest = os.path.expanduser('~/.local/bin/kitty')
try:
with suppress(EnvironmentError):
os.remove(dest)
except EnvironmentError:
pass
try:
with suppress(EnvironmentError):
os.makedirs(os.path.dirname(dest))
except EnvironmentError:
pass
subprocess.check_call(['tar', 'xJf', tarball])
os.symlink(os.path.abspath('bin/kitty'), dest)
print('kitty installed to ~/.local/kitty.app')