032: fix coroutine never awaited in debug.py + task_logger.py
debug.py: _test_db_api and _get_dataset_structure were already async def but called SupersetClient methods (authenticate, get_databases, get_dataset) without await. Added await to 4 calls. task_logger.py: _add_log callback is async def but _log() called it without await, silently dropping all task log messages (RuntimeWarning: coroutine never awaited). Changed to fire-and-forget via asyncio.ensure_future when a running event loop is available, drops gracefully otherwise.
This commit is contained in:
@@ -4,6 +4,7 @@
|
|||||||
# @RELATION DEPENDS_ON -> [TaskManager]
|
# @RELATION DEPENDS_ON -> [TaskManager]
|
||||||
# @RELATION DEPENDS_ON -> [EventBus]
|
# @RELATION DEPENDS_ON -> [EventBus]
|
||||||
# @INVARIANT Each TaskLogger instance is bound to a specific task_id and default source.
|
# @INVARIANT Each TaskLogger instance is bound to a specific task_id and default source.
|
||||||
|
import asyncio
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
@@ -81,13 +82,21 @@ class TaskLogger:
|
|||||||
metadata: dict[str, Any] | None = None,
|
metadata: dict[str, Any] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Internal logging method."""
|
"""Internal logging method."""
|
||||||
self._add_log(
|
coro = self._add_log(
|
||||||
task_id=self._task_id,
|
task_id=self._task_id,
|
||||||
level=level,
|
level=level,
|
||||||
message=message,
|
message=message,
|
||||||
source=source or self._default_source,
|
source=source or self._default_source,
|
||||||
metadata=metadata,
|
metadata=metadata,
|
||||||
)
|
)
|
||||||
|
# Fire-and-forget: schedule the async callback if a loop is running,
|
||||||
|
# otherwise drop gracefully (sync context, no event loop available).
|
||||||
|
try:
|
||||||
|
loop = asyncio.get_running_loop()
|
||||||
|
if loop.is_running():
|
||||||
|
asyncio.ensure_future(coro)
|
||||||
|
except RuntimeError:
|
||||||
|
pass
|
||||||
# #endregion _log
|
# #endregion _log
|
||||||
# #region debug [TYPE Function]
|
# #region debug [TYPE Function]
|
||||||
# @PURPOSE: Log a DEBUG level message.
|
# @PURPOSE: Log a DEBUG level message.
|
||||||
|
|||||||
@@ -167,8 +167,8 @@ class DebugPlugin(PluginBase):
|
|||||||
raise ValueError(f"Environment '{name}' not found.")
|
raise ValueError(f"Environment '{name}' not found.")
|
||||||
|
|
||||||
client = SupersetClient(env_config)
|
client = SupersetClient(env_config)
|
||||||
client.authenticate()
|
await client.authenticate()
|
||||||
count, dbs = client.get_databases()
|
count, dbs = await client.get_databases()
|
||||||
log.debug(f"Found {count} databases in {name}")
|
log.debug(f"Found {count} databases in {name}")
|
||||||
results[name] = {
|
results[name] = {
|
||||||
"count": count,
|
"count": count,
|
||||||
@@ -203,9 +203,9 @@ class DebugPlugin(PluginBase):
|
|||||||
raise ValueError(f"Environment '{env_name}' not found.")
|
raise ValueError(f"Environment '{env_name}' not found.")
|
||||||
|
|
||||||
client = SupersetClient(env_config)
|
client = SupersetClient(env_config)
|
||||||
client.authenticate()
|
await client.authenticate()
|
||||||
|
|
||||||
dataset_response = client.get_dataset(dataset_id)
|
dataset_response = await client.get_dataset(dataset_id)
|
||||||
log.debug(f"Retrieved dataset structure for {dataset_id}")
|
log.debug(f"Retrieved dataset structure for {dataset_id}")
|
||||||
return dataset_response.get('result') or {}
|
return dataset_response.get('result') or {}
|
||||||
# endregion _get_dataset_structure
|
# endregion _get_dataset_structure
|
||||||
|
|||||||
Reference in New Issue
Block a user