pankajastro commented on code in PR #73370:
URL: https://github.com/apache/airflow/pull/73370#discussion_r4056501805
##########
providers/common/sql/src/airflow/providers/common/sql/datafusion/object_storage_provider.py:
##########
@@ -53,6 +57,53 @@ def get_scheme(self) -> str:
return "s3://"
+class GCSObjectStorageProvider(ObjectStorageProvider):
+ """GCS Object Storage Provider using DataFusion's GoogleCloud."""
+
+ @property
+ def get_storage_type(self) -> StorageType:
+ """Return the storage type."""
+ return StorageType.GCS
+
+ def create_object_store(self, path: str, connection_config:
ConnectionConfig | None = None):
+ """Create a GCS object store using DataFusion's GoogleCloud."""
+ if connection_config is None:
+ raise ValueError(f"connection_config must be provided for
{self.get_storage_type}")
+
+ credentials = connection_config.credentials
+ key_path = credentials.get("key_path")
+ keyfile_dict = credentials.get("keyfile_dict")
+ temp_key_path: str | None = None
+
+ try:
+ bucket = self.get_bucket(path)
+
+ if not key_path and keyfile_dict:
+ # DataFusion's GoogleCloud binding only accepts a file path,
not inline JSON,
+ # so materialize keyfile_dict to a temp file. The credentials
file is read once
+ # at construction time and never touched again, so it's safe
to delete right after.
+ key_content = keyfile_dict if isinstance(keyfile_dict, str)
else json.dumps(keyfile_dict)
+ with tempfile.NamedTemporaryFile(mode="w", suffix=".json",
delete=False) as key_file:
+ key_file.write(key_content)
+ temp_key_path = key_path = key_file.name
+
+ gcs_store = GoogleCloud(bucket_name=bucket,
service_account_path=key_path)
Review Comment:
Fixed — pre-flight `Path.is_file()` check plus widened to `except
BaseException` so the panic is caught cleanly. Updated
`test_gcs_provider_failure` and added a real (unmocked) repro test per your
note.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @pankajastro before posting
##########
providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py:
##########
@@ -169,6 +169,26 @@ def _fetch_extra_configs(keys: list[str]) -> dict[str,
Any]:
credentials = self._remove_none_values(credentials)
extra_config = _fetch_extra_configs(["region", "endpoint"])
+ case "google_cloud_platform":
+ try:
+ from airflow.providers.google.common.hooks.base_google
import get_field
+ except ImportError:
+ from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
+
+ raise AirflowOptionalProviderFeatureException(
+ "Failed to import get_field. To use the GCS storage
functionality, please install "
+ "the apache-airflow-providers-google package."
+ )
+ extra_dejson = conn.extra_dejson
+ key_path = get_field(extra_dejson, "key_path")
+ keyfile_dict = get_field(extra_dejson, "keyfile_dict")
+ if key_path and keyfile_dict:
+ raise ValueError(
+ "The `keyfile_dict` and `key_path` fields are mutually
exclusive. "
+ "Please provide only one value."
+ )
+ credentials = self._remove_none_values({"key_path": key_path,
"keyfile_dict": keyfile_dict})
Review Comment:
Good catch — added an explicit `GOOGLE_APPLICATION_CREDENTIALS` fallback
(only when neither `key_path` nor `keyfile_dict` is set), instead of relying on
the binding's narrower built-in behavior.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @pankajastro before posting
##########
providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py:
##########
@@ -169,6 +169,26 @@ def _fetch_extra_configs(keys: list[str]) -> dict[str,
Any]:
credentials = self._remove_none_values(credentials)
extra_config = _fetch_extra_configs(["region", "endpoint"])
+ case "google_cloud_platform":
+ try:
+ from airflow.providers.google.common.hooks.base_google
import get_field
+ except ImportError:
+ from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
+
+ raise AirflowOptionalProviderFeatureException(
+ "Failed to import get_field. To use the GCS storage
functionality, please install "
+ "the apache-airflow-providers-google package."
+ )
+ extra_dejson = conn.extra_dejson
+ key_path = get_field(extra_dejson, "key_path")
+ keyfile_dict = get_field(extra_dejson, "keyfile_dict")
Review Comment:
Added — raises `ValueError` naming the field for `key_secret_name`,
`credential_config_file`, or `impersonation_chain` instead of falling through
to metadata-server identity.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @pankajastro before posting
##########
providers/common/sql/src/airflow/providers/common/sql/datafusion/engine.py:
##########
@@ -169,6 +169,26 @@ def _fetch_extra_configs(keys: list[str]) -> dict[str,
Any]:
credentials = self._remove_none_values(credentials)
extra_config = _fetch_extra_configs(["region", "endpoint"])
+ case "google_cloud_platform":
+ try:
+ from airflow.providers.google.common.hooks.base_google
import get_field
Review Comment:
Agreed — dropped the dependency entirely and inlined the trivial extras
lookup. Nothing here needs the SDK stack.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @pankajastro before posting
##########
providers/common/sql/src/airflow/providers/common/sql/datafusion/object_storage_provider.py:
##########
@@ -53,6 +57,53 @@ def get_scheme(self) -> str:
return "s3://"
+class GCSObjectStorageProvider(ObjectStorageProvider):
+ """GCS Object Storage Provider using DataFusion's GoogleCloud."""
Review Comment:
Added a GCS Storage section to `operators.rst` with the credential priority
order and an example.
---
Drafted-by: Claude Code (Sonnet 5); reviewed by @pankajastro before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
