Solve circular import error in Astro runtime while debugging

cosmos/config.py

@@ -9,7 +9,7 @@
 from collections.abc import Callable, Iterator
 from dataclasses import InitVar, dataclass, field
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import yaml
 
@@ -34,7 +34,9 @@
 from cosmos.dbt.executable import get_system_dbt
 from cosmos.exceptions import CosmosValueError
 from cosmos.log import get_logger
-from cosmos.profiles import BaseProfileMapping
+
+if TYPE_CHECKING:
+    from cosmos.profiles import BaseProfileMapping
 
 logger = get_logger(__name__)
 
@@ -321,7 +323,9 @@ def validate_profiles_yml(self) -> None:
             raise CosmosValueError(f"The file {self.profiles_yml_filepath} does not exist.")
 
     def get_profile_type(self) -> str:
-        if isinstance(self.profile_mapping, BaseProfileMapping):
+        from cosmos.profiles.base import BaseProfileMapping as _BaseProfileMapping
+
+        if isinstance(self.profile_mapping, _BaseProfileMapping):
             return str(self.profile_mapping.dbt_profile_type)
 
         profile_path = self._get_profile_path()

cosmos/log.py

@@ -2,8 +2,6 @@
 
 import logging
 
-from cosmos.settings import rich_logging
-
 
 class CosmosRichLogger(logging.Logger):
     """Custom Logger that prepends ``(astronomer-cosmos)`` to each log message in the scheduler."""
@@ -24,7 +22,14 @@ def get_logger(name: str) -> logging.Logger:
     as long as the ``rich_logging`` setting is True:
     [2023-08-09T14:20:55.532+0100] {subprocess.py:94} INFO - (astronomer-cosmos) - 13:20:55  Completed successfully
     """
-    if rich_logging:
+    # Import cosmos.settings at call time (not module level) to avoid circular imports
+    # during Airflow plugin discovery. getattr tolerates the module being partially
+    # initialized (rich_logging not yet defined) by falling back to False.
+    import cosmos.settings as _settings
+
+    _rich = getattr(_settings, "rich_logging", False)
+
+    if _rich:
         cls = logging.getLoggerClass()
         try:
             logging.setLoggerClass(CosmosRichLogger)

tests/test_log.py

@@ -1,16 +1,15 @@
 import pytest
 
-import cosmos.log
 from cosmos.log import CosmosRichLogger, get_logger
 from cosmos.provider_info import get_provider_info
 
 
 def test_get_logger(monkeypatch):
-    monkeypatch.setattr(cosmos.log, "rich_logging", False)
+    monkeypatch.setattr("cosmos.settings.rich_logging", False)
     standard_logger = get_logger("test-get-logger-example1")
     assert not isinstance(standard_logger, CosmosRichLogger)
 
-    monkeypatch.setattr(cosmos.log, "rich_logging", True)
+    monkeypatch.setattr("cosmos.settings.rich_logging", True)
     custom_logger = get_logger("test-get-logger-example2")
     assert isinstance(custom_logger, CosmosRichLogger)
 
@@ -21,7 +20,7 @@ def test_get_logger(monkeypatch):
 
 
 def test_rich_logging(monkeypatch, caplog):
-    monkeypatch.setattr(cosmos.log, "rich_logging", False)
+    monkeypatch.setattr("cosmos.settings.rich_logging", False)
     standard_logger = get_logger("test-rich-logging-example1")
 
     with caplog.at_level("INFO"):
@@ -32,7 +31,7 @@ def test_rich_logging(monkeypatch, caplog):
     assert log_output.count("\n") == 1
 
     caplog.clear()
-    monkeypatch.setattr(cosmos.log, "rich_logging", True)
+    monkeypatch.setattr("cosmos.settings.rich_logging", True)
     custom_logger = get_logger("test-rich-logging-example2")
     with caplog.at_level("INFO"):
         custom_logger.info("Hello, world!")