dabla commented on code in PR #67016: URL: https://github.com/apache/airflow/pull/67016#discussion_r3836790169
########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,152 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import os +from functools import cached_property +from pathlib import Path + +import structlog + +from airflow.dag_processing.bundles.base import BaseDagBundle +from airflow.providers.microsoft.azure.hooks.wasb import WasbHook + + +class WasbDagBundle(BaseDagBundle): + """ + WASB Dag bundle - exposes a directory in Azure Blob Storage as a Dag bundle. + + This allows Airflow to load Dags directly from an Azure Blob Storage container. + + :param wasb_conn_id: Airflow connection ID for Azure Blob Storage. Defaults to WasbHook.default_conn_name. + :param container_name: The name of the blob container containing the Dag files. + :param prefix: Optional subdirectory within the container where the Dags are stored. + If empty, Dags are assumed to be at the root of the container. + """ + + supports_versioning = False + + def __init__( + self, + *, + wasb_conn_id: str = WasbHook.default_conn_name, + container_name: str, + prefix: str = "", Review Comment: **[warning]** Unresolved concern from previous review: `prefix: str = ""` default is inconsistent with `WasbHook.sync_to_local_dir`. `WasbHook.sync_to_local_dir` already declares `prefix: str | None = None` as its default, using `if prefix:` to distinguish "no prefix" from a real prefix value. Having the bundle always pass `""` instead of `None` when no prefix is given works (since `""` is falsy) but is semantically inconsistent ? it misrepresents "no prefix" as an empty string rather than an absent value. Changing the bundle's default to `None` and propagating that through would make the API more honest: ```suggestion prefix: str | None = None, ``` --- Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before posting ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,152 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import os +from functools import cached_property +from pathlib import Path + +import structlog + +from airflow.dag_processing.bundles.base import BaseDagBundle +from airflow.providers.microsoft.azure.hooks.wasb import WasbHook + + +class WasbDagBundle(BaseDagBundle): + """ + WASB Dag bundle - exposes a directory in Azure Blob Storage as a Dag bundle. + + This allows Airflow to load Dags directly from an Azure Blob Storage container. + + :param wasb_conn_id: Airflow connection ID for Azure Blob Storage. Defaults to WasbHook.default_conn_name. + :param container_name: The name of the blob container containing the Dag files. + :param prefix: Optional subdirectory within the container where the Dags are stored. + If empty, Dags are assumed to be at the root of the container. + """ + + supports_versioning = False + + def __init__( + self, + *, + wasb_conn_id: str = WasbHook.default_conn_name, + container_name: str, + prefix: str = "", + **kwargs, + ) -> None: + super().__init__(**kwargs) + self.wasb_conn_id = wasb_conn_id + self.container_name = container_name + self.prefix = prefix + self.wasb_dags_dir: Path = self.base_dir + + log = structlog.get_logger(__name__) + self._log = log.bind( + bundle_name=self.name, + version=self.version, + container_name=self.container_name, + prefix=self.prefix, + wasb_conn_id=self.wasb_conn_id, + ) + + def _initialize(self): + with self.lock(): + if not self.wasb_dags_dir.exists(): + self._log.info("Creating local Dags directory: %s", self.wasb_dags_dir) Review Comment: **[warning]** Structlog bound logger called with printf-style positional args ? this is the stdlib logging convention, not structlog's native API. Structlog's native API takes a message and keyword arguments: ```python self._log.info("event_name", key=value) ``` Calling it as `self._log.info("msg %s", arg)` passes `arg` as a positional argument that structlog collects into a `positional_args` field (or processes differently depending on the renderer). The `%s` in the message will not be substituted and will appear verbatim in structured output. The same issue applies at line 121 in `refresh()`. Suggested fix: ```python self._log.info("Creating local Dags directory", local_dir=str(self.wasb_dags_dir)) ``` And in `refresh()`: ```python self._log.debug( "Downloading Dags from WASB", container_name=self.container_name, prefix=self.prefix, local_dir=str(self.wasb_dags_dir), ) ``` The tests currently mock `bundle._log.debug` and assert on the positional call signature, which passes regardless of runtime structlog behaviour ? so this bug isn't caught by tests. --- Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before posting ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,152 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import os +from functools import cached_property +from pathlib import Path + +import structlog + +from airflow.dag_processing.bundles.base import BaseDagBundle +from airflow.providers.microsoft.azure.hooks.wasb import WasbHook + + +class WasbDagBundle(BaseDagBundle): + """ + WASB Dag bundle - exposes a directory in Azure Blob Storage as a Dag bundle. + + This allows Airflow to load Dags directly from an Azure Blob Storage container. + + :param wasb_conn_id: Airflow connection ID for Azure Blob Storage. Defaults to WasbHook.default_conn_name. + :param container_name: The name of the blob container containing the Dag files. + :param prefix: Optional subdirectory within the container where the Dags are stored. + If empty, Dags are assumed to be at the root of the container. + """ + + supports_versioning = False + + def __init__( + self, + *, + wasb_conn_id: str = WasbHook.default_conn_name, + container_name: str, + prefix: str = "", + **kwargs, + ) -> None: + super().__init__(**kwargs) + self.wasb_conn_id = wasb_conn_id + self.container_name = container_name + self.prefix = prefix + self.wasb_dags_dir: Path = self.base_dir + + log = structlog.get_logger(__name__) + self._log = log.bind( + bundle_name=self.name, + version=self.version, + container_name=self.container_name, + prefix=self.prefix, + wasb_conn_id=self.wasb_conn_id, + ) + + def _initialize(self): + with self.lock(): + if not self.wasb_dags_dir.exists(): + self._log.info("Creating local Dags directory: %s", self.wasb_dags_dir) + os.makedirs(self.wasb_dags_dir) + + if not self.wasb_dags_dir.is_dir(): + raise NotADirectoryError(f"Local Dags path: {self.wasb_dags_dir} is not a directory.") + + if not self.wasb_hook.check_for_container(container_name=self.container_name): + raise ValueError(f"WASB container '{self.container_name}' does not exist.") + + if self.prefix: + if not self.wasb_hook.check_for_prefix( + container_name=self.container_name, prefix=self.prefix, delimiter="/" + ): + raise ValueError( + f"WASB prefix 'wasb://{self.container_name}/{self.prefix}' does not exist." + ) + self.refresh() + + def initialize(self) -> None: + self._initialize() + super().initialize() + + @cached_property + def wasb_hook(self) -> WasbHook: + return WasbHook(wasb_conn_id=self.wasb_conn_id) + + def __repr__(self): + return ( + f"<WasbDagBundle(" + f"name={self.name!r}, " + f"container_name={self.container_name!r}, " + f"prefix={self.prefix!r}, " + f"version={self.version!r}" + f")>" + ) + + def get_current_version(self) -> str | None: + """Return the current version of the Dag bundle. Currently not supported.""" + return None + + @property + def path(self) -> Path: + """Return the local path to the Dag files.""" + return self.wasb_dags_dir + + def refresh(self) -> None: + """Refresh the Dag bundle by re-downloading the Dags from Azure Blob Storage.""" + if self.version: + raise ValueError("Refreshing a specific version is not supported") + + with self.lock(): + self._log.debug( + "Downloading Dags from wasb://%s/%s to %s", + self.container_name, + self.prefix, + self.wasb_dags_dir, + ) + self.wasb_hook.sync_to_local_dir( + container_name=self.container_name, + prefix=self.prefix, + local_dir=self.wasb_dags_dir, + delete_stale=True, + ) + + def view_url(self, version: str | None = None) -> str | None: + """ + Return a URL for viewing the Dags in Azure Blob Storage. Currently, versioning is not supported. + + This method is deprecated and will be removed when the minimum supported Airflow version is 3.1. + Use `view_url_template` instead. + """ + return self.view_url_template() + + def view_url_template(self) -> str | None: + """Return a URL for viewing the Dags in Azure Blob Storage. Currently, versioning is not supported.""" + if self.version: + raise ValueError("WASB url with version is not supported") Review Comment: **[nit]** Dead code: `_view_url_template` is never assigned anywhere in this class. ```python if hasattr(self, "_view_url_template") and self._view_url_template: return self._view_url_template ``` `self._view_url_template` is never set, so this branch is never taken. It appears to have been copied from another bundle implementation where that attribute is meaningful. Safe to remove. --- Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before posting ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/wasb.py: ########## @@ -463,6 +475,98 @@ def download( # TODO: rework the interface as it might also return Awaitable return blob_client.download_blob(offset=offset, length=length, **kwargs) # type: ignore[return-value] + def _sync_to_local_dir_delete_stale_local_files( + self, current_wasb_objects: list[Path], local_dir: Path + ) -> None: + current_wasb_keys = {key.resolve() for key in current_wasb_objects} + + for item in local_dir.rglob("*"): Review Comment: **[nit]** Log message says "Deleted" before the deletion has actually happened. If `item.unlink()` raises, the log will claim a deletion that never occurred. Use present-tense wording or log after the operation: ```suggestion self.log.debug("Deleting stale local file: %s", item) ``` Same pattern at line 489 for the empty-directory removal. --- Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla 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]
