Release v1.4.2

Author: pcisar

docs/changelog.txt

@@ -2,6 +2,13 @@
 Changelog
 #########
 
+Version 1.4.2
+=============
+
+* New `.ServerConfig` options: `~ServerConfig.encoding` and `~ServerConfig.encoding_errors`.
+* New `.connect_server` parameters: `encoding` and `encoding_errors`.
+* Requires firebird-base 1.3.1
+
 Version 1.4.1
 =============
 

docs/conf.py

@@ -23,10 +23,10 @@
 author = 'Pavel Císař'
 
 # The short X.Y version
-version = '1.4.1'
+version = '1.4.2'
 
 # The full version, including alpha/beta/rc tags
-release = '1.4.1'
+release = '1.4.2'
 
 
 # -- General configuration ---------------------------------------------------

firebird/driver/__init__.py

@@ -59,4 +59,4 @@
      Server, Statement
 
 #: Current driver version, SEMVER string.
-__VERSION__ = '1.4.1'
+__VERSION__ = '1.4.2'

firebird/driver/config.py

@@ -71,6 +71,13 @@ def __init__(self, name: str, *, optional: bool=False, description: str=None):
         #: Use trusted authentication, default: False
         self.trusted_auth: BoolOption = \
             BoolOption('trusted_auth', "Use trusted authentication", default=False)
+        #: Encoding used for text data exchange with server
+        self.encoding: StrOption = \
+            StrOption('encoding', "Encoding used for text data exchange with server",
+                      default='ascii')
+        #: Handler used for encoding errors. See `codecs error handlers <codecs>` for details.
+        self.encoding_errors: StrOption = \
+            StrOption('encoding_errors', "Handler used for encoding errors", default='strict')
 
 class DatabaseConfig(Config):
     """Database configuration.

firebird/driver/core.py

@@ -664,8 +664,9 @@ class SPB_ATTACH:
     """
     def __init__(self, *, user: str = None, password: str = None, trusted_auth: bool = False,
                  config: str = None, auth_plugin_list: str = None, expected_db: str=None,
-                 encoding: str='ascii', role: str=None):
+                 encoding: str='ascii', errors: str='strict', role: str=None):
         self.encoding: str = encoding
+        self.errors: str = errors
         self.user: str = user
         self.password: str = password
         self.trusted_auth: bool = trusted_auth
@@ -689,38 +690,43 @@ def parse_buffer(self, buffer: bytes) -> None:
             while not spb.is_eof():
                 tag = spb.get_tag()
                 if tag == SPBItem.CONFIG:
-                    self.config = spb.get_string(encoding=self.encoding)
+                    self.config = spb.get_string(encoding=self.encoding, errors=self.errors)
                 elif tag == SPBItem.AUTH_PLUGIN_LIST:
                     self.auth_plugin_list = spb.get_string()
                 elif tag == SPBItem.TRUSTED_AUTH:
                     self.trusted_auth = True
                 elif tag == SPBItem.USER_NAME:
-                    self.user = spb.get_string(encoding=self.encoding)
+                    self.user = spb.get_string(encoding=self.encoding, errors=self.errors)
                 elif tag == SPBItem.PASSWORD:
-                    self.password = spb.get_string(encoding=self.encoding)
+                    self.password = spb.get_string(encoding=self.encoding, errors=self.errors)
                 elif tag == SPBItem.SQL_ROLE_NAME:
-                    self.role = spb.get_string(encoding=self.encoding)
+                    self.role = spb.get_string(encoding=self.encoding, errors=self.errors)
                 elif tag == SPBItem.EXPECTED_DB:
-                    self.expected_db = spb.get_string(encoding=self.encoding)
+                    self.expected_db = spb.get_string(encoding=self.encoding, errors=self.errors)
     def get_buffer(self) -> bytes:
         """Create SPB_ATTACH from stored information.
         """
         with a.get_api().util.get_xpb_builder(XpbKind.SPB_ATTACH) as spb:
             if self.config is not None:
