"""Lint CLI.""" import json import logging import os from typing import Annotated import typer from backfill.models import Manifest logger: logging.Logger = logging.getLogger(__name__) lint_cli: typer.Typer = typer.Typer() class ValidateManifestError(Exception): """Error while validating a manifest file.""" pass @lint_cli.command("manifest", short_help="Validate a manifest.json using the linter") def lint_manifest( manifest_file_path: Annotated[ str, typer.Option( help="Path to manifest json file to lint.", ), ], check_keys_exist_locally: Annotated[ bool, typer.Option( help="Set to confirm the manifest file's keys exist locally", ), ] = False, ) -> None: """CLI command to lint a manifest.json.""" try: manifest = _validate_manifest(manifest_file_path) except Exception as ex: logger.error("Failed to validate", ex) raise typer.Exit(1) from ex if check_keys_exist_locally: do_keys_exist = _do_manifest_keys_exist(manifest) if not do_keys_exist: logger.error("Not all keys referenced by manifest exist locally") raise typer.Exit(1) def _validate_manifest(manifest_file_path: str) -> Manifest: """Helper fn for validating manifest file.""" if not os.path.isfile(manifest_file_path): raise ValidateManifestError("missing file %s", manifest_file_path) try: with open(manifest_file_path) as fp: manifest_file_contents = json.load(fp) logger.info("successfully loaded manifest from %s", manifest_file_path) except Exception as ex: raise ValidateManifestError( "failed to load file %s", manifest_file_path ) from ex try: manifest = Manifest.model_validate(manifest_file_contents) except Exception as ex: raise ValidateManifestError( "does not comply with manifest file format", manifest_file_path ) from ex logger.info("successfully validated manifest file %s", manifest_file_path) return manifest def _do_manifest_keys_exist(manifest: Manifest) -> bool: """Helper fn for confirming the keys referenced in this manifest file exist.""" missing_keys = [] for job in manifest.jobs: for key in job.keys: if not os.path.isfile(key): missing_keys.append(key) if len(missing_keys): logger.info( "missing keys locally", extra={ "resources": { "missing_keys": missing_keys, } }, ) return False return True