-
Notifications
You must be signed in to change notification settings - Fork 74
refactor(config): typed runtime settings via pydantic-settings #1098
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
de31283
1f847fa
f771566
4507d76
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,22 @@ | ||
| # Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| """Constants shared across the PyAirbyte codebase.""" | ||
| """Constants shared across the PyAirbyte codebase. | ||
|
|
||
| Non-secret runtime settings can be configured in optional `airbyte.yaml` or | ||
| `airbyte.toml` files in the current working directory. Environment variables | ||
| override file values, and file values override defaults. Secret-shaped values | ||
| remain resolved through `airbyte.secrets`. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import logging | ||
| import os | ||
| from pathlib import Path | ||
|
|
||
| from airbyte.settings import Settings | ||
|
|
||
|
|
||
| logger = logging.getLogger("airbyte") | ||
| _SETTINGS = Settings() | ||
|
|
||
|
|
||
| DEBUG_MODE = False # Set to True to enable additional debug logging. | ||
|
|
@@ -59,12 +67,13 @@ | |
|
|
||
|
|
||
| DEFAULT_PROJECT_DIR: Path = _try_create_dir_if_missing( | ||
| Path(os.getenv("AIRBYTE_PROJECT_DIR", "") or Path.cwd()).expanduser().absolute(), | ||
| Path(_SETTINGS.project_dir or Path.cwd()).expanduser().absolute(), | ||
| desc="project", | ||
| ) | ||
| """Default project directory. | ||
|
|
||
| Can be overridden by setting the `AIRBYTE_PROJECT_DIR` environment variable. | ||
| Values use `AIRBYTE_PROJECT_DIR`, then `project_dir` from `airbyte.yaml` or `airbyte.toml`, then | ||
| the current working directory. | ||
|
|
||
| If not set, defaults to the current working directory. | ||
|
|
||
|
|
@@ -76,26 +85,25 @@ | |
|
|
||
|
|
||
| DEFAULT_INSTALL_DIR: Path = _try_create_dir_if_missing( | ||
| Path(os.getenv("AIRBYTE_INSTALL_DIR", "") or DEFAULT_PROJECT_DIR).expanduser().absolute(), | ||
| Path(_SETTINGS.install_dir or DEFAULT_PROJECT_DIR).expanduser().absolute(), | ||
| desc="install", | ||
| ) | ||
| """Default install directory for connectors. | ||
|
|
||
| If not set, defaults to `DEFAULT_PROJECT_DIR` (`AIRBYTE_PROJECT_DIR` env var) or the current | ||
| working directory if neither is set. | ||
| Values use `AIRBYTE_INSTALL_DIR`, then `install_dir` from `airbyte.yaml` or `airbyte.toml`, then | ||
| `DEFAULT_PROJECT_DIR`. | ||
|
|
||
| If a path is specified that does not yet exist, PyAirbyte will attempt to create it. | ||
| """ | ||
|
|
||
|
|
||
| DEFAULT_CACHE_ROOT: Path = ( | ||
| (Path(os.getenv("AIRBYTE_CACHE_ROOT", "") or (DEFAULT_PROJECT_DIR / ".cache"))) | ||
| .expanduser() | ||
| .absolute() | ||
| (Path(_SETTINGS.cache_root or (DEFAULT_PROJECT_DIR / ".cache"))).expanduser().absolute() | ||
| ) | ||
| """Default cache root is `.cache` in the current working directory. | ||
|
|
||
| The default location can be overridden by setting the `AIRBYTE_CACHE_ROOT` environment variable. | ||
| Values use `AIRBYTE_CACHE_ROOT`, then `cache_root` from `airbyte.yaml` or `airbyte.toml`, then the | ||
| default `.cache` path. | ||
|
|
||
| Overriding this can be useful if you always want to store cache files in a specific location. | ||
| For example, in ephemeral environments like Google Colab, you might want to store cache files in | ||
|
|
@@ -115,72 +123,57 @@ | |
| """The default number of records to include in each batch of an Arrow dataset.""" | ||
|
|
||
|
|
||
| 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"} | ||
|
|
||
|
|
||
| TEMP_DIR_OVERRIDE: Path | None = ( | ||
| Path(os.environ["AIRBYTE_TEMP_DIR"]) if os.getenv("AIRBYTE_TEMP_DIR") else None | ||
| ) | ||
| TEMP_DIR_OVERRIDE: Path | None = _SETTINGS.temp_dir | ||
|
github-advanced-security[bot] marked this conversation as resolved.
Fixed
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🚫 Not fixing. Re-raise of the same false positive on the newer commit — |
||
| """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. | ||
| Values use `AIRBYTE_TEMP_DIR`, then `temp_dir` from `airbyte.yaml` or `airbyte.toml`, then 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( | ||
| key="AIRBYTE_TEMP_FILE_CLEANUP", | ||
| default="true", | ||
| ) | ||
| ) | ||
| TEMP_FILE_CLEANUP = _SETTINGS.temp_file_cleanup | ||
| """Whether to clean up temporary files after use. | ||
|
|
||
| This value is read from the `AIRBYTE_TEMP_FILE_CLEANUP` environment variable. If the variable is | ||
| not set, the default value is `True`. | ||
| Values use `AIRBYTE_TEMP_FILE_CLEANUP`, then `temp_file_cleanup` from `airbyte.yaml` or | ||
| `airbyte.toml`, then the default value `True`. | ||
| """ | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| AIRBYTE_OFFLINE_MODE = _str_to_bool( | ||
| os.getenv( | ||
| key="AIRBYTE_OFFLINE_MODE", | ||
| default="false", | ||
| ) | ||
| ) | ||
| AIRBYTE_OFFLINE_MODE = _SETTINGS.offline_mode | ||
| """Enable or disable offline mode. | ||
|
|
||
| When offline mode is enabled, PyAirbyte will attempt to fetch metadata for connectors from the | ||
| Airbyte registry but will not raise an error if the registry is unavailable. This can be useful in | ||
| environments without internet access or with air-gapped networks. | ||
|
|
||
| Offline mode also disables telemetry, similar to a `DO_NOT_TRACK` setting, ensuring no usage data | ||
| is sent from your environment. You may also specify a custom registry URL via the`_REGISTRY_ENV_VAR` | ||
| environment variable if you prefer to use a different registry source for metadata. | ||
| is sent from your environment. You may also specify a custom registry URL via the | ||
| `_REGISTRY_ENV_VAR` environment variable if you prefer to use a different registry source for | ||
| metadata. | ||
|
|
||
| This setting helps you make informed choices about data privacy and operation in restricted and | ||
| air-gapped environments. | ||
|
|
||
| Values use `AIRBYTE_OFFLINE_MODE`, then `offline_mode` from `airbyte.yaml` or `airbyte.toml`, then | ||
| the default value `False`. | ||
| """ | ||
|
|
||
| AIRBYTE_PRINT_FULL_ERROR_LOGS: bool = _str_to_bool( | ||
| os.getenv( | ||
| key="AIRBYTE_PRINT_FULL_ERROR_LOGS", | ||
| default=os.getenv("CI", "false"), | ||
| ) | ||
| ) | ||
| AIRBYTE_PRINT_FULL_ERROR_LOGS: bool = _SETTINGS.print_full_error_logs | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🚫 Not fixing. False positive — |
||
| """Whether to print full error logs when an error occurs. | ||
| This setting helps in debugging by providing detailed logs when errors occur. This is especially | ||
| helpful in ephemeral environments like CI/CD pipelines where log files may not be persisted after | ||
| the pipeline run. | ||
|
|
||
| If not set, the default value is `False` for non-CI environments. | ||
| If running in a CI environment ("CI" env var is set), then the default value is `True`. | ||
|
|
||
| Values use `AIRBYTE_PRINT_FULL_ERROR_LOGS`, then `print_full_error_logs` from `airbyte.yaml` or | ||
| `airbyte.toml`, then the `CI`-derived default. | ||
| """ | ||
|
|
||
| NO_UV: bool = os.getenv("AIRBYTE_NO_UV", "").lower() not in {"1", "true", "yes"} | ||
| NO_UV: bool = _SETTINGS.no_uv | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🚫 Not fixing. False positive — |
||
| """Whether to use uv for Python package management. | ||
|
|
||
| This value is determined by the `AIRBYTE_NO_UV` environment variable. When `AIRBYTE_NO_UV` | ||
|
|
@@ -189,6 +182,9 @@ | |
| If the variable is not set or set to any other value, uv will be used by default. | ||
| This provides a safe fallback mechanism for environments where uv is not available | ||
| or causes issues. | ||
|
|
||
| Values use `AIRBYTE_NO_UV`, then `no_uv` from `airbyte.yaml` or `airbyte.toml`, then the existing | ||
| default. | ||
| """ | ||
|
|
||
| SECRETS_HYDRATION_PREFIX = "secret_reference::" | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| # Copyright (c) 2026 Airbyte, Inc., all rights reserved. | ||
| """Typed runtime settings for PyAirbyte. | ||
|
|
||
| PyAirbyte reads these settings once when `airbyte.constants` is imported. | ||
| Optional `airbyte.yaml` and `airbyte.toml` files in the current working | ||
| directory may provide values for the same settings. Environment variables take | ||
| precedence over file values, and file values take precedence over defaults. | ||
|
|
||
| This module intentionally contains only non-secret runtime settings. Cloud | ||
| credentials, connector credentials, and other secret-shaped values continue to | ||
| resolve through `airbyte.secrets`. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import os | ||
| from pathlib import Path | ||
| from typing import Annotated | ||
|
|
||
| from pydantic import AliasChoices, BeforeValidator, Field | ||
| from pydantic_settings import ( | ||
| BaseSettings, | ||
| PydanticBaseSettingsSource, | ||
| SettingsConfigDict, | ||
| TomlConfigSettingsSource, | ||
| YamlConfigSettingsSource, | ||
| ) | ||
|
|
||
|
|
||
| def _str_to_bool(value: object) -> bool: | ||
| """Convert a value using PyAirbyte's legacy truthiness rules.""" | ||
| if isinstance(value, bool): | ||
| return value | ||
| return bool(value) and str(value).lower() not in {"", "0", "false", "f", "no", "n", "off"} | ||
|
|
||
|
|
||
| def _empty_path_to_none(value: object) -> object: | ||
| """Treat unset and empty path values as absent.""" | ||
| if not value: | ||
| return None | ||
| return value | ||
|
|
||
|
|
||
| def _parse_no_uv(value: object) -> bool: | ||
| """Preserve the inverted legacy AIRBYTE_NO_UV behavior.""" | ||
| if isinstance(value, bool): | ||
| return value | ||
| return str(value).lower() not in {"1", "true", "yes"} | ||
|
Comment on lines
+44
to
+48
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🙋 Human Input Needed: Correct catch, and it exposes a genuine ambiguity I don't want to resolve unilaterally. The root cause is that
Neither is right while the underlying inversion stands. My preference is to fix the inversion itself in a follow-up (so |
||
|
|
||
|
|
||
| BoolSetting = Annotated[bool, BeforeValidator(_str_to_bool)] | ||
| OptionalPathSetting = Annotated[Path | None, BeforeValidator(_empty_path_to_none)] | ||
| NoUvSetting = Annotated[bool, BeforeValidator(_parse_no_uv)] | ||
|
|
||
|
|
||
| class Settings(BaseSettings): | ||
| """Non-secret runtime settings loaded from environment or local config files.""" | ||
|
|
||
| project_dir: OptionalPathSetting = None | ||
| install_dir: OptionalPathSetting = None | ||
| cache_root: OptionalPathSetting = None | ||
| temp_dir: OptionalPathSetting = None | ||
| temp_file_cleanup: BoolSetting = True | ||
| offline_mode: BoolSetting = False | ||
| print_full_error_logs: BoolSetting = Field( | ||
| default_factory=lambda: _str_to_bool(os.getenv("CI", "false")), | ||
| validation_alias=AliasChoices("AIRBYTE_PRINT_FULL_ERROR_LOGS", "print_full_error_logs"), | ||
| ) | ||
| no_uv: NoUvSetting = True | ||
|
|
||
| model_config = SettingsConfigDict( | ||
| env_prefix="AIRBYTE_", | ||
| yaml_file=("airbyte.yaml",), | ||
| toml_file=("airbyte.toml",), | ||
| ) | ||
|
|
||
| @classmethod | ||
| def settings_customise_sources( | ||
| cls, | ||
| settings_cls: type[BaseSettings], | ||
| init_settings: PydanticBaseSettingsSource, | ||
| env_settings: PydanticBaseSettingsSource, | ||
| dotenv_settings: PydanticBaseSettingsSource, | ||
| file_secret_settings: PydanticBaseSettingsSource, | ||
| ) -> tuple[PydanticBaseSettingsSource, ...]: | ||
| """Load local config files below environment and dotenv sources.""" | ||
| return ( | ||
| init_settings, | ||
| env_settings, | ||
| dotenv_settings, | ||
| YamlConfigSettingsSource(settings_cls), | ||
| TomlConfigSettingsSource(settings_cls), | ||
| file_secret_settings, | ||
| ) | ||
Uh oh!
There was an error while loading. Please reload this page.