-                spb.insert_string(SPBItem.CONFIG, self.config, encoding=self.encoding)
+                spb.insert_string(SPBItem.CONFIG, self.config, encoding=self.encoding,
+                                  errors=self.errors)
             if self.trusted_auth:
                 spb.insert_tag(SPBItem.TRUSTED_AUTH)
             else:
                 if self.user is not None:
-                    spb.insert_string(SPBItem.USER_NAME, self.user, encoding=self.encoding)
+                    spb.insert_string(SPBItem.USER_NAME, self.user, encoding=self.encoding,
+                                      errors=self.errors)
                 if self.password is not None:
-                    spb.insert_string(SPBItem.PASSWORD, self.password, encoding=self.encoding)
+                    spb.insert_string(SPBItem.PASSWORD, self.password,
+                                      encoding=self.encoding, errors=self.errors)
             if self.role is not None:
-                spb.insert_string(SPBItem.SQL_ROLE_NAME, self.role, encoding=self.encoding)
+                spb.insert_string(SPBItem.SQL_ROLE_NAME, self.role, encoding=self.encoding,
+                                  errors=self.errors)
             if self.auth_plugin_list is not None:
                 spb.insert_string(SPBItem.AUTH_PLUGIN_LIST, self.auth_plugin_list)
             if self.expected_db is not None:
-                spb.insert_string(SPBItem.EXPECTED_DB, self.expected_db, encoding=self.encoding)
+                spb.insert_string(SPBItem.EXPECTED_DB, self.expected_db,
+                                  encoding=self.encoding, errors=self.errors)
             result = spb.get_buffer()
         return result
 
@@ -5232,7 +5238,8 @@ class Server(LoggingIdMixin):
     Note:
         Implements context manager protocol to call `.close()` automatically.
     """
-    def __init__(self, svc: iService, spb: bytes, host: str, encoding: str):
+    def __init__(self, svc: iService, spb: bytes, host: str, encoding: str,
+                 encoding_errors: str):
         self._svc: iService = svc
         #: Service Parameter Buffer (SPB) used to connect the service manager
         self.spb: bytes = spb
@@ -5244,8 +5251,10 @@ def __init__(self, svc: iService, spb: bytes, host: str, encoding: str):
         self.response: CBuffer = CBuffer(USHRT_MAX)
         self._eof: bool = False
         self.__line_buffer: List[str] = []
-        #: Encoding for string values
+        #: Encoding used for text data exchange with server
         self.encoding: str = encoding
+        #: Handler used for encoding errors. See: `codecs#error-handlers`
+        self.encoding_errors: str = encoding_errors
         #
         self.__ev: float = None
         self.__info: ServerInfoProvider = None
@@ -5308,7 +5317,7 @@ def _read_output(self, *, init: str='', timeout: int=-1) -> None:
         tag = self.response.get_tag()
         if tag != self.mode:  # pragma: no cover
             raise InterfaceError(f"Service responded with error code: {tag}")
-        data = self.response.read_sized_string(encoding=self.encoding)
+        data = self.response.read_sized_string(encoding=self.encoding, errors=self.encoding_errors)
         init += data
         if data and self.mode is SrvInfoCode.LINE:
             init += '\n'
@@ -5438,7 +5447,8 @@ def user(self) -> ServerUserServices:
 
 def connect_server(server: str, *, user: str=None, password: str=None,
                    crypt_callback: iCryptKeyCallbackImpl=None,
-                   expected_db: str=None, role: str=None, encoding: str='ascii') -> Server:
+                   expected_db: str=None, role: str=None, encoding: str=None,
+                   encoding_errors: str=None) -> Server:
     """Establishes a connection to server's service manager.
 
     Arguments:
