phanikumv commented on code in PR #72476: URL: https://github.com/apache/airflow/pull/72476#discussion_r4067921627
########## airflow-core/src/airflow/dag_processing/bundles/manifest.py: ########## @@ -0,0 +1,412 @@ +# 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 hashlib +import json +import os +import stat +from collections.abc import Iterator, Mapping +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any + +from airflow.exceptions import AirflowException + +MANIFEST_FILE_NAME = ".airflow-bundle-manifest.json" +IGNORED_DIR_NAMES = frozenset({".git", "__pycache__"}) +# ".git" as a file name covers git-worktree checkouts, where .git is a pointer file. +IGNORED_FILE_NAMES = frozenset({MANIFEST_FILE_NAME, ".git"}) +IGNORED_FILE_SUFFIXES = frozenset({".pyc"}) +MANIFEST_SCHEMA_VERSION = 1 +# "sha256-" (not "sha256:") keeps a version usable as a path segment: Airflow builds +# cache and tracking paths from the raw value. +SHA256_VERSION_PREFIX = "sha256-" + + +class BundleManifestError(AirflowException): + """ + Base class for bundle manifest errors. + + Subclasses ``AirflowException`` because every Dag bundle entry point Airflow calls + is expected to fail with one, so a bundle that cannot be read is reported per + bundle rather than escaping as an unrelated error type. + """ + + +class BundleManifestSourceChangedError(BundleManifestError): + """Raised when bundle source files change while a manifest is being built.""" + + +@dataclass(frozen=True) +class BundleSourceFile: + """A source file and the stat metadata used to detect source changes during publishing.""" + + path: Path + relative_path: str + size: int + mtime_ns: int + ctime_ns: int + mode: int + + def build_signature_record(self) -> dict[str, Any]: + return { + "path": self.relative_path, + "size": self.size, + "mtime_ns": self.mtime_ns, + "ctime_ns": self.ctime_ns, + "mode": self.mode, + } + + +@dataclass(frozen=True) +class BundleSourceSnapshot: + """A deterministic snapshot of the source tree metadata.""" + + root: Path + files: tuple[BundleSourceFile, ...] + signature: str + + +def is_ignored_bundle_file_name(file_name: str) -> bool: + """Return whether manifest collection excludes one file, judged by its name alone.""" + return file_name in IGNORED_FILE_NAMES or PurePosixPath(file_name).suffix in IGNORED_FILE_SUFFIXES + + +@dataclass(frozen=True) +class BundleVersionManifest: + """Full manifest plus the compact release-reference payload written to latest.json.""" + + version: str + manifest: dict[str, Any] + ref_payload: dict[str, Any] + source_snapshot: BundleSourceSnapshot + + +def _raise_source_walk_error(error: OSError) -> None: + raise BundleManifestSourceChangedError( + f"Bundle source changed or became unreadable while collecting manifest metadata: {error.filename}" + ) from error + + +def _iter_manifest_file_paths(root: Path) -> Iterator[tuple[Path, os.stat_result]]: + for dirpath, dirnames, filenames in os.walk( + root, + followlinks=False, + onerror=_raise_source_walk_error, + ): + retained_dirnames: list[str] = [] + for dirname in sorted(dirnames): + if dirname in IGNORED_DIR_NAMES: + continue + path = Path(dirpath) / dirname + try: + file_stat = path.lstat() + except FileNotFoundError as e: + raise BundleManifestSourceChangedError( + f"Bundle source directory disappeared while collecting manifest metadata: {path}" + ) from e Review Comment: ```suggestion path = Path(dirpath) / dirname try: file_stat = path.lstat() except OSError as e: raise BundleManifestSourceChangedError( f"Bundle source directory disappeared or became unreadable while collecting manifest metadata: {path}" ) from e ``` -- 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]
