Repository: astronomer/astronomer-cosmos

Issue:
[Bug] Remote target directory might not work when specifying local directory
### Astronomer Cosmos Version

1.10.0

### dbt-core version

1.9.4

### Versions of dbt adapters

_No response_

### LoadMode

AUTOMATIC

### ExecutionMode

AIRFLOW_ASYNC

### InvocationMode

None

### airflow version

2.10.5

### Operating System

Linux

### If a you think it's an UI issue, what browsers are you seeing the problem on?

_No response_

### Deployment

Docker-Compose

### Deployment details

_No response_

### What happened?

When configuring a remote target directory that points to a local path while using `AIRFLOW ASYNC`, like so:

```bash
AIRFLOW__COSMOS__REMOTE_TARGET_PATH=/usr/local/airflow/cosmos
AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID=file_default
```

you might encounter a `FileNotFoundError` when the system attempts to write files to the local directory:

```bash
FileNotFoundError: [Errno 2] No such file or directory: '/usr/local/airflow/cosmos/simple_dag_async__dbt_async/run/jaffle_shop/models/example/my_second_dbt_model.sql'
```

This suggests that the target directory structure doesn’t exist prior to the write operation. When using a local directory as the remote target path for compiled SQL (especially in async mode), Cosmos should ensure the necessary subdirectories are created before attempting file writes.

It would be helpful to add logic that checks for and creates the target directory tree if it doesn't already exist.

### Relevant log output

```shell
[2025-05-04, 23:21:11 +07] {taskinstance.py:3313} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 768, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 734, in _execute_callable
    return ExecutionCallableRunner(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/operator_helpers.py", line 252, in run
    return self.func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 424, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/cosmos/operators/_asynchronous/__init__.py", line 46, in execute
    self.build_and_run_cmd(
  File "/usr/local/lib/python3.12/site-packages/cosmos/operators/local.py", line 797, in build_and_run_cmd
    result = self.run_command(
             ^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/cosmos/operators/virtualenv.py", line 117, in run_command
    return super().run_command(cmd, env, context, run_as_async=run_as_async, async_context=async_context)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/cosmos/operators/local.py", line 597, in run_command
    self._handle_async_execution(tmp_project_dir, context, async_context)
  File "/usr/local/lib/python3.12/site-packages/cosmos/operators/local.py", line 528, in _handle_async_execution
    self._upload_sql_files(tmp_project_dir, "run")
  File "/usr/local/lib/python3.12/site-packages/cosmos/operators/local.py", line 332, in _upload_sql_files
    ObjectStoragePath(file_path).copy(dest_object_storage_path)
  File "/usr/local/lib/python3.12/site-packages/airflow/io/path.py", line 368, in copy
    self._cp_file(dst, **kwargs)
  File "/usr/local/lib/python3.12/site-packages/airflow/io/path.py", line 297, in _cp_file
    with self.open("rb") as f1, dst.open("wb") as f2:
                                ^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/io/path.py", line 162, in open
    return TrackingFileWrapper(self, self.fs.open(self.path, mode=mode, **kwargs))
                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/fsspec/spec.py", line 1301, in open
    f = self._open(
        ^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/fsspec/implementations/local.py", line 195, in _open
    return LocalFileOpener(path, mode, fs=self, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/fsspec/implementations/local.py", line 359, in __init__
    self._open()
  File "/usr/local/lib/python3.12/site-packages/fsspec/implementations/local.py", line 364, in _open
    self.f = open(self.path, mode=self.mode)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
FileNotFoundError: [Errno 2] No such file or directory: '/usr/local/airflow/cosmos/simple_dag_async__dbt_async/run/jaffle_shop/models/example/my_second_dbt_model.sql'
```

### How to reproduce

Just set the environment variables like below:

```bash
AIRFLOW__COSMOS__REMOTE_TARGET_PATH=/usr/local/airflow/cosmos
AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID=file_default
```

And the Apache Airflow integrating with Astronomer Cosmos:

```python
# [START airflow_async_execution_mode_example]
simple_dag_async = DbtDag(
    # dbt/cosmos-specific parameters
    project_config=ProjectConfig(
        DBT_ROOT_PATH / "jaffle_shop",
    ),
    profile_config=profile_config,
    execution_config=ExecutionConfig(
        execution_mode=ExecutionMode.AIRFLOW_ASYNC,
        async_py_requirements=[f"dbt-bigquery=={DBT_ADAPTER_VERSION}"],
    ),
    render_config=RenderConfig(select=["path:models"], test_behavior=TestBehavior.NONE),
    # normal dag parameters
    schedule=None,
    start_date=datetime(2023, 1, 1),
    catchup=False,
    dag_id="simple_dag_async",
    tags=["simple"],
    operator_args={"location": "US", "install_deps": True, "full_refresh": True},
)
# [END airflow_async_execution_mode_example]
```

