feat: added constant for overriding directories

airbyte/_executors/util.py

@@ -1,6 +1,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 from __future__ import annotations
 
+import tempfile
 from pathlib import Path
 from typing import TYPE_CHECKING, Literal, cast
 
@@ -16,6 +17,7 @@
 from airbyte._executors.python import VenvExecutor
 from airbyte._util.meta import which
 from airbyte._util.telemetry import EventState, log_install_state  # Non-public API
+from airbyte.constants import OVERRIDE_TEMP_DIR
 from airbyte.sources.registry import ConnectorMetadata, InstallType, get_connector_metadata
 
 
@@ -202,8 +204,7 @@ def get_connector_executor(  # noqa: PLR0912, PLR0913 # Too complex
         if ":" not in docker_image:
             docker_image = f"{docker_image}:{version or 'latest'}"
 
-        temp_dir = Path.home() / ".airbyte" / "temp"
-        temp_dir.mkdir(parents=True, exist_ok=True)
+        temp_dir = OVERRIDE_TEMP_DIR if OVERRIDE_TEMP_DIR else Path(tempfile.gettempdir())
 
         local_mount_dir = Path().absolute() / name
         local_mount_dir.mkdir(exist_ok=True)

airbyte/_util/temp_files.py

@@ -11,6 +11,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from airbyte.constants import OVERRIDE_TEMP_DIR
+
 
 if TYPE_CHECKING:
     from collections.abc import Generator
@@ -21,8 +23,7 @@ def as_temp_files(files_contents: list[dict | str]) -> Generator[list[str], Any,
     """Write the given contents to temporary files and yield the file paths as strings."""
     temp_files: list[Any] = []
     try:
-        temp_dir = Path.home() / ".airbyte" / "temp"
-        temp_dir.mkdir(parents=True, exist_ok=True)
+        temp_dir = OVERRIDE_TEMP_DIR
 
         for content in files_contents:
             use_json = isinstance(content, dict)

airbyte/constants.py

@@ -64,6 +64,20 @@ def _str_to_bool(value: str) -> bool:
     """Convert a string value of an environment values to a boolean value."""
     return bool(value) and value.lower() not in {"", "0", "false", "f", "no", "n", "off"}
 
+OVERRIDE_TEMP_DIR: Path | None = (
+    None
+    if "AIRBYTE_TEMP_DIR" not in os.environ
+    else Path(os.environ["AIRBYTE_TEMP_DIR"])
+)
+"""The directory to use for temporary files.
+
+This value is read from the `AIRBYTE_TEMP_DIR` environment variable. If the variable is not set,
+Tempfile will use the system's default temporary directory.
+
+This can be useful if you want to store temporary files in a specific location (or) when you
+need your temporary files to exist in user level directories, and not in system level directories for 
+permissions reasons.
+"""
 
 TEMP_FILE_CLEANUP = _str_to_bool(
     os.getenv(