Add WriteInReadOnlyReplicationError (#287)

Co-authored-by: Christoph Heer <christoph.heer@sap.com>

CHANGES.rst

@@ -1,6 +1,15 @@
 Changelog
 =========
 
+2.3.0
+-----
+
+Features
+~~~~~~~~
+
+- ``sqlalchemy_hana.errors`` will now raise a ``WriteInReadOnlyReplicationError`` error for error
+  messages indicating a write statement in a read-only replication
+
 2.2.0
 -----
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "sqlalchemy-hana"
-version = "2.2.0"
+version = "2.3.0"
 description = "SQLAlchemy dialect for SAP HANA"
 keywords = ["sqlalchemy", "sap", "hana"]
 requires-python = "~=3.8"

sqlalchemy_hana/errors.py

@@ -36,6 +36,10 @@ class SequenceCacheTimeoutError(HANAError):
     """Exception raised when the sequence cache times out."""
 
 
+class WriteInReadOnlyReplicationError(HANAError):
+    """Exception raised when a write statement is executed in a read only replication."""
+
+
 class LockWaitTimeoutError(HANAError):
     """Exception raised when a lock wait times out."""
 
@@ -168,6 +172,10 @@ def convert_dbapi_error(dbapi_error: DBAPIError) -> DBAPIError:
         return StatementExecutionError.from_dbapi_error(dbapi_error)
     if error.errorcode == 397:
         return InvalidObjectNameError.from_dbapi_error(dbapi_error)
+    if error.errortext.startswith(
+        "feature not supported: writable statement not allowed in read-enabled replication"
+    ):
+        return WriteInReadOnlyReplicationError.from_dbapi_error(dbapi_error)
     return dbapi_error
 
 
@@ -184,4 +192,5 @@ def convert_dbapi_error(dbapi_error: DBAPIError) -> DBAPIError:
     "StatementExecutionError",
     "InvalidObjectNameError",
     "convert_dbapi_error",
+    "WriteInReadOnlyReplicationError",
 )

test/test_errors.py

@@ -20,6 +20,7 @@
     StatementExecutionError,
     StatementTimeoutError,
     TransactionCancelledError,
+    WriteInReadOnlyReplicationError,
     convert_dbapi_error,
 )
 
@@ -81,6 +82,12 @@ def test_convert_dbapi_error_txsavepoint_not_found(self) -> None:
             ),
             (613, "", StatementTimeoutError),
             (397, "", InvalidObjectNameError),
+            (
+                7,
+                "feature not supported: writable statement not allowed in "
+                "read-enabled replication: line 1 col 1",
+                WriteInReadOnlyReplicationError,
+            ),
         ],
     )
     def test_convert_dbapi_error(