### Anything else :)?

_No response_

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Contact Details

tntuan0910@gmail.com

Hints from issue comments:
@dosubot: <!-- Answer -->
To resolve the `FileNotFoundError` when using a local directory as the remote target path in Astronomer Cosmos with `AIRFLOW ASYNC` mode, ensure that the necessary subdirectories are created before file writes. You can achieve this by using the `_configure_remote_target_path` function from the `cosmos/operators/local.py` file, which checks if the `remote_target_path` is set and creates the necessary directories using `mkdir(parents=True, exist_ok=True)` if they do not exist. Make sure this function is called before any file operations that might lead to a `FileNotFoundError` <sup>[[1]](https://github.com/astronomer/astronomer-cosmos/blob/main/cosmos/operators/local.py)</sup>.

Additionally, verify that the `remote_target_path` and `remote_target_path_conn_id` are correctly configured in your Cosmos settings. The `remote_target_path` should be a valid path supported by Airflow's Object Store feature, and the `remote_target_path_conn_id` should be specified to establish the connection <sup>[[2]](https://github.com/astronomer/astronomer-cosmos/blob/main/docs/configuration/cosmos-conf.rst)</sup>.


<!-- Dosu Comment Footer -->
*To continue talking to [Dosu](https://dosu.dev), mention @dosu.*

---
<sup>Help Dosu learn! Give it feedback: [Great Response](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=great_response) | [Irrelevant Answer](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=irrelevant_answer) | [Incorrect Sources](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=incorrect_sources) | [Too Verbose](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=hallucination) | [Bug Report](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/1396c443-b439-4679-99fb-818bcdad8cb1?feedback_type=other)</sup>


Solution patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index d0ffd9d6..c74ced87 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -57,7 +57,7 @@ repos:
           - --py37-plus
           - --keep-runtime-typing
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.11.8
+    rev: v0.11.9
     hooks:
       - id: ruff
         args:
diff --git a/cosmos/operators/local.py b/cosmos/operators/local.py
index 08164979..4cbb420e 100644
--- a/cosmos/operators/local.py
+++ b/cosmos/operators/local.py
@@ -329,6 +329,7 @@ class AbstractDbtLocalBase(AbstractDbtBase):
         for file_path in files:
             dest_file_path = self._construct_dest_file_path(dest_target_dir, file_path, source_run_dir, resource_type)
             dest_object_storage_path = ObjectStoragePath(dest_file_path, conn_id=dest_conn_id)
+            dest_object_storage_path.parent.mkdir(parents=True, exist_ok=True)
             ObjectStoragePath(file_path).copy(dest_object_storage_path)
             self.log.debug("Copied %s to %s", file_path, dest_object_storage_path)
 


Test patch:
diff --git a/tests/operators/test_local.py b/tests/operators/test_local.py
index b68458d8..9647fb02 100644
--- a/tests/operators/test_local.py
+++ b/tests/operators/test_local.py
@@ -1724,3 +1724,31 @@ def test_handle_post_execution_with_multiple_callbacks(
 
     for callback_fn in multiple_callbacks:
         callback_fn.assert_called_once_with("/tmp/project_dir", arg1="value1", context=context)
+
+
+@pytest.mark.skipif(not AIRFLOW_IO_AVAILABLE, reason="Airflow did not have Object Storage until the 2.8 release")
+@patch("airflow.io.path.ObjectStoragePath")
+def test_upload_sql_files_creates_parent_directories(mock_object_storage_path):
+    """Test that parent directories are created during file uploads."""
+
+    operator = ConcreteDbtLocalBaseOperator(
+        profile_config=profile_config,
+        task_id="test-task",
+        project_dir="test/dir",
+    )
+
+    with patch.object(
+        operator, "_configure_remote_target_path", return_value=("dest/dir", "mock_conn_id")
+    ), patch.object(operator, "_construct_dest_file_path", return_value="dest/path/file.sql"), patch(
+        "pathlib.Path.rglob", return_value=[Path("file.sql")]
+    ), patch(
+        "pathlib.Path.is_file", return_value=True
+    ):
+
+        mock_dest_path = MagicMock()
+        mock_dest_path.parent = MagicMock()
+        mock_object_storage_path.return_value = mock_dest_path
+
+        operator._upload_sql_files("tmp_dir", "compiled")
+
+        mock_dest_path.parent.mkdir.assert_called_with(parents=True, exist_ok=True)