checkout: Extract data classes to data.py Move data class models and Context container classes from api.py into a dedicated data.py helper module. Change-Id: I4ee2f96e2abd551b8f7d067fce34a5d0de7f998a Reviewed-on: https://pigweed-review.googlesource.com/c/infra/recipes/+/417033
diff --git a/recipe_modules/checkout/api.py b/recipe_modules/checkout/api.py index f9831f8..5ec338d 100644 --- a/recipe_modules/checkout/api.py +++ b/recipe_modules/checkout/api.py
@@ -55,7 +55,6 @@ from __future__ import annotations import contextlib -import dataclasses import re import urllib import xml.etree.ElementTree @@ -72,359 +71,19 @@ ) from PB.recipe_modules.pigweed.checkout.options import Options +from . import data + PIGWEED_REMOTE = 'https://pigweed.googlesource.com/pigweed/pigweed' - -def to_dict(obj) -> dict[str, Any]: - try: - # Modifications to the dict returned by the built-in vars() function - # modify the original data structure. Always create a copy for this - # function to return. - return __builtins__['vars'](obj).copy() - except TypeError: # pragma: no cover - keys = [x for x in obj.__slots__ if not x.startswith('__')] - return {k: getattr(obj, k) for k in keys} - - -@dataclasses.dataclass -class Manifest: - remotes: dict[str, Remote] = dataclasses.field(default_factory=dict) - projects: list[Project] = dataclasses.field(default_factory=list) - - def dict(self) -> dict[str, Any]: - return { - 'remotes': {k: v.dict() for k, v in self.remotes.items()}, - 'projects': [x.dict() for x in self.projects], - } - - -class Url: - def __init__(self, url: str, *args, **kwargs): - super().__init__(*args, **kwargs) - self.url: str = url - self.https: str | None = None - - def dict(self) -> dict[str, Any]: - return to_dict(self) - - -@dataclasses.dataclass -class Remote: - """Remote config from manifest.""" - - name: str - fetch: Url - review: str | None = None - revision: str | None = None - alias: str | None = None - - def dict(self) -> dict[str, Any]: - res = to_dict(self) - res['fetch'] = res['fetch'].dict() - return res - - -@dataclasses.dataclass -class Project: - """Key variables describing a repository/project.""" - - name: str - path: str - remote: str - revision: str - upstream: str - url: str | None = None - - def path_object(self, root: config_types.Path) -> config_types.Path: - return root / self.path - - def dict(self) -> dict[str, Any]: - return to_dict(self) - - -def _str_or_none(x: Any | None) -> str | None: - if x is None: - return x - return str(x) - - -def _int_or_none(x: Any | None) -> int | None: - if x is None: - return x - return int(x) - - -@dataclasses.dataclass -class Change: - """Data from buildbucket.""" - - number: int - remote: str | None - ref: str | None - rebase: bool | None = None - project: str | None = None - branch: str | None = None - gerrit_name: str | None = None - submitted: bool = False - patchset: int | None = None - applied: bool = dataclasses.field(default=False, repr=False) - path: str | None = None - base: str | None = None - base_type: str | None = None - is_merge: bool = False - commit_message: str = '' - topic: str | None = None - current_revision: str | None = None - primary: bool = False - submit_requirements: dict[str, str] = dataclasses.field( - default_factory=dict, - ) - uploader: str | None = None - - def __post_init__(self): - self.number = _int_or_none(self.number) - self.remote = _str_or_none(self.remote) - self.ref = _str_or_none(self.ref) - self.branch = _str_or_none(self.branch) - self.gerrit_name = _str_or_none(self.gerrit_name) - self.patchset = _int_or_none(self.patchset) - self.base = _str_or_none(self.base) - self.base_type = _str_or_none(self.base_type) - - @property - def gerrit_host(self) -> str: - return f'https://{self.gerrit_name}-review.googlesource.com' - - @property - def gerrit_url(self) -> str: - if not self.number: - return self.gitiles_url - return f'{self.gerrit_host}/c/{self.number}' - - @property - def gitiles_url(self) -> str: - return f'{self.remote}/+/{self.ref}' - - @property - def name(self) -> str: - return f'{self.gerrit_name}:{self.number}' - - @property - def name_with_path(self) -> str: - return f'{self.name} ({self.path})' - - -@dataclasses.dataclass -class Submodule: - """Submodule properties.""" - - api: recipe_api.RecipeApi = dataclasses.field(repr=False) - hash: str - relative_path: str - path: config_types.Path - name: str - describe: str - remote: str - initialized: bool - modified: bool - conflict: bool - branch: str - url: str - update: str - ignore: str - shallow: bool - fetchRecurseSubmodules: bool - - def __lt__(self, other: Submodule) -> bool: - return (self.relative_path, self.url) < ( - other.relative_path, - other.url, - ) - - -@dataclasses.dataclass -class StatusOfChanges: - """Changes that were applied or not applied.""" - - applied: tuple[Change, ...] - not_applied: tuple[Change, ...] - - -@dataclasses.dataclass(slots=False) -class CheckoutContext: - _api: recipe_api.RecipeApi = dataclasses.field(repr=False) - options: Options | None = None - changes: list[Change] | None = None # List of triggering changes. - top: config_types.Path = None # Actual checkout root. - # Logical checkout root. Usually identical to 'top', but occasionally a - # subdirectory instead. - root: config_types.Path = None - # Which triggering changes were applied or not applied. - status: StatusOfChanges | None = None - # Remotes that should be treated identically. - equivalent_remotes: dict[str, list[str]] | None = dataclasses.field( - default_factory=dict - ) - manifest: Manifest | None = None # Parsed repo manifest. - # Path to a JSON file containing metadata about the triggering changes. - changes_json: config_types.Path | None = None - bazel_overrides: dict[str, config_types.Path] = dataclasses.field( - default_factory=dict - ) - - # Current revision number. - def revision(self) -> str: - if hasattr(self, '_revision'): - return self._revision - - self._revision = self._api.checkout.get_revision(self.top) - return self._revision - - @property - def manifest_path(self) -> config_types.Path: - return self.top / self.options.manifest_file - - def applied_changes(self) -> list[Change]: - return [x for x in self.changes if x.applied] - - # Repo manifest with all projects pinned. - def manifest_snapshot(self): - if not self.options.use_repo: - return None - - if hasattr(self, '_manifest_snapshot'): - return self._manifest_snapshot - - with self._api.context(cwd=self.top): - self._manifest_snapshot = self._api.repo.manifest_snapshot() - return self._manifest_snapshot - - # Equivalent of manifest_snapshot() but not as strictly formatted. - def submodule_snapshot(self): - if self.options.use_repo: - return None - - if hasattr(self, '_submodule_snapshot'): - return self._submodule_snapshot - - with self._api.context(cwd=self.top): - # To get step_test_data line to pass pylint. - raw_io_stream_output = self._api.raw_io.test_api.stream_output_text - - self._submodule_snapshot = ( - self._api.git( - 'submodule-status', - 'submodule', - 'status', - '--recursive', - stdout=self._api.raw_io.output_text(), - step_test_data=lambda: raw_io_stream_output( - 'submodule status filler text', - ), - ok_ret='any', - ).stdout.strip() - or '' - ) - return self._submodule_snapshot - - def snapshot_to_dir(self, directory: config_types.Path) -> None: - self._api.file.ensure_directory('mkdir', directory) - if self.manifest_snapshot(): - self._api.file.write_text( - 'write manifest.xml', - directory / 'manifest.xml', - self.manifest_snapshot(), - ) - - if self.submodule_snapshot(): - self._api.file.write_text( - 'write submodule snapshot', - directory / 'submodules.log', - self.submodule_snapshot(), - ) - - with self._api.context(cwd=self.top): - log = self._api.git( - 'log', - 'log', - '--oneline', - '-n', - '10', - stdout=self._api.raw_io.output_text(), - ok_ret='any', - ).stdout - self._api.file.write_text( - 'write git log', - directory / 'git.log', - log, - ) - - def submodules(self, recursive: bool = False) -> list[Submodule]: - """Return data about all submodules.""" - - cmd = [ - 'python3', - self._api.checkout.resource('submodule_status.py'), - self.top, - self._api.json.output(), - ] - - if recursive: - cmd.append('--recursive') - - submodules = [] - submodule_status = self._api.step( - 'submodule status', - cmd, - step_test_data=lambda: self._api.json.test_api.output({}), - ).json.output - for sub in submodule_status.values(): - sub['remote'] = self._api.sso.sso_to_https(sub['remote']) - if sub['remote'].endswith('.git'): - sub['remote'] = sub['remote'][:-4] - sub['relative_path'] = sub['path'] - sub['path'] = self.top / sub['path'] - try: - submodules.append(Submodule(self._api, **sub)) - except Exception as exc: # pragma: no cover - self._api.step.empty( - 'exception', - status='FAILURE', - raise_on_failure=False, - ).presentation.step_summary_text = repr(exc) - raise ValueError(f'submodule {sub["path"]}: {sub}') - - return submodules - - _REMOTE_REGEX = re.compile(r'^https://(?P<host>[^/]+)/(?P<project>.+)$') - - def gerrit_host(self) -> str | None: - match = self._REMOTE_REGEX.match(self.options.remote) - if not match: - return # pragma: no cover - - gerrit_review_host = f'{match.group("host")}' - if '-review' not in gerrit_review_host: - gerrit_review_host = gerrit_review_host.replace('.', '-review.', 1) - return gerrit_review_host - - def gerrit_project(self) -> str | None: - match = self._REMOTE_REGEX.match(self.options.remote) - if not match: - return # pragma: no cover - - return match.group('project') - - def remotes_equivalent( - self, - remote1: str | None, - remote2: str | None, - ) -> bool: - return self._api.checkout.remotes_equivalent( - self.equivalent_remotes, - remote1, - remote2, - ) +Change = data.Change +CheckoutContext = data.CheckoutContext +Manifest = data.Manifest +Project = data.Project +Remote = data.Remote +StatusOfChanges = data.StatusOfChanges +Submodule = data.Submodule +Url = data.Url +to_dict = data.to_dict class CheckoutApi(recipe_api.RecipeApi):
diff --git a/recipe_modules/checkout/data.py b/recipe_modules/checkout/data.py new file mode 100644 index 0000000..773ffb4 --- /dev/null +++ b/recipe_modules/checkout/data.py
@@ -0,0 +1,376 @@ +# Copyright 2026 The Pigweed Authors +# +# Licensed 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 +# +# https://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. +"""Data models for the checkout recipe module.""" + +from __future__ import annotations + +import dataclasses +import re +from typing import Any + +from recipe_engine import config_types, recipe_api + +from PB.recipe_modules.pigweed.checkout.options import Options + + +def to_dict(obj) -> dict[str, Any]: + try: + # Modifications to the dict returned by the built-in vars() function + # modify the original data structure. Always create a copy for this + # function to return. + return __builtins__['vars'](obj).copy() + except TypeError: # pragma: no cover + keys = [x for x in obj.__slots__ if not x.startswith('__')] + return {k: getattr(obj, k) for k in keys} + + +@dataclasses.dataclass +class Manifest: + remotes: dict[str, Remote] = dataclasses.field(default_factory=dict) + projects: list[Project] = dataclasses.field(default_factory=list) + + def dict(self) -> dict[str, Any]: + return { + 'remotes': {k: v.dict() for k, v in self.remotes.items()}, + 'projects': [x.dict() for x in self.projects], + } + + +class Url: + def __init__(self, url: str, *args, **kwargs): + super().__init__(*args, **kwargs) + self.url: str = url + self.https: str | None = None + + def dict(self) -> dict[str, Any]: + return to_dict(self) + + +@dataclasses.dataclass +class Remote: + """Remote config from manifest.""" + + name: str + fetch: Url + review: str | None = None + revision: str | None = None + alias: str | None = None + + def dict(self) -> dict[str, Any]: + res = to_dict(self) + res['fetch'] = res['fetch'].dict() + return res + + +@dataclasses.dataclass +class Project: + """Key variables describing a repository/project.""" + + name: str + path: str + remote: str + revision: str + upstream: str + url: str | None = None + + def path_object(self, root: config_types.Path) -> config_types.Path: + return root / self.path + + def dict(self) -> dict[str, Any]: + return to_dict(self) + + +def _str_or_none(x: Any | None) -> str | None: + if x is None: + return x + return str(x) + + +def _int_or_none(x: Any | None) -> int | None: + if x is None: + return x + return int(x) + + +@dataclasses.dataclass +class Change: + """Data from buildbucket.""" + + number: int + remote: str | None + ref: str | None + rebase: bool | None = None + project: str | None = None + branch: str | None = None + gerrit_name: str | None = None + submitted: bool = False + patchset: int | None = None + applied: bool = dataclasses.field(default=False, repr=False) + path: str | None = None + base: str | None = None + base_type: str | None = None + is_merge: bool = False + commit_message: str = '' + topic: str | None = None + current_revision: str | None = None + primary: bool = False + submit_requirements: dict[str, str] = dataclasses.field( + default_factory=dict, + ) + uploader: str | None = None + + def __post_init__(self): + self.number = _int_or_none(self.number) + self.remote = _str_or_none(self.remote) + self.ref = _str_or_none(self.ref) + self.branch = _str_or_none(self.branch) + self.gerrit_name = _str_or_none(self.gerrit_name) + self.patchset = _int_or_none(self.patchset) + self.base = _str_or_none(self.base) + self.base_type = _str_or_none(self.base_type) + + @property + def gerrit_host(self) -> str: + return f'https://{self.gerrit_name}-review.googlesource.com' + + @property + def gerrit_url(self) -> str: + if not self.number: + return self.gitiles_url + return f'{self.gerrit_host}/c/{self.number}' + + @property + def gitiles_url(self) -> str: + return f'{self.remote}/+/{self.ref}' + + @property + def name(self) -> str: + return f'{self.gerrit_name}:{self.number}' + + @property + def name_with_path(self) -> str: + return f'{self.name} ({self.path})' + + +@dataclasses.dataclass +class Submodule: + """Submodule properties.""" + + api: recipe_api.RecipeApi = dataclasses.field(repr=False) + hash: str + relative_path: str + path: config_types.Path + name: str + describe: str + remote: str + initialized: bool + modified: bool + conflict: bool + branch: str + url: str + update: str + ignore: str + shallow: bool + fetchRecurseSubmodules: bool + + def __lt__(self, other: Submodule) -> bool: + return (self.relative_path, self.url) < ( + other.relative_path, + other.url, + ) + + +@dataclasses.dataclass +class StatusOfChanges: + """Changes that were applied or not applied.""" + + applied: tuple[Change, ...] + not_applied: tuple[Change, ...] + + +@dataclasses.dataclass(slots=False) +class CheckoutContext: + _api: recipe_api.RecipeApi = dataclasses.field(repr=False) + options: Options | None = None + changes: list[Change] | None = None # List of triggering changes. + top: config_types.Path = None # Actual checkout root. + # Logical checkout root. Usually identical to 'top', but occasionally a + # subdirectory instead. + root: config_types.Path = None + # Which triggering changes were applied or not applied. + status: StatusOfChanges | None = None + # Remotes that should be treated identically. + equivalent_remotes: dict[str, list[str]] | None = dataclasses.field( + default_factory=dict + ) + manifest: Manifest | None = None # Parsed repo manifest. + # Path to a JSON file containing metadata about the triggering changes. + changes_json: config_types.Path | None = None + bazel_overrides: dict[str, config_types.Path] = dataclasses.field( + default_factory=dict + ) + + # Current revision number. + def revision(self) -> str: + if hasattr(self, '_revision'): + return self._revision + + self._revision = self._api.checkout.get_revision(self.top) + return self._revision + + @property + def manifest_path(self) -> config_types.Path: + return self.top / self.options.manifest_file + + def applied_changes(self) -> list[Change]: + return [x for x in self.changes if x.applied] + + # Repo manifest with all projects pinned. + def manifest_snapshot(self): + if not self.options.use_repo: + return None + + if hasattr(self, '_manifest_snapshot'): + return self._manifest_snapshot + + with self._api.context(cwd=self.top): + self._manifest_snapshot = self._api.repo.manifest_snapshot() + return self._manifest_snapshot + + # Equivalent of manifest_snapshot() but not as strictly formatted. + def submodule_snapshot(self): + if self.options.use_repo: + return None + + if hasattr(self, '_submodule_snapshot'): + return self._submodule_snapshot + + with self._api.context(cwd=self.top): + # To get step_test_data line to pass pylint. + raw_io_stream_output = self._api.raw_io.test_api.stream_output_text + + self._submodule_snapshot = ( + self._api.git( + 'submodule-status', + 'submodule', + 'status', + '--recursive', + stdout=self._api.raw_io.output_text(), + step_test_data=lambda: raw_io_stream_output( + 'submodule status filler text', + ), + ok_ret='any', + ).stdout.strip() + or '' + ) + return self._submodule_snapshot + + def snapshot_to_dir(self, directory: config_types.Path) -> None: + self._api.file.ensure_directory('mkdir', directory) + if self.manifest_snapshot(): + self._api.file.write_text( + 'write manifest.xml', + directory / 'manifest.xml', + self.manifest_snapshot(), + ) + + if self.submodule_snapshot(): + self._api.file.write_text( + 'write submodule snapshot', + directory / 'submodules.log', + self.submodule_snapshot(), + ) + + with self._api.context(cwd=self.top): + log = self._api.git( + 'log', + 'log', + '--oneline', + '-n', + '10', + stdout=self._api.raw_io.output_text(), + ok_ret='any', + ).stdout + self._api.file.write_text( + 'write git log', + directory / 'git.log', + log, + ) + + def submodules(self, recursive: bool = False) -> list[Submodule]: + """Return data about all submodules.""" + + cmd = [ + 'python3', + self._api.checkout.resource('submodule_status.py'), + self.top, + self._api.json.output(), + ] + + if recursive: + cmd.append('--recursive') + + submodules = [] + submodule_status = self._api.step( + 'submodule status', + cmd, + step_test_data=lambda: self._api.json.test_api.output({}), + ).json.output + for sub in submodule_status.values(): + sub['remote'] = self._api.sso.sso_to_https(sub['remote']) + if sub['remote'].endswith('.git'): + sub['remote'] = sub['remote'][:-4] + sub['relative_path'] = sub['path'] + sub['path'] = self.top / sub['path'] + try: + submodules.append(Submodule(self._api, **sub)) + except Exception as exc: # pragma: no cover + self._api.step.empty( + 'exception', + status='FAILURE', + raise_on_failure=False, + ).presentation.step_summary_text = repr(exc) + raise ValueError(f'submodule {sub["path"]}: {sub}') + + return submodules + + _REMOTE_REGEX = re.compile(r'^https://(?P<host>[^/]+)/(?P<project>.+)$') + + def gerrit_host(self) -> str | None: + match = self._REMOTE_REGEX.match(self.options.remote) + if not match: + return # pragma: no cover + + gerrit_review_host = f'{match.group("host")}' + if '-review' not in gerrit_review_host: + gerrit_review_host = gerrit_review_host.replace('.', '-review.', 1) + return gerrit_review_host + + def gerrit_project(self) -> str | None: + match = self._REMOTE_REGEX.match(self.options.remote) + if not match: + return # pragma: no cover + + return match.group('project') + + def remotes_equivalent( + self, + remote1: str | None, + remote2: str | None, + ) -> bool: + return self._api.checkout.remotes_equivalent( + self.equivalent_remotes, + remote1, + remote2, + )