Optimizations suggested by refurb

Author: pcisar

firebird/driver/core.py

@@ -142,10 +142,12 @@ def _encode_timestamp(v: Union[datetime.datetime, datetime.date]) -> bytes:
 
 def _is_fixed_point(dialect: int, datatype: SQLDataType, subtype: int,
                     scale: int) -> bool:
-    return ((datatype in [SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64]
-             and (subtype or scale)) or
+    return ((datatype in (SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64)
+             and (subtype or scale))
+            or
             ((dialect < 3) and scale
-             and (datatype in [SQLDataType.DOUBLE, SQLDataType.D_FLOAT])))
+             and (datatype in (SQLDataType.DOUBLE, SQLDataType.D_FLOAT)))
+            )
 
 def _get_external_data_type_name(dialect: int, datatype: SQLDataType,
                                  subtype: int, scale: int) -> str:
@@ -168,7 +170,7 @@ def _get_external_data_type_name(dialect: int, datatype: SQLDataType,
         return 'BIGINT'
     elif datatype == SQLDataType.FLOAT:
         return 'FLOAT'
-    elif datatype in [SQLDataType.DOUBLE, SQLDataType.D_FLOAT]:
+    elif datatype in (SQLDataType.DOUBLE, SQLDataType.D_FLOAT):
         return 'DOUBLE'
     elif datatype == SQLDataType.TIMESTAMP:
         return 'TIMESTAMP'
@@ -184,7 +186,7 @@ def _get_external_data_type_name(dialect: int, datatype: SQLDataType,
         return 'UNKNOWN'
 
 def _get_internal_data_type_name(data_type: SQLDataType) -> str:
-    if data_type in [SQLDataType.DOUBLE, SQLDataType.D_FLOAT]:
+    if data_type in (SQLDataType.DOUBLE, SQLDataType.D_FLOAT):
         value = SQLDataType.DOUBLE
     else:
         value = data_type
@@ -215,7 +217,7 @@ def _check_integer_range(value: int, dialect: int, datatype: SQLDataType,
 
 def _is_str_param(value: Any, datatype: SQLDataType) -> bool:
     return ((isinstance(value, str) and datatype != SQLDataType.BLOB) or
-            datatype in [SQLDataType.TEXT, SQLDataType.VARYING])
+            datatype in (SQLDataType.TEXT, SQLDataType.VARYING))
 
 def create_meta_descriptors(meta: iMessageMetadata) -> List[ItemMetadata]:
     result = []
@@ -364,7 +366,7 @@ def get_buffer(self) -> bytes:
             isolation = (Isolation.READ_COMMITTED_RECORD_VERSION
                          if self.isolation == Isolation.READ_COMMITTED
                          else self.isolation)
-            if isolation in [Isolation.SNAPSHOT, Isolation.SERIALIZABLE]:
+            if isolation in (Isolation.SNAPSHOT, Isolation.SERIALIZABLE):
                 tpb.insert_tag(isolation)
             elif isolation == Isolation.READ_COMMITTED_READ_CONSISTENCY:
                 tpb.insert_tag(TPBItem.READ_CONSISTENCY)
@@ -777,7 +779,7 @@ def callback(result, length, updated):
         self.__queue: PriorityQueue = weakref.proxy(queue)
         self._db_handle: a.FB_API_HANDLE = db_handle
         self._isc_status: a.ISC_STATUS_ARRAY = a.ISC_STATUS_ARRAY(0)
-        self.event_names: List[str] = list(event_names)
+        self.event_names: List[str] = event_names
 
         self.__results: a.RESULT_VECTOR = a.RESULT_VECTOR(0)
         self.__closed: bool = False
@@ -1700,7 +1702,7 @@ def _determine_field_precision(self, meta: ItemMetadata) -> int:
             # for example for queries with dynamically computed fields
             return 0
         # Special case for automatic RDB$DB_KEY fields.
-        if (meta.field in ['DB_KEY', 'RDB$DB_KEY']):
+        if (meta.field in ('DB_KEY', 'RDB$DB_KEY')):
             return 0
         precision = self.__precision_cache.get((meta.relation, meta.field))
         if precision is not None:
@@ -1860,8 +1862,7 @@ def transaction_manager(self, default_tpb: bytes=None,
             default_action: Default action to be performed on implicit transaction end.
         """
         assert self._att is not None
-        transaction = TransactionManager(self, default_tpb if default_tpb else self.default_tpb,
-                                         default_action)
+        transaction = TransactionManager(self, default_tpb or self.default_tpb, default_action)
         self._transactions.append(transaction)
         return transaction
     def begin(self, tpb: bytes=None) -> None:
@@ -2241,9 +2242,8 @@ def __isolation(self) -> Isolation:
         if cnt == 1:
             # The value is `TraInfoIsolation` that maps to `Isolation`
             return Isolation(self.response.read_byte())
-        else:
-            # The values are `TraInfoIsolation` + `TraInfoReadCommitted` that maps to `Isolation`
-            return Isolation(self.response.read_byte() + self.response.read_byte())
+        # The values are `TraInfoIsolation` + `TraInfoReadCommitted` that maps to `Isolation`
+        return Isolation(self.response.read_byte() + self.response.read_byte())
     def __access(self) -> TraInfoAccess:
         return TraInfoAccess(self.response.read_sized_int())
     def __lock_timeout(self) -> int:
@@ -2421,7 +2421,7 @@ def begin(self, tpb: bytes=None) -> None:
         """
         assert not self.__closed
         self._finish()  # Make sure that previous transaction (if any) is ended
-        self._tra = self._connection()._att.start_transaction(tpb if tpb else self.default_tpb)
+        self._tra = self._connection()._att.start_transaction(tpb or self.default_tpb)
     def commit(self, *, retaining: bool=False) -> None:
         """Commits the transaction managed by this instance.
 
@@ -2561,7 +2561,7 @@ def begin(self, tpb: bytes=None) -> None:
         self._finish()  # Make sure that previous transaction (if any) is ended
         with self._dtc.start_builder() as builder:
             for con in self._connections:
-                builder.add_with_tpb(con._att, tpb if tpb else self.default_tpb)
+                builder.add_with_tpb(con._att, tpb or self.default_tpb)
             self._tra = builder.start()
     def prepare(self) -> None:
         """Manually triggers the first phase of a two-phase commit (2PC).
@@ -3296,12 +3296,12 @@ def _pack_input(self, meta: iMessageMetadata, buffer: bytes,
                         value = str(value)
                     if isinstance(value, str) and self._encoding:
                         value = value.encode(self._encoding)
-                    if (datatype in [SQLDataType.TEXT, SQLDataType.VARYING]
+                    if (datatype in (SQLDataType.TEXT, SQLDataType.VARYING)
                         and len(value) > length):
                         raise ValueError(f"Value of parameter ({i}) is too long,"
                                          f" expected {length}, found {len(value)}")
                     memmove(buf_addr + offset, value, len(value))
-                elif datatype in [SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64]:
+                elif datatype in (SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64):
                     # It's scalled integer?
                     scale = in_meta.get_scale(i)
                     if in_meta.get_subtype(i) or scale:
@@ -3471,7 +3471,7 @@ def _unpack_output(self) -> Tuple:
                         value = value.decode(self._encoding)
                 elif datatype == SQLDataType.BOOLEAN:
                     value = bool((0).from_bytes(buffer[offset], 'little'))
-                elif datatype in [SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64]:
+                elif datatype in (SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64):
                     value = (0).from_bytes(buffer[offset:offset + length], 'little', signed=True)
                     # It's scalled integer?
                     if desc.subtype or desc.scale:
@@ -3832,8 +3832,7 @@ def fetch_next(self) -> Optional[Tuple]:
         self._last_fetch_status = self._result.fetch_next(self._stmt._out_buffer)
         if self._last_fetch_status == StateResult.OK:
             return self._unpack_output()
-        else:
-            return None
+        return None
     def fetch_prior(self) -> Optional[Tuple]:
         """Fetch the previous row of a scrollable query result set.
 
@@ -3843,8 +3842,7 @@ def fetch_prior(self) -> Optional[Tuple]:
         self._last_fetch_status = self._result.fetch_prior(self._stmt._out_buffer)
         if self._last_fetch_status == StateResult.OK:
             return self._unpack_output()
-        else:
-            return None
+        return None
     def fetch_first(self) -> Optional[Tuple]:
         """Fetch the first row of a scrollable query result set.
 
@@ -3854,8 +3852,7 @@ def fetch_first(self) -> Optional[Tuple]:
         self._last_fetch_status = self._result.fetch_first(self._stmt._out_buffer)
         if self._last_fetch_status == StateResult.OK:
             return self._unpack_output()
-        else:
-            return None
+        return None
     def fetch_last(self) -> Optional[Tuple]:
         """Fetch the last row of a scrollable query result set.
 
@@ -3865,8 +3862,7 @@ def fetch_last(self) -> Optional[Tuple]:
         self._last_fetch_status = self._result.fetch_last(self._stmt._out_buffer)
         if self._last_fetch_status == StateResult.OK:
             return self._unpack_output()
-        else:
-            return None
+        return None
     def fetch_absolute(self, position: int) -> Optional[Tuple]:
         """Fetch the row of a scrollable query result set specified by absolute position.
 
@@ -3879,8 +3875,7 @@ def fetch_absolute(self, position: int) -> Optional[Tuple]:
         self._last_fetch_status = self._result.fetch_absolute(position, self._stmt._out_buffer)
         if self._last_fetch_status == StateResult.OK:
             return self._unpack_output()
-        else:
-            return None
+        return None
     def fetch_relative(self, offset: int) -> Optional[Tuple]:
         """Fetch the row of a scrollable query result set specified by relative position.
 
@@ -3894,8 +3889,7 @@ def fetch_relative(self, offset: int) -> Optional[Tuple]:
         self._last_fetch_status = self._result.fetch_relative(offset, self._stmt._out_buffer)
         if self._last_fetch_status == StateResult.OK:
             return self._unpack_output()
-        else:
-            return None
+        return None
     def setinputsizes(self, sizes: Sequence[Type]) -> None:
         """Required by Python DB API 2.0, but pointless for Firebird, so it does nothing.
         """
@@ -3946,15 +3940,15 @@ def description(self) -> Tuple[DESCRIPTION]:
             for meta in self._stmt._out_desc:
                 scale = meta.scale
                 precision = 0
-                if meta.datatype in [SQLDataType.TEXT, SQLDataType.VARYING]:
+                if meta.datatype in (SQLDataType.TEXT, SQLDataType.VARYING):
                     vtype = str
                     if meta.subtype in (4, 69):  # UTF8 and GB18030
                         dispsize = meta.length // 4
                     elif meta.subtype == 3:  # UNICODE_FSS
                         dispsize = meta.length // 3
                     else:
                         dispsize = meta.length
-                elif (meta.datatype in [SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64]
+                elif (meta.datatype in (SQLDataType.SHORT, SQLDataType.LONG, SQLDataType.INT64)
                       and (meta.subtype or meta.scale)):
                     vtype = decimal.Decimal
                     precision = self._connection._determine_field_precision(meta)
@@ -3968,7 +3962,7 @@ def description(self) -> Tuple[DESCRIPTION]:
                 elif meta.datatype == SQLDataType.INT64:
                     vtype = int
                     dispsize = 20
-                elif meta.datatype in [SQLDataType.FLOAT, SQLDataType.D_FLOAT, SQLDataType.DOUBLE]:
+                elif meta.datatype in (SQLDataType.FLOAT, SQLDataType.D_FLOAT, SQLDataType.DOUBLE):
                     # Special case, dialect 1 DOUBLE/FLOAT
                     # could be Fixed point
                     if (self._stmt._dialect < 3) and meta.scale:
@@ -4023,10 +4017,10 @@ def affected_rows(self) -> int:
         if self._stmt is None:
             return -1
         result = -1
-        if (self._executed and self._stmt.type in [StatementType.SELECT,
+        if (self._executed and self._stmt.type in (StatementType.SELECT,
                                                    StatementType.INSERT,
                                                    StatementType.UPDATE,
-                                                   StatementType.DELETE]):
+                                                   StatementType.DELETE)):
             info = create_string_buffer(64)
             self._stmt._istmt.get_info(bytes([23, 1]), info) # bytes(isc_info_sql_records, isc_info_end)
             if ord(info[0]) != 23:  # pragma: no cover
@@ -4512,7 +4506,7 @@ def local_restore(self, *, backup_stream: BinaryIO,
                 else:  # pragma: no cover
                     raise InterfaceError(f"Service responded with error code: {tag}")
                 tag = self._srv().response.get_tag()
-            keep_going = no_data or request_length != 0 or len(line) > 0
+            keep_going = no_data or request_length != 0 or line
     def nbackup(self, *, database: FILESPEC, backup: FILESPEC, level: int=0,
                 direct: bool=None, flags: SrvNBackupFlag=SrvNBackupFlag.NONE,
                 role: str=None, guid: str=None) -> None:
@@ -5301,10 +5295,9 @@ def _reset_output(self) -> None:
     def _make_request(self, timeout: int) -> bytes:
         if timeout == -1:
             return None
-        else:
-            return b''.join([SrvInfoCode.TIMEOUT.to_bytes(1, 'little'),
-                             (4).to_bytes(2, 'little'),
-                             timeout.to_bytes(4, 'little'), isc_info_end.to_bytes(1, 'little')])
+        return b''.join([SrvInfoCode.TIMEOUT.to_bytes(1, 'little'),
+                         (4).to_bytes(2, 'little'),
+                         timeout.to_bytes(4, 'little'), isc_info_end.to_bytes(1, 'little')])
     def _fetch_complex_info(self, request: bytes, timeout: int=-1) -> None:
         send = self._make_request(timeout)
         self.response.clear()
@@ -5494,7 +5487,7 @@ def connect_server(server: str, *, user: str=None, password: str=None,
     srv_config = driver_config.get_server(server)
     if srv_config is None:
         srv_config = driver_config.server_defaults
-        host = server if server else None
+        host = server or None
     else:
         host = srv_config.host.value
     if host is None:

firebird/driver/fbapi.py

@@ -43,6 +43,7 @@
 from ctypes.util import find_library
 from locale import getpreferredencoding
 from pathlib import Path
+from contextlib import suppress
 import platform
 from .config import driver_config
 from .hooks import APIHook, register_class, get_callbacks
@@ -1932,11 +1933,9 @@ def __init__(self, filename: Path = None):
             else:
                 filename = find_library('fbclient')
                 if not filename:
-                    try:
+                    with suppress(Exception):
                         ctypes.CDLL('libfbclient.so')
                         filename = 'libfbclient.so'
-                    except Exception:
-                        pass
             if not filename:
                 raise Exception("The location of Firebird Client Library could not be determined.")
         elif not filename.exists():
@@ -1945,7 +1944,7 @@ def __init__(self, filename: Path = None):
                 raise Exception(f"Firebird Client Library '{filename}' not found")
             filename = file_name
         self.client_library: ctypes.CDLL = None
-        if sys.platform in ['win32', 'cygwin', 'os2', 'os2emx']:
+        if sys.platform in ('win32', 'cygwin', 'os2', 'os2emx'):
             self.client_library: ctypes.CDLL = ctypes.WinDLL(str(filename))
         else:
             self.client_library: ctypes.CDLL = ctypes.CDLL(str(filename))

firebird/driver/interfaces.py

@@ -40,6 +40,7 @@
 import threading
 import datetime
 from warnings import warn
+from contextlib import suppress
 from ctypes import memmove, memset, create_string_buffer, cast, byref, string_at, sizeof, \
      c_char_p, c_void_p, c_byte, c_ulong
 from .types import Error, DatabaseError, InterfaceError, FirebirdWarning, BCD, \
@@ -1657,10 +1658,8 @@ def _get_intf(self):
                 a.IVersionCallback_struct,
                 a.IVersionCallback)
     def __callback(self, this: a.IVersionCallback, status: a.IStatus, text: c_char_p):
-        try:
+        with suppress(Exception):
             self.callback(text.decode())
-        except Exception:
-            pass
     def callback(self, text: str) -> None:
         "Method called by engine"
 
@@ -1677,13 +1676,11 @@ def _get_intf(self):
                 a.ICryptKeyCallback)
     def __callback(self, this: a.ICryptKeyCallback, data_length: a.Cardinal, data: c_void_p,
                    buffer_length: a.Cardinal, buffer: c_void_p) -> a.Cardinal:
-        try:
+        with suppress(Exception):
             key = self.get_crypt_key(data[:data_length], buffer_length)
             key_size = min(len(key), buffer_length)
             memmove(buffer, key, key_size)
             return key_size
-        except Exception:
-            pass
     def get_crypt_key(self, data: bytes, max_key_size: int) -> bytes:
         "Should return crypt key"
         return b''
@@ -1701,10 +1698,8 @@ def _get_intf(self):
                 a.IOffsetsCallback)
     def __callback(self, this: a.IOffsetsCallback, status: a.IStatus, index: a.Cardinal,
                    offset: a.Cardinal, nullOffset: a.Cardinal) -> None:
-        try:
+        with suppress(Exception):
             self.set_offset(index, offset, nullOffset)
-        except Exception:
-            pass
     def set_offset(self, index: int, offset: int, nullOffset: int) -> None:
         "Method called by engine"
 
@@ -1720,10 +1715,8 @@ def _get_intf(self):
                 a.IEventCallback_struct,
                 a.IEventCallback)
     def __callback(self, this: a.IVersionCallback, length: a.Cardinal, events: a.BytePtr) -> None:
-        try:
+        with suppress(Exception):
             self.events_arrived(string_at(events, length))
-        except Exception:
-            pass
     def events_arrived(self, events: bytes) -> None:
         "Method called by engine"
 
@@ -1736,10 +1729,8 @@ def __init__(self):
     def _get_intf(self):
         return (a.ITimer_VTable, a.ITimer_VTablePtr, a.ITimer_struct, a.ITimer)
     def __callback(self, this: a.ITimer) -> None:
-        try:
+        with suppress(Exception):
             self.handler()
-        except Exception:
-            pass
     def handler(self) -> None:
         "Timer callback handler"
 

firebird/driver/types.py

@@ -61,7 +61,7 @@ class DatabaseError(Error):
     #: Returned SQLCODE or None
     sqlcode: int = None
     #: Tuple with all returned GDS error codes
-    gds_codes: Tuple[int] = tuple()
+    gds_codes: Tuple[int] = ()
 
 class DataError(DatabaseError):
     """Exception raised for errors that are due to problems with the processed
@@ -1392,7 +1392,7 @@ def get_timezone(timezone: str=None) -> datetime.tzinfo:
     database instead zoned time, and to handle offset-based timezones in format required by
     Firebird.
     """
-    if timezone[0] in ['+', '-']:
+    if timezone[0] in ('+', '-'):
         timezone = 'UTC' + timezone
     result = tz.gettz(timezone)
     if result is not None and not hasattr(result, '_timezone_'):