feat: BI-6450 split debug query for logging and inspector (#1379)

* add inspector query

* fixes

* tmp test

* trying to fix debug query in good inspector

* rm debug strings

* lint

* lint

* typo fix

Author: chl0rum

lib/dl_api_lib/dl_api_lib/dataset/validator.py

@@ -1178,6 +1178,7 @@ def get_schema_info(exists: bool) -> Optional[SchemaInfo]:
                     raise common_exc.SourceDoesNotExist(
                         db_message="",
                         query="",
+                        inspector_query="",
                         params=params,
                     )
             return None

lib/dl_compeng_pg/dl_compeng_pg/compeng_asyncpg/exec_adapter_asyncpg.py

@@ -80,11 +80,13 @@ async def _execute_ddl(self, query: Union[str, sa.sql.base.Executable]) -> None:
         await self._execute(query)
 
     @contextmanager
-    def handle_db_errors(self, query_text: str) -> Generator[None, None, None]:
+    def handle_db_errors(self, debug_query: str, inspector_query: str | None) -> Generator[None, None, None]:
         try:
             yield
         except Exception as wrapper_exc:
-            trans_exc = self._error_transformer.make_bi_error(wrapper_exc=wrapper_exc, debug_compiled_query=query_text)
+            trans_exc = self._error_transformer.make_bi_error(
+                wrapper_exc=wrapper_exc, debug_query=debug_query, inspector_query=inspector_query
+            )
             raise trans_exc from wrapper_exc
 
     async def _execute_and_fetch(
@@ -107,7 +109,7 @@ async def _execute_and_fetch(
         async def chunked_data_gen() -> AsyncGenerator[list[list[TBIDataValue]], None]:
             """Fetch data in chunks"""
 
-            with self.handle_db_errors(query_text=query_text):
+            with self.handle_db_errors(debug_query=query_text, inspector_query=query_text):  # TODO: BI-6448
                 cur = await self._conn.cursor(query_text, *params)
                 while True:
                     chunk = []

lib/dl_connector_bitrix_gds/dl_connector_bitrix_gds/core/adapter.py

@@ -163,7 +163,9 @@ async def _run_query(self, dba_query: DBAdapterQuery) -> Any:
         if limit is not None:
             request_params["limit"] = limit
 
-        with self.handle_execution_error(query_text):
+        # TODO: BI-6448 check it works and replace
+        # debug_query = compile_query_for_debug(dba_query.query, self.get_dialect())
+        with self.handle_execution_error(debug_query=query_text, inspector_query=query_text):
             resp = await self._session.post(
                 url=api_url,
                 params=request_params,
@@ -250,6 +252,7 @@ def _parse_response_body(self, body: Any, dba_query: DBAdapterQuery) -> dict:
                 message=f"Unexpected API response body: {err.args[0]}",
                 db_message="",
                 query=dba_query.debug_compiled_query,
+                inspector_query=dba_query.inspector_query,
                 orig=None,
                 details={},
             ) from err
@@ -297,9 +300,12 @@ async def _get_user_tables(self, schema_ident: SchemaIdent) -> list[str]:
         body: dict[str, Any] = {
             "key": self._target_dto.token,
         }
+        # TODO: BI-6448 example if we want hide portal (want we? don't know)
+        # base_url = "https://{portal}/bitrix/tools/biconnector/gds.php?show_tables"
+        # api_url = base_url.format(portal=self._target_dto.portal)
         api_url: str = f"https://{self._target_dto.portal}/bitrix/tools/biconnector/gds.php?show_tables"
 
-        with self.handle_execution_error(api_url):
+        with self.handle_execution_error(debug_query=api_url, inspector_query=api_url):
             resp = await self._session.post(
                 url=api_url,
                 json=body,

lib/dl_connector_bundle_chs3/dl_connector_bundle_chs3/chs3_base/core/adapter.py

@@ -65,9 +65,11 @@ async def _make_query(self, dba_q: DBAdapterQuery, mirroring_mode: bool = False)
         )
 
         debug_query = dba_q.debug_compiled_query if dba_q.debug_compiled_query is not None else query_str_raw
+        debug_compiled_query = reduce(lambda a, kv: a.replace(*kv), secrets_hidden, debug_query) % ()
 
         dba_q = dba_q.clone(
             query=reduce(lambda a, kv: a.replace(*kv), secrets, query_str_raw) % (),
-            debug_compiled_query=reduce(lambda a, kv: a.replace(*kv), secrets_hidden, debug_query) % (),
+            debug_compiled_query=debug_compiled_query,
+            inspector_query=debug_compiled_query,  # TODO: BI-6448
         )
         return await super()._make_query(dba_q, mirroring_mode)

lib/dl_connector_chyt/dl_connector_chyt/core/adapters.py

@@ -78,14 +78,16 @@ def get_ch_settings(self) -> dict:
     def make_exc(  # TODO:  Move to ErrorTransformer
         cls,
         wrapper_exc: Exception,
-        orig_exc: Optional[Exception],
-        debug_compiled_query: Optional[str],
+        orig_exc: Exception | None,
+        debug_query: str | None,
+        inspector_query: str | None,
         **kwargs: Any,
     ) -> tuple[type[exc.DatabaseQueryError], DBExcKWArgs]:
         exc_cls, kw = super().make_exc(
             wrapper_exc=wrapper_exc,
             orig_exc=orig_exc,
-            debug_compiled_query=debug_compiled_query,
+            debug_query=debug_query,
+            inspector_query=inspector_query,
             **kwargs,
         )
         return ensure_db_message(exc_cls, kw)

lib/dl_connector_clickhouse/dl_connector_clickhouse/core/clickhouse_base/adapters.py

@@ -183,9 +183,13 @@ def _convert_headers_to_dsn_params(headers: dict[str, str]) -> dict[str, str]:
 
     @classmethod
     def make_exc(  # TODO:  Move to ErrorTransformer
-        cls, wrapper_exc: Exception, orig_exc: Optional[Exception], debug_compiled_query: Optional[str]
+        cls,
+        wrapper_exc: Exception,
+        orig_exc: Exception | None,
+        debug_query: str | None,
+        inspector_query: str | None,
     ) -> tuple[type[exc.DatabaseQueryError], DBExcKWArgs]:
-        exc_cls, kw = super().make_exc(wrapper_exc, orig_exc, debug_compiled_query)
+        exc_cls, kw = super().make_exc(wrapper_exc, orig_exc, debug_query, inspector_query)
 
         ch_exc_cls: Optional[type[exc.DatabaseQueryError]] = None
         if isinstance(wrapper_exc, ch_exc.DatabaseException):
@@ -517,6 +521,7 @@ async def _parse_response_body(self, resp: ClientResponse) -> AsyncGenerator[tup
                     debug_info=dict(msg="Unexpected last event", event=event, data=data, chunk=decoded_chunk),
                     # TODO: provide the actual query
                     query="",
+                    inspector_query="",
                     # TODO?: show the chunk here for the user databases?
                     db_message="",
                 )
@@ -542,7 +547,9 @@ async def execute(self, query: DBAdapterQuery) -> AsyncRawExecutionResult:
         if resp.status != 200:
             bytes_body = await resp.read()
             body = bytes_body.decode(errors="replace")
-            db_exc = self.make_exc(resp.status, body, debug_compiled_query=query.debug_compiled_query)
+            db_exc = self.make_exc(
+                resp.status, body, debug_query=query.debug_compiled_query, inspector_query=query.inspector_query
+            )
             raise db_exc
 
         # Primarily for DashSQL
@@ -563,7 +570,9 @@ def _safe_col_converter(col_conv: Callable[[str], Any], val: str) -> Any:
             try:
                 return col_conv(val)
             except ValueError as err:
-                raise exc.DataParseError(f"Cannot convert {val!r}", query=query.debug_compiled_query) from err
+                raise exc.DataParseError(
+                    f"Cannot convert {val!r}", query=query.debug_compiled_query, inspector_query=query.inspector_query
+                ) from err
 
         events_generator = self._parse_response_body(resp)
         with self.wrap_execute_excs(query=query, stage="meta"):
@@ -642,7 +651,8 @@ def make_exc(  # TODO:  Move to ErrorTransformer
         self,
         status_code: int,  # noqa
         err_body: str,
-        debug_compiled_query: Optional[str] = None,
+        debug_query: str | None = None,
+        inspector_query: str | None = None,
     ) -> exc.DatabaseQueryError:
         exc_cls: type[exc.DatabaseQueryError]
         try:
@@ -652,7 +662,8 @@ def make_exc(  # TODO:  Move to ErrorTransformer
 
         return exc_cls(
             db_message=err_body,
-            query=debug_compiled_query,
+            query=debug_query,
+            inspector_query=inspector_query,
             orig=None,
             details={},
             params=exc_params,

lib/dl_connector_clickhouse/dl_connector_clickhouse/core/clickhouse_base/exc.py

@@ -16,6 +16,7 @@ def __init__(
         self,
         db_message: Optional[str] = None,
         query: Optional[str] = None,
+        inspector_query: Optional[str] = None,
         message: Optional[str] = None,
         details: Optional[dict[str, Any]] = None,
         orig: Optional[Exception] = None,
@@ -25,6 +26,7 @@ def __init__(
         super(ClickHouseSourceDoesNotExistError, self).__init__(
             db_message=db_message,
             query=query,
+            inspector_query=inspector_query,
             message=message,
             details=details,
             orig=orig,

lib/dl_connector_metrica/dl_connector_metrica/core/adapters_metrica_x.py

@@ -55,9 +55,13 @@ def _get_db_engine(self, db_name: str, disable_streaming: bool = False) -> Engin
 
     @classmethod
     def make_exc(  # TODO:  Move to ErrorTransformer
-        cls, wrapper_exc: Exception, orig_exc: Optional[Exception], debug_compiled_query: Optional[str]
+        cls,
+        wrapper_exc: Exception,
+        orig_exc: Exception | None,
+        debug_query: str | None,
+        inspector_query: str | None,
     ) -> tuple[type[exc.DatabaseQueryError], DBExcKWArgs]:
-        exc_cls, kw = super().make_exc(wrapper_exc, orig_exc, debug_compiled_query)
+        exc_cls, kw = super().make_exc(wrapper_exc, orig_exc, debug_query, inspector_query)
 
         if isinstance(
             orig_exc,

lib/dl_connector_mssql/dl_connector_mssql/core/adapters_mssql.py

@@ -255,9 +255,13 @@ def get_exc_class(cls, err_msg: str) -> Optional[type[exc.DatabaseQueryError]]:
 
     @classmethod
     def make_exc(  # TODO:  Move to ErrorTransformer
-        cls, wrapper_exc: Exception, orig_exc: Optional[Exception], debug_compiled_query: Optional[str]
+        cls,
+        wrapper_exc: Exception,
+        orig_exc: Exception | None,
+        debug_query: str | None,
+        inspector_query: str | None,
     ) -> tuple[type[exc.DatabaseQueryError], DBExcKWArgs]:
-        exc_cls, kw = super().make_exc(wrapper_exc, orig_exc, debug_compiled_query)
+        exc_cls, kw = super().make_exc(wrapper_exc, orig_exc, debug_query, inspector_query)
 
         db_msg = kw["db_message"]
         specific_exc_cls = cls.get_exc_class(db_msg)  # type: ignore  # TODO: fix

lib/dl_connector_mssql/dl_connector_mssql/core/exc.py

@@ -14,6 +14,7 @@ def __init__(
         self,
         db_message: Optional[str] = None,
         query: Optional[str] = None,
+        inspector_query: Optional[str] = None,
         message: Optional[str] = None,
         details: Optional[dict[str, Any]] = None,
         orig: Optional[Exception] = None,
@@ -23,6 +24,7 @@ def __init__(
         super(SyncMssqlSourceDoesNotExistError, self).__init__(
             db_message=db_message,
             query=query,
+            inspector_query=inspector_query,
             message=message,
             details=details,
             orig=orig,