dabla commented on code in PR #67016: URL: https://github.com/apache/airflow/pull/67016#discussion_r3511184310
########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,154 @@ +# 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 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, + ) + self._wasb_hook: WasbHook | None = None + + 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() + + @property + def wasb_hook(self): + if self._wasb_hook is None: + self._wasb_hook = WasbHook(wasb_conn_id=self.wasb_conn_id) + return self._wasb_hook + + 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") + if hasattr(self, "_view_url_template") and self._view_url_template: + return self._view_url_template + account_url = self.wasb_hook.blob_service_client.url + url = f"{account_url.rstrip('/')}/{self.container_name}" + if self.prefix: + url += f"/{self.prefix}" Review Comment: `return f"{url}/{self.prefix}"` ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,154 @@ +# 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 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__) Review Comment: I would try extending `WasbDagBundle` with `LogginMixin` instead of instantiating log yourself. If in the future we would for example say we use some other kind of logger, the WasbDagBundle would then be also updated without any intervention needed. ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,154 @@ +# 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 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, + ) + self._wasb_hook: WasbHook | None = None + + 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() + + @property + def wasb_hook(self): Review Comment: This could simply become a cached_property I think, no need to keep self._wasb_hook. ########## providers/microsoft/azure/src/airflow/providers/microsoft/azure/bundles/wasb.py: ########## @@ -0,0 +1,154 @@ +# 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 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__) Review Comment: Hmm I see others do it like that as well, maybe something to refactor in another PR? Maybe BaseDagBundle should extend the LogginMixin. ########## 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("*"): + if item.is_file() and item.resolve() not in current_wasb_keys: + self.log.debug("Deleted stale local file: %s", item) + item.unlink() + for root, dirs, _ in os.walk(local_dir, topdown=False): + for d in dirs: + dir_path = os.path.join(root, d) + if not os.listdir(dir_path): + self.log.debug("Deleted stale empty directory: %s", dir_path) + os.rmdir(dir_path) + + def _sync_to_local_dir_if_changed( + self, container_name: str, blob: BlobProperties, local_target_path: Path + ) -> None: + should_download = False + download_logs: list[str] = [] + download_log_params: list[Any] = [] + + if not local_target_path.exists(): + should_download = True + download_logs.append("Local file %s does not exist.") + download_log_params.append(local_target_path) + else: + local_stats = local_target_path.stat() + if blob.size != local_stats.st_size: + should_download = True + download_logs.append("Blob size (%s) and local file size (%s) differ.") + download_log_params.extend([blob.size, local_stats.st_size]) + + blob_last_modified = blob.last_modified + if blob_last_modified and local_stats.st_mtime < blob_last_modified.timestamp(): + should_download = True + download_logs.append("Blob last modified (%s) and local file last modified (%s) differ.") + download_log_params.extend([blob_last_modified.timestamp(), local_stats.st_mtime]) + + if should_download: + self.get_file( + file_path=str(local_target_path), + container_name=container_name, + blob_name=blob.name, + ) + download_logs.append("Downloaded %s to %s") + download_log_params.extend([blob.name, local_target_path.as_posix()]) + self.log.debug(" ".join(download_logs), *download_log_params) + else: + self.log.debug( + "Local file %s is up-to-date with blob %s. Skipping download.", + local_target_path.as_posix(), + blob.name, + ) + + def sync_to_local_dir( + self, + container_name: str, + local_dir: Path, + prefix: str = "", + delete_stale: bool = True, + ) -> None: + """Download files from an Azure Blob Storage container to a local directory.""" + self.log.debug("Downloading data from wasb://%s/%s to %s", container_name, prefix, local_dir) + + local_wasb_objects: list[Path] = [] + container = self._get_container_client(container_name) + self.check_for_variable_type("container", container, ContainerClient) + container = cast("ContainerClient", container) + + for blob in container.list_blobs(name_starts_with=prefix or None): Review Comment: Couldn't this be avoided if prefix would be None by default instead of empty string? -- 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]