@@ -5449,7 +5459,10 @@ def connect_server(server: str, *, user: str=None, password: str=None,
         expected_db: Database that would be accessed (for using services with non-default
                      security database)
         role: SQL role used for connection.
-        encoding: Encoding for string values passed in parameter buffer.
+        encoding: Encoding for string values passed in parameter buffer. Default is
+           `.ServerConfig.encoding`.
+        encoding_errors: Error handler used for encoding errors. Default is
+           `.ServerConfig.encoding_errors`.
 
     Hooks:
         Event `.ServerHook.ATTACHED`: Executed before `Service` instance is
@@ -5484,7 +5497,8 @@ def connect_server(server: str, *, user: str=None, password: str=None,
         if crypt_callback is not None:
             provider.set_dbcrypt_callback(crypt_callback)
         svc = provider.attach_service_manager(host, spb_buf)
-    con = Server(svc, spb_buf, host, encoding)
+    con = Server(svc, spb_buf, host, srv_config.encoding.value if encoding is None else encoding,
+                 srv_config.encoding_errors.value if encoding_errors is None else encoding_errors)
     for hook in get_callbacks(ServerHook.ATTACHED, con):
         hook(con)
     return con

firebird/driver/interfaces.py

@@ -1149,9 +1149,9 @@ def insert_bytes(self, tag: int, value: bytes) -> None:
         "Inserts a clumplet with value containing passed bytes."
         self.vtable.insertBytes(self, self.status, tag, value, len(value))
         self._check()
-    def insert_string(self, tag: int, value: str, *, encoding='ascii') -> None:
+    def insert_string(self, tag: int, value: str, *, encoding: str='ascii', errors: str='strict') -> None:
         "Inserts a clumplet with value containing passed string."
-        self.vtable.insertString(self, self.status, tag, value.encode(encoding))
+        self.vtable.insertString(self, self.status, tag, value.encode(encoding, errors))
         self._check()
     def insert_tag(self, tag: int) -> None:
         "Inserts a clumplet without a value."
@@ -1200,11 +1200,11 @@ def get_bigint(self) -> int:
         result = self.vtable.getBigInt(self, self.status)
         self._check()
         return result
-    def get_string(self, *, encoding='ascii') -> str:
+    def get_string(self, *, encoding: str='ascii', errors: str='strict') -> str:
         "Returns value of current clumplet as string."
         result = self.vtable.getString(self, self.status)
         self._check()
-        return string_at(result).decode(encoding)
+        return string_at(result).decode(encoding, errors)
     def get_bytes(self) -> bytes:
         "Returns value of current clumplet as bytes."
         buffer = self.vtable.getBytes(self, self.status)

setup.cfg

@@ -5,7 +5,7 @@ all-files=True
 
 [metadata]
 name = firebird-driver
-version = 1.4.1
+version = 1.4.2
 description = Firebird driver
 long_description = file: README.rst
 long_description_content_type = text/x-rst; charset=UTF-8
@@ -36,7 +36,7 @@ classifiers =
 zip_safe = True
 python_requires = >=3.8, <4
 install_requires =
-    firebird-base>=1.3.0
+    firebird-base>=1.3.1
     python-dateutil>=2.8
 packages = find_namespace:
 

test/test_driver.py

@@ -658,7 +658,9 @@ def setUp(self):
         super().setUp()
         self.dbfile = os.path.join(self.dbpath, self.FBTEST_DB)
         self.db1 = os.path.join(self.dbpath, 'fbtest-1.fdb')
-        cfg = driver_config.register_database('dts-1')
+        cfg = driver_config.get_database('dts-1')
+        if cfg is None:
+            cfg = driver_config.register_database('dts-1')
         cfg.server.value = 'FBTEST_HOST'
         cfg.database.value = self.db1
         cfg.no_linger.value = True
@@ -668,7 +670,9 @@ def setUp(self):
         self.con1.commit()
 
         self.db2 = os.path.join(self.dbpath, 'fbtest-2.fdb')
-        cfg = driver_config.register_database('dts-2')
+        cfg = driver_config.get_database('dts-2')
+        if cfg is None:
+            cfg = driver_config.register_database('dts-2')
         cfg.server.value = 'FBTEST_HOST'
         cfg.database.value = self.db2
         cfg.no_linger.value = True
@@ -1567,7 +1571,7 @@ def test_running(self):
             svc.info.get_log()
             self.assertTrue(svc.is_running())
             # fetch materialized
-            svc.readlines()
+            print(''.join(svc.readlines()))
             self.assertFalse(svc.is_running())
     def test_wait(self):
         with connect_server(FBTEST_HOST, user='SYSDBA', password=FBTEST_PASSWORD) as svc: