BI-5486: Add data type of conn-icons (#491)

* Add data type of conn-icons

* Add logging

* Add error handling

* Add error handling

* Issue fixes

* Issue fixes

lib/dl_api_connector/dl_api_connector/connection_info.py

@@ -1,20 +1,49 @@
+import base64
+from importlib.abc import Traversable
+from importlib.resources import as_file
+import logging
 from typing import (
     ClassVar,
     Optional,
     final,
 )
 
+import attr
+
 from dl_i18n.localizer_base import (
     Localizer,
     Translatable,
 )
 
 
+LOGGER = logging.getLogger(__name__)
+
+
+@attr.s
 class ConnectionInfoProvider:
     title_translatable: ClassVar[Translatable]
     alias: ClassVar[Optional[str]] = None
+    icon_data_standard: Optional[bytes] = attr.ib(default=None)
+    icon_data_nav: Optional[bytes] = attr.ib(default=None)
+    icon_data_standard_filepath: Optional[Traversable] = attr.ib(default=None)
+    icon_data_nav_filepath: Optional[Traversable] = attr.ib(default=None)
 
     @classmethod
     @final
     def get_title(cls, localizer: Localizer) -> str:
         return localizer.translate(cls.title_translatable)
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard = self.get_icon_file(self.icon_data_standard_filepath)
+        self.icon_data_nav = self.get_icon_file(self.icon_data_nav_filepath)
+
+    def get_icon_file(self, filepath: Traversable | None) -> bytes | None:
+        if not filepath:
+            return None
+        try:
+            with as_file(filepath) as file:
+                with open(file, "rb") as icon_file:
+                    return base64.b64encode(icon_file.read())
+        except Exception:
+            LOGGER.info(f"Connector icon reading by path {str(filepath)} failed")
+            return None

lib/dl_api_lib/dl_api_lib/connection_info.py

@@ -8,17 +8,17 @@ class InfoProviderNotFound(Exception):
     pass
 
 
-CONNECTOR_INFO_PROVIDER_CLS_BY_TYPE: dict[ConnectionType, Type[ConnectionInfoProvider]] = {}
+CONNECTOR_INFO_PROVIDER_CLS_BY_TYPE: dict[ConnectionType, ConnectionInfoProvider] = {}
 
 
 def register_connector_info_provider_cls(
     conn_type: ConnectionType,
     info_provider_cls: Type[ConnectionInfoProvider],
 ) -> None:
-    CONNECTOR_INFO_PROVIDER_CLS_BY_TYPE[conn_type] = info_provider_cls
+    CONNECTOR_INFO_PROVIDER_CLS_BY_TYPE[conn_type] = info_provider_cls()
 
 
-def get_connector_info_provider_cls(conn_type: ConnectionType) -> Type[ConnectionInfoProvider]:
+def get_connector_info_provider(conn_type: ConnectionType) -> ConnectionInfoProvider:
     if (conn_info_provider := CONNECTOR_INFO_PROVIDER_CLS_BY_TYPE.get(conn_type)) is not None:
         return conn_info_provider
     raise InfoProviderNotFound(conn_type)

lib/dl_api_lib/dl_api_lib/connector_availability/base.py

@@ -18,8 +18,10 @@
 )
 
 from dl_api_commons.base_models import TenantDef
+from dl_api_connector.connection_info import ConnectionInfoProvider
 from dl_api_lib.connection_forms.registry import CONN_FORM_FACTORY_BY_TYPE
-from dl_api_lib.connection_info import get_connector_info_provider_cls
+from dl_api_lib.connection_info import get_connector_info_provider
+from dl_api_lib.exc import ConnectorIconNotFoundException
 from dl_api_lib.i18n.localizer import Translatable
 from dl_configs.connector_availability import (
     ConnectorAvailabilityConfigSettings,
@@ -142,10 +144,10 @@ class ConnectorIconRole(DynamicEnum):
 class ConnectorIconSrcConfig:
     icon_type: ConnectorIconSrcType = attr.ib()
 
-    def as_dict(self, conn_type: str) -> dict[str, Any]:
+    def as_dict(self, conn: Connector) -> dict[str, Any] | None:
         return dict(
             type=self.icon_type.value,
-            conn_type=conn_type,
+            conn_type=conn.conn_type.value,
         )
 
     @classmethod
@@ -158,10 +160,18 @@ def from_settings(cls, settings: ObjectLikeConfig | ConnectorIconSrc) -> Connect
 class ConnectorIconSrcConfigData(ConnectorIconSrcConfig):
     data: Optional[str] = attr.ib(default=None)
 
-    def as_dict(self, conn_type: str) -> dict[str, Any]:
-        data = dict(standart="", nav="")
+    def as_dict(self, conn: Connector) -> dict[str, Any] | None:
+        if not conn.connector_info_provider.icon_data_standard or not conn.connector_info_provider.icon_data_nav:
+            return None
+
+        data = dict(
+            standard=b"data:image/svg+xml;base64," + conn.connector_info_provider.icon_data_standard,
+            nav=b"data:image/svg+xml;base64," + conn.connector_info_provider.icon_data_nav,
+        )
+        base_dict = super().as_dict(conn=conn)
+        assert base_dict
         return dict(
-            **super().as_dict(conn_type=conn_type),
+            **base_dict,
             data=data,
         )
 
@@ -177,13 +187,15 @@ def from_settings(cls, settings: ObjectLikeConfig | ConnectorIconSrc) -> Connect
 class ConnectorIconSrcConfigUrl(ConnectorIconSrcConfig):
     url_prefix: str = attr.ib()
 
-    def as_dict(self, conn_type: str) -> dict[str, Any]:
+    def as_dict(self, conn: Connector) -> dict[str, Any] | None:
         url = dict(
-            standart=f"{self.url_prefix.rstrip('/')}/{ConnectorIconRole.standard.value}/{conn_type}.svg",
-            nav=f"{self.url_prefix.rstrip('/')}/{ConnectorIconRole.nav.value}/{conn_type}.svg",
+            standard=f"{self.url_prefix.rstrip('/')}/{ConnectorIconRole.standard.value}/{conn.conn_type.value}.svg",
+            nav=f"{self.url_prefix.rstrip('/')}/{ConnectorIconRole.nav.value}/{conn.conn_type.value}.svg",
         )
+        base_dict = super().as_dict(conn=conn)
+        assert base_dict
         return dict(
-            **super().as_dict(conn_type=conn_type),
+            **base_dict,
             url=url,
         )
 
@@ -235,7 +247,11 @@ def visibility_mode(self) -> ConnectorAvailability:
 
     @property
     def alias(self) -> str:
-        return get_connector_info_provider_cls(self.conn_type).alias or self.conn_type_str
+        return get_connector_info_provider(self.conn_type).alias or self.conn_type_str
+
+    @property
+    def connector_info_provider(self) -> ConnectionInfoProvider:
+        return get_connector_info_provider(self.conn_type)
 
     @property
     def conn_type_str(self) -> str:
@@ -246,7 +262,7 @@ def backend_driven_form(self) -> bool:
         return self.conn_type in CONN_FORM_FACTORY_BY_TYPE
 
     def get_title(self, localizer: Localizer) -> str:
-        return localizer.translate(get_connector_info_provider_cls(self.conn_type).title_translatable)
+        return localizer.translate(get_connector_info_provider(self.conn_type).title_translatable)
 
 
 @attr.s(kw_only=True)
@@ -317,15 +333,18 @@ def fill_icon_dict_by_conn_type(self) -> None:
             return
         for conn in self._iter_connectors():
             conn_type = conn.conn_type.value
-            icon_src = self.icon_src.as_dict(conn_type=conn_type)
-            self._icon_by_conn_type[conn_type] = deepcopy(icon_src)
+            icon_src = self.icon_src.as_dict(conn=conn)
+            if icon_src is not None:
+                self._icon_by_conn_type[conn_type] = deepcopy(icon_src)
 
     def list_icons(self) -> list[dict[str, Any]]:
         self.fill_icon_dict_by_conn_type()
         return list(self._icon_by_conn_type.values())
 
     def get_icon(self, conn_type: str) -> Optional[dict[str, Any]]:
         self.fill_icon_dict_by_conn_type()
+        if conn_type not in self._icon_by_conn_type:
+            raise ConnectorIconNotFoundException()
         return self._icon_by_conn_type[conn_type]
 
     def __attrs_post_init__(self) -> None:

lib/dl_api_lib/dl_api_lib/error_handling.py

@@ -84,6 +84,7 @@
     dl_query_processing.exc.InvalidGroupByConfiguration: status.BAD_REQUEST,
     common_exc.WrongQueryParameterization: status.BAD_REQUEST,
     RequestTimeoutError: status.FAILED_DEPENDENCY,
+    exc.ConnectorIconNotFoundException: status.NOT_FOUND,
 }
 
 

lib/dl_api_lib/dl_api_lib/exc.py

@@ -43,3 +43,8 @@ class TooManyFieldsError(DLValidationError):
 class DLValidationFatal(_DLValidationResult):
     err_code = _DLValidationResult.err_code + ["FATAL"]
     default_message = "Validation encountered a fatal error."
+
+
+class ConnectorIconNotFoundException(DLBaseException):
+    default_message = "Connector icon not found"
+    err_code = ["ICON_NOT_FOUND"]

lib/dl_api_lib/dl_api_lib_tests/db/control_api/test_info.py

@@ -90,6 +90,9 @@ def test_get_connector_icon(self, client):
         resp = form_resp.json
         assert resp["icons"]
         assert len(resp["icons"]) > 0
-        for icons in resp["icons"]:
-            assert "type" in icons
-            assert icons["type"] == "data"
+        for icon in resp["icons"]:
+            assert "type" in icon
+            assert icon["type"] == "data"
+            assert "data" in icon
+            assert icon["data"].get("standard") is not None
+            assert icon["data"].get("nav") is not None

lib/dl_connector_bigquery/dl_connector_bigquery/api/connection_info.py

@@ -1,7 +1,18 @@
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_bigquery.api.i18n.localizer import Translatable
+from dl_connector_bigquery.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class BigQueryConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-bigquery")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("bigquery.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("bigquery.svg")
+        super().__attrs_post_init__()

lib/dl_connector_bitrix_gds/dl_connector_bitrix_gds/api/connection_info.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_bitrix_gds.api.i18n.localizer import Translatable
+from dl_connector_bitrix_gds.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class BitrixGDSConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-bitrix")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("bitrix24.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("bitrix24.svg")
+        super().__attrs_post_init__()

lib/dl_connector_bundle_chs3/dl_connector_bundle_chs3/chs3_gsheets/api/connection_info.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_bundle_chs3.chs3_base.api.i18n.localizer import Translatable
+from dl_connector_bundle_chs3.chs3_gsheets.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class GSheetsFileS3ConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-gsheets_v2")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("gsheets.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("gsheets.svg")
+        super().__attrs_post_init__()

lib/dl_connector_bundle_chs3/dl_connector_bundle_chs3/chs3_yadocs/api/connection_info.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_bundle_chs3.chs3_base.api.i18n.localizer import Translatable
+from dl_connector_bundle_chs3.chs3_yadocs.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class YaDocsFileS3ConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-yadocs")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("yadocs.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("yadocs.svg")
+        super().__attrs_post_init__()

lib/dl_connector_bundle_chs3/dl_connector_bundle_chs3/file/api/connection_info.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_bundle_chs3.chs3_base.api.i18n.localizer import Translatable
+from dl_connector_bundle_chs3.file.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class FileS3ConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-file")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("file.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("file.svg")
+        super().__attrs_post_init__()

lib/dl_connector_chyt/dl_connector_chyt/api/connection_info.py

@@ -1,9 +1,20 @@
 from __future__ import annotations
 
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_chyt.api.i18n.localizer import Translatable
+from dl_connector_chyt.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class CHYTConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-chyt")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("chyt.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("chyt.svg")
+        super().__attrs_post_init__()

lib/dl_connector_clickhouse/dl_connector_clickhouse/api/connection_info.py

@@ -1,7 +1,18 @@
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_clickhouse.api.i18n.localizer import Translatable
+from dl_connector_clickhouse.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class ClickHouseConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-clickhouse")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("clickhouse.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("clickhouse.svg")
+        super().__attrs_post_init__()

lib/dl_connector_greenplum/dl_connector_greenplum/api/connection_info.py

@@ -1,7 +1,18 @@
+from importlib.resources import files
+
 from dl_api_connector.connection_info import ConnectionInfoProvider
 
 from dl_connector_greenplum.api.i18n.localizer import Translatable
+from dl_connector_greenplum.assets.icons import (
+    nav,
+    standard,
+)
 
 
 class GreenplumConnectionInfoProvider(ConnectionInfoProvider):
     title_translatable = Translatable("label_connector-greenplum")
+
+    def __attrs_post_init__(self) -> None:
+        self.icon_data_standard_filepath = files(standard).joinpath("greenplum.svg")
+        self.icon_data_nav_filepath = files(nav).joinpath("greenplum.svg")
+        super().__attrs_post_init__()