chore: raise domain specific errors for sql module

Raise domain specific errors on the SQL module. This includes raising
`TransientError`s whenever an operation is retriable.

Also remove the unnecessary factory functions for the sql draw streams
and use appropriate factory class methods.

Included also are other minor fixes and improvements.

app/mods/common/src/sghi/idr/client/common/domain/etl_protocol.py

@@ -134,21 +134,21 @@ def _proto_definition_to_proto_instance(
             dmf_def,
         )
         return SimpleETLProtocol(
-            id=protocol_definition["id"],
-            name=protocol_definition["name"],
-            description=protocol_definition.get("description"),
-            data_sink_factory=protocol_definition["data_sink_factory"],
-            data_source_factory=protocol_definition["data_source_factory"],
-            data_processor_factory=protocol_definition[
+            id=protocol_definition["id"],  # pyright: ignore
+            name=protocol_definition["name"],  # pyright: ignore
+            description=protocol_definition.get("description"),  # pyright: ignore  # noqa: E501
+            data_sink_factory=protocol_definition["data_sink_factory"],  # pyright: ignore  # noqa: E501
+            data_source_factory=protocol_definition["data_source_factory"],  # pyright: ignore  # noqa: E501
+            data_processor_factory=protocol_definition[  # pyright: ignore
                 "data_processor_factory"
             ],
-            metadata_consumer=protocol_definition[
+            metadata_consumer=protocol_definition[  # pyright: ignore
                 "metadata_consumer_factory"
             ](),
-            metadata_supplier=protocol_definition[
+            metadata_supplier=protocol_definition[  # pyright: ignore
                 "metadata_supplier_factory"
             ](),
-            drain_metadata_factory=dmf,
+            drain_metadata_factory=dmf,  # pyright: ignore
         )
 
     @staticmethod
@@ -161,7 +161,7 @@ def _get_data_meta_factory_instance(
             case Callable():
                 return dmf_def()
             case _:
-                assert_never(dmf_def)
+                assert_never(dmf_def)  # pyright: ignore
 
 
 class FromFactoriesETLProtocolSupplier(ETLProtocolSupplier):

app/mods/common/src/sghi/idr/client/common/lib/config/etl_protocols_config.py

@@ -76,10 +76,10 @@ class ProtocolDefinition(_ProtocolDefinitionOptional, total=True):
 @cache
 def _get_required_proto_definition_fields() -> set[str]:
     all_fields: set[str] = set(
-        typed_dict_keys(_RawProtocolDefinition).keys(),
+        typed_dict_keys(_RawProtocolDefinition).keys(),  # type: ignore
     )
     optional_fields: set[str] = set(
-        typed_dict_keys(_ProtocolDefinitionOptional).keys(),
+        typed_dict_keys(_ProtocolDefinitionOptional).keys(),  # type: ignore
     )
     return all_fields.difference(optional_fields)
 

app/mods/http/src/sghi/idr/client/http/exceptions.py

@@ -2,14 +2,14 @@
 
 
 class HTTPTransportError(IDRClientError):
-    """Unknown error occurred while performing a HTTP Transport operation."""
+    """Unknown error occurred while performing an HTTP Transport operation."""
 
     ...
 
 
 class HTTPTransportTransientError(HTTPTransportError, TransientError):
     """
-    Recoverable error occurred while performing a HTTP Transport operation.
+    Recoverable error occurred while performing an HTTP Transport operation.
     """
 
     ...

app/mods/http/src/sghi/idr/client/http/lib/common.py

@@ -1,3 +1,5 @@
+from collections.abc import Sequence
+
 from requests.exceptions import (
     ConnectionError,
     RequestException,
@@ -9,6 +11,22 @@
 from ..exceptions import HTTPTransportError, HTTPTransportTransientError
 from ..typings import ResponsePredicate
 
+# =============================================================================
+# CONSTANTS
+# =============================================================================
+
+KNOWN_REQUESTS_TRANSIENT_BASE_ERRORS: Sequence[type[BaseException]] = (
+    ConnectionError,
+    RequestException,
+    Timeout,
+    TooManyRedirects,
+)
+
+
+# =============================================================================
+# UTILITIES
+# =============================================================================
+
 
 def if_response_has_status_factory(
     *http_status: int,
@@ -63,23 +81,26 @@ def if_request_accepted(response: Response) -> bool:
 
 
 def to_appropriate_domain_error(
-    exp: RequestException,
+    exp: BaseException,
     message: str | None = None,
+    known_transient_errors: tuple[type[BaseException]] = KNOWN_REQUESTS_TRANSIENT_BASE_ERRORS,  # noqa: E501
 ) -> HTTPTransportError:
-    """Map a :exp:`RequestException` to the appropriate domain error.
+    """Map an exception to the appropriate domain error.
 
-    Given a `RequestException`, either map the error to an
+    Given an exception, either map the error to an
     :exp:`HTTPTransportTransientError` if the error is retryable, or else
     map it to a :exp:`HTTPTransportError`.
 
-    :param exp: A `RequestException` to be mapped to the appropriate domain
-        specific error.
+    :param exp: An exception to be mapped to the appropriate domain specific
+        error.
     :param message: An optional error message to pass to the returned
         exception.
+    :param known_transient_errors: A tuple of exception types that are safe to
+        retry.
 
     :return: An appropriate domain specific error based on the given
-        `RequestException`.
+        exception.
     """
-    if isinstance(exp, ConnectionError | Timeout | TooManyRedirects):
+    if isinstance(exp, known_transient_errors):
         return HTTPTransportTransientError(message=message)
     return HTTPTransportError(message=message)

app/mods/idr_server/src/sghi/idr/client/idr_server/__init__.py

@@ -44,15 +44,15 @@
 
 def fyj_cbs_etl_protocol_factory() -> FYJCBSETLProtocol:
     return SimpleETLProtocol(
-        id="fyj-cbs",
-        name="FyJ CBS ETL Protocol",
-        description="Fahari ya Jamii, CBS ETL Protocol",
-        data_sink_factory=HTTPDataSink.from_data_sink_meta,
-        data_source_factory=SimpleSQLDatabase.from_data_source_meta,
-        data_processor_factory=IDRServerDataProcessor,
-        drain_metadata_factory=fyj_cbs_drain_meta_factory(),
-        metadata_consumer=fyj_cbs_metadata_consumer_factory(),
-        metadata_supplier=fyj_cbs_metadata_supplier_factory(),
+        id="fyj-cbs",  # pyright: ignore
+        name="FyJ CBS ETL Protocol",  # pyright: ignore
+        description="Fahari ya Jamii, CBS ETL Protocol",  # pyright: ignore
+        data_sink_factory=HTTPDataSink.from_data_sink_meta,  # pyright: ignore
+        data_source_factory=SimpleSQLDatabase.from_data_source_meta,  # pyright: ignore  # noqa: E501
+        data_processor_factory=IDRServerDataProcessor,  # pyright: ignore
+        drain_metadata_factory=fyj_cbs_drain_meta_factory(),  # pyright: ignore
+        metadata_consumer=fyj_cbs_metadata_consumer_factory(),  # pyright: ignore  # noqa: E501
+        metadata_supplier=fyj_cbs_metadata_supplier_factory(),  # pyright: ignore  # noqa: E501
     )
 
 

app/mods/idr_server/src/sghi/idr/client/idr_server/lib/idr_sever_v1_api.py

@@ -22,9 +22,9 @@
 )
 from sghi.idr.client.http.lib.http_transport import HTTPTransport
 from sghi.idr.client.sql.domain import (
+    PDDataFrameDataSourceStream,
     SimpleSQLDatabaseDescriptor,
     SimpleSQLQuery,
-    pd_data_frame_data_source_stream_factory,
 )
 from toolz import pipe
 from toolz.curried import map
@@ -328,7 +328,7 @@ def handle_get_data_source_meta_response(
                             _result["database_name"],
                         ),
                         "isolation_level": "REPEATABLE READ",
-                        "data_source_stream_factory": pd_data_frame_data_source_stream_factory,  # noqa: E501
+                        "data_source_stream_factory": PDDataFrameDataSourceStream.of,  # noqa: E501
                     },
                 ),
                 map(lambda _kwargs: SimpleSQLDatabaseDescriptor(**_kwargs)),

app/mods/sql/src/sghi/idr/client/sql/domain/__init__.py

@@ -12,8 +12,6 @@
     SimpleSQLDatabase,
     SimpleSQLDataSourceStream,
     SQLRawData,
-    pd_data_frame_data_source_stream_factory,
-    simple_data_source_stream_factory,
 )
 
 __all__ = [
@@ -28,6 +26,4 @@
     "SimpleSQLDatabaseDescriptor",
     "SimpleSQLDataSourceStream",
     "SimpleSQLQuery",
-    "pd_data_frame_data_source_stream_factory",
-    "simple_data_source_stream_factory",
 ]

app/mods/sql/src/sghi/idr/client/sql/domain/metadata.py

@@ -101,12 +101,9 @@ class SimpleSQLDatabaseDescriptor(BaseSQLDataSourceMetadata):
 
     @property
     def data_source_stream_factory(self) -> DataSourceStreamFactory:
-        from .operations import pd_data_frame_data_source_stream_factory
+        from .operations import SimpleSQLDataSourceStream
 
-        return (
-            self._data_source_stream_factory
-            or pd_data_frame_data_source_stream_factory
-        )
+        return self._data_source_stream_factory or SimpleSQLDataSourceStream.of
 
     @property
     def database_url(self) -> str | URL:

app/mods/sql/src/sghi/idr/client/sql/domain/operations.py

@@ -15,7 +15,9 @@
 )
 from sghi.idr.client.core.lib import type_fqn
 from sqlalchemy import Connection, CursorResult, Engine, Row, create_engine
+from sqlalchemy.exc import SQLAlchemyError
 
+from ..lib import to_appropriate_domain_error
 from ..typings import ReadIsolationLevels
 from .metadata import (
     BaseSQLDataSourceMetadata,
@@ -48,37 +50,6 @@
 """The maximum number of rows to extract from a database at any one time."""
 
 
-# =============================================================================
-# HELPERS
-# =============================================================================
-
-
-def pd_data_frame_data_source_stream_factory(
-    sql_data_source: "BaseSQLDataSource[Any, SimpleSQLQuery, PDDataFrame]",
-    extract_metadata: SimpleSQLQuery,
-) -> "PDDataFrameDataSourceStream":
-    # TODO: Add a check to ensure that the `sql_data_source` given is not
-    #  disposed.
-    return PDDataFrameDataSourceStream(
-        sql_data_source,
-        extract_metadata,
-        sql_data_source.engine.connect(),
-    )
-
-
-def simple_data_source_stream_factory(
-    sql_data_source: "BaseSQLDataSource[Any, SimpleSQLQuery, SQLRawData]",
-    extract_metadata: SimpleSQLQuery,
-) -> "SimpleSQLDataSourceStream":
-    # TODO: Add a check to ensure that the `sql_data_source` given is not
-    #  disposed.
-    return SimpleSQLDataSourceStream(
-        sql_data_source,
-        extract_metadata,
-        sql_data_source.engine.connect(),
-    )
-
-
 # =============================================================================
 # BASE OPERATIONS CLASSES
 # =============================================================================
@@ -92,14 +63,11 @@ class BaseSQLDataSource(
 ):
     """An SQL Database."""
 
+    _engine: Engine = field()
     _data_source_stream_factory: Callable[
         ["BaseSQLDataSource[Any, _EM, _RD]", _EM],
         DataSourceStream[_EM, _RD],
-    ] = field(
-        default=simple_data_source_stream_factory,
-        kw_only=True,
-    )
-    _engine: Engine = field()
+    ] = field()
 
     def __attrs_post_init__(self) -> None:
         self._logger: Logger = logging.getLogger(type_fqn(self.__class__))
@@ -209,7 +177,7 @@ def of_sqlite_in_memory(
         data_source_stream_factory: Callable[
             ["SimpleSQLDatabase[Any]", SimpleSQLQuery],
             DataSourceStream[SimpleSQLQuery, Any],
-        ] = simple_data_source_stream_factory,
+        ] | None = None,
     ) -> Self:
         return cls(
             name=name,  # pyright: ignore
@@ -275,12 +243,34 @@ def draw(self) -> tuple[PDDataFrame, float]:
                 self.draw_metadata.name,
             )
             raise DataSourceStream.StopDraw from None
+        except SQLAlchemyError as exp:
+            _err_msg: str = "Error while drawing from sql source."
+            self._logger.exception(_err_msg)
+            raise to_appropriate_domain_error(exp, message=_err_msg) from exp
 
     def dispose(self) -> None:
         self._is_disposed = True
         self._connection.close()
         self._logger.debug("Disposal complete.")
 
+    @classmethod
+    def of(
+        cls,
+        sql_data_source: BaseSQLDataSource[Any, SimpleSQLQuery, PDDataFrame],
+        draw_metadata: SimpleSQLQuery,
+    ) -> Self:
+        try:
+            return cls(
+                data_source=sql_data_source,  # pyright: ignore
+                draw_metadata=draw_metadata,  # pyright: ignore
+                connection=sql_data_source.engine.connect(),  # pyright: ignore
+            )
+        except SQLAlchemyError as exp:
+            _err_msg: str = (
+                "Unable to initialize 'PDDataFrameDataSourceStream'."
+            )
+            raise to_appropriate_domain_error(exp, message=_err_msg) from exp
+
 
 @define(order=False, slots=True)
 class SimpleSQLDataSourceStream(
@@ -334,9 +324,29 @@ def draw(self) -> tuple[SQLRawData, float]:
                 self.draw_metadata.name,
             )
             raise DataSourceStream.StopDraw from None
+        except SQLAlchemyError as exp:
+            _err_msg: str = "Error while drawing from sql source."
+            self._logger.exception(_err_msg)
+            raise to_appropriate_domain_error(exp, message=_err_msg) from exp
 
     def dispose(self) -> None:
         self._is_disposed = True
         self._extraction_result.close()
         self._connection.close()
         self._logger.debug("Disposal complete.")
+
+    @classmethod
+    def of(
+        cls,
+        sql_data_source: BaseSQLDataSource[Any, SimpleSQLQuery, PDDataFrame],
+        draw_metadata: SimpleSQLQuery,
+    ) -> Self:
+        try:
+            return cls(
+                data_source=sql_data_source,  # pyright: ignore
+                draw_metadata=draw_metadata,  # pyright: ignore
+                connection=sql_data_source.engine.connect(),  # pyright: ignore
+            )
+        except SQLAlchemyError as exp:
+            _err_msg: str = "Unable to initialize 'SimpleSQLDataSourceStream'."
+            raise to_appropriate_domain_error(exp, message=_err_msg) from exp

app/mods/sql/src/sghi/idr/client/sql/exceptions.py

@@ -0,0 +1,11 @@
+from sghi.idr.client.core.exceptions import IDRClientError, TransientError
+
+
+class SQLError(IDRClientError):
+    """An ambiguous error occurred while operating on SQL resources."""
+    ...
+
+
+class SQLTransientError(SQLError, TransientError):
+    """A recoverable error occurred while operating on an SQL resource."""
+    ...

app/mods/sql/src/sghi/idr/client/sql/lib/__init__.py

@@ -0,0 +1,8 @@
+from .common import to_appropriate_domain_error
+from .config import DBInstanceConfig, DBInstancesInitializer
+
+__all__ = [
+    "DBInstanceConfig",
+    "DBInstancesInitializer",
+    "to_appropriate_domain_error",
+]