|
16 | 16 | from collections import defaultdict |
17 | 17 | from collections.abc import Iterable |
18 | 18 | from itertools import chain |
| 19 | +from typing import TYPE_CHECKING |
19 | 20 |
|
20 | 21 | from sqlalchemy.engine.reflection import Inspector |
21 | 22 |
|
|
44 | 45 | get_materialized_view_definition, |
45 | 46 | query_tables, |
46 | 47 | ) |
| 48 | + |
| 49 | +if TYPE_CHECKING: |
| 50 | + from metadata.ingestion.source.database.questdb.models import QuestDBTableRow |
47 | 51 | from metadata.utils.logger import ingestion_logger |
48 | 52 |
|
49 | 53 | logger = ingestion_logger() |
@@ -78,15 +82,14 @@ def create(cls, config_dict: dict, metadata: OpenMetadata, pipeline_name: str | |
78 | 82 |
|
79 | 83 | def __init__(self, config: WorkflowSource, metadata: OpenMetadata) -> None: |
80 | 84 | super().__init__(config, metadata) |
| 85 | + self._tables_cache: defaultdict[str, dict[str, QuestDBTableRow]] = defaultdict(dict) |
81 | 86 | try: |
82 | 87 | rows = query_tables(self.connection) |
83 | | - self._tables_cache: defaultdict[str, dict[str, object]] = defaultdict(dict) |
84 | 88 | for row in rows: |
85 | 89 | self._tables_cache[row.table_type][row.name] = row |
86 | 90 | except Exception as exc: |
87 | 91 | logger.debug(traceback.format_exc()) |
88 | 92 | logger.warning("Failed to load QuestDB table catalog: %s — partition details will be unavailable", exc) |
89 | | - self._tables_cache = {} |
90 | 93 |
|
91 | 94 | def get_database_names(self) -> Iterable[str]: |
92 | 95 | yield QUESTDB_DEFAULT_DATABASE |
|
0 commit comments