# Copyright 2025 Google LLC

#

# 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

#

# 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.

import argparse

import itertools

import json

import logging

import os

import re

import shutil

import subprocess

import sys

import yaml

from datetime import date, datetime

from functools import lru_cache

from pathlib import Path

from typing import Dict, List

import build.util

logger = logging.getLogger()

BUILD_REQUEST_FILE = "build-request.json"

CONFIGURE_REQUEST_FILE = "configure-request.json"

RELEASE_STAGE_REQUEST_FILE = "release-stage-request.json"

STATE_YAML_FILE = "state.yaml"

INPUT_DIR = "input"

LIBRARIAN_DIR = "librarian"

OUTPUT_DIR = "output"

REPO_DIR = "repo"

SOURCE_DIR = "source"

_GITHUB_BASE = "https://github.com"

_GENERATOR_INPUT_HEADER_TEXT = (

"# DO NOT EDIT THIS FILE OUTSIDE OF `.librarian/generator-input`\n"

"# The source of truth for this file is `.librarian/generator-input`\n"

)

def _read_text_file(path: str) -> str:

"""Helper function that reads a text file path and returns the content.

Args:

path(str): The file path to read.

Returns:

str: The contents of the file.

"""

with open(path, "r") as f:

return f.read()

def _write_text_file(path: str, updated_content: str):

"""Helper function that writes a text file path with the given content.

Args:

path(str): The file path to write.

updated_content(str): The contents to write to the file.

"""

os.makedirs(Path(path).parent, exist_ok=True)

with open(path, "w") as f:

f.write(updated_content)

def _read_json_file(path: str) -> Dict:

"""Helper function that reads a json file path and returns the loaded json content.

Args:

path(str): The file path to read.

Returns:

dict: The parsed JSON content.

Raises:

FileNotFoundError: If the file is not found at the specified path.

json.JSONDecodeError: If the file does not contain valid JSON.

IOError: If there is an issue reading the file.

"""

with open(path, "r") as f:

return json.load(f)

def _write_json_file(path: str, updated_content: Dict):

"""Helper function that writes a json file with the given dictionary.

Args:

path(str): The file path to write.

updated_content(Dict): The dictionary to write.

"""

with open(path, "w") as f:

json.dump(updated_content, f, indent=2)

f.write("\n")

def _add_new_library_source_roots(library_config: Dict, library_id: str) -> None:

"""Adds the default source_roots to the library configuration if not present.

Args:

library_config(Dict): The library configuration.

library_id(str): The id of the library.

"""

if library_config["source_roots"] is None:

library_config["source_roots"] = [f"packages/{library_id}"]

def _add_new_library_preserve_regex(library_config: Dict, library_id: str) -> None:

"""Adds the default preserve_regex to the library configuration if not present.

Args:

library_config(Dict): The library configuration.

library_id(str): The id of the library.

"""

if library_config["preserve_regex"] is None:

library_config["preserve_regex"] = [

f"packages/{library_id}/CHANGELOG.md",

"docs/CHANGELOG.md",

"samples/README.txt",

"scripts/client-post-processing",

"samples/snippets/README.rst",

"tests/system",

]

def _add_new_library_remove_regex(library_config: Dict, library_id: str) -> None:

"""Adds the default remove_regex to the library configuration if not present.

Args:

library_config(Dict): The library configuration.

library_id(str): The id of the library.

"""

if library_config["remove_regex"] is None:

library_config["remove_regex"] = [f"packages/{library_id}"]

def _add_new_library_tag_format(library_config: Dict) -> None:

"""Adds the default tag_format to the library configuration if not present.

Args:

library_config(Dict): The library configuration.

"""

if "tag_format" not in library_config:

library_config["tag_format"] = "{id}-v{version}"

def _get_new_library_config(request_data: Dict) -> Dict:

"""Finds and returns the configuration for a new library.

Args:

request_data(Dict): The request data from which to extract the new

library config.

Returns:

Dict: The unmodified configuration of a new library, or an empty

dictionary if not found.

"""

for library_config in request_data.get("libraries", []):

all_apis = library_config.get("apis", [])

for api in all_apis:

if api.get("status") == "new":

return library_config

return {}

def _add_new_library_version(library_config: Dict) -> None:

"""Adds the library version to the configuration if it's not present.

Args:

library_config(Dict): The library configuration.

"""

if "version" not in library_config or not library_config["version"]:

library_config["version"] = "0.0.0"

def _prepare_new_library_config(library_config: Dict) -> Dict:

"""

Prepares the new library's configuration by removing temporary keys and

adding default values.

Args:

library_config (Dict): The raw library configuration.

Returns:

Dict: The prepared library configuration.

"""

# remove status key from new library config.

all_apis = library_config.get("apis", [])

for api in all_apis:

if "status" in api:

del api["status"]

library_id = _get_library_id(library_config)

_add_new_library_source_roots(library_config, library_id)

_add_new_library_preserve_regex(library_config, library_id)

_add_new_library_remove_regex(library_config, library_id)

_add_new_library_tag_format(library_config)

_add_new_library_version(library_config)

return library_config

def _create_new_changelog_for_library(library_id: str, output: str):

"""Creates a new changelog for the library.

Args:

library_id(str): The id of the library.

output(str): Path to the directory in the container where code

should be generated.

"""

package_changelog_path = f"{output}/packages/{library_id}/CHANGELOG.md"

docs_changelog_path = f"{output}/packages/{library_id}/docs/CHANGELOG.md"

changelog_content = f"# Changelog\n\n[PyPI History][1]\n\n[1]: https://pypi.org/project/{library_id}/#history\n"

os.makedirs(os.path.dirname(package_changelog_path), exist_ok=True)

_write_text_file(package_changelog_path, changelog_content)

os.makedirs(os.path.dirname(docs_changelog_path), exist_ok=True)

_write_text_file(docs_changelog_path, changelog_content)

def handle_configure(

librarian: str = LIBRARIAN_DIR,

source: str = SOURCE_DIR,

repo: str = REPO_DIR,

input: str = INPUT_DIR,

output: str = OUTPUT_DIR,

):

"""Onboards a new library by completing its configuration.

This function reads a partial library definition from `configure-request.json`,

fills in missing fields like the version, source roots, and preservation

rules, and writes the complete configuration to `configure-response.json`.

It ensures that new libraries conform to the repository's standard structure.

See https://github.com/googleapis/librarian/blob/main/doc/container-contract.md#configure-container-command

Args:

librarian(str): Path to the directory in the container which contains

the librarian configuration.

source(str): Path to the directory in the container which contains

API protos.

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

input(str): The path to the directory in the container

which contains additional generator input.

output(str): Path to the directory in the container where code

should be generated.

Raises:

ValueError: If configuring a new library fails.

"""

try:

# configure-request.json contains the library definitions.

request_data = _read_json_file(f"{librarian}/{CONFIGURE_REQUEST_FILE}")

new_library_config = _get_new_library_config(request_data)

_update_global_changelog(

f"{repo}/CHANGELOG.md",

f"{output}/CHANGELOG.md",

[new_library_config],

)

prepared_config = _prepare_new_library_config(new_library_config)

is_mono_repo = _is_mono_repo(input)

library_id = _get_library_id(prepared_config)

path_to_library = f"packages/{library_id}" if is_mono_repo else "."

if not Path(f"{repo}/{path_to_library}").exists():

# Create a `CHANGELOG.md` and `docs/CHANGELOG.md` file for the new library

_create_new_changelog_for_library(library_id, output)

# Write the new library configuration to configure-response.json.

_write_json_file(f"{librarian}/configure-response.json", prepared_config)

except Exception as e:

raise ValueError("Configuring a new library failed.") from e

logger.info("'configure' command executed.")

def _get_library_id(request_data: Dict) -> str:

"""Retrieve the library id from the given request dictionary

Args:

request_data(Dict): The contents `generate-request.json`.

Raises:

ValueError: If the key `id` does not exist in `request_data`.

Returns:

str: The id of the library in `generate-request.json`

"""

library_id = request_data.get("id")

if not library_id:

raise ValueError("Request file is missing required 'id' field.")

return library_id

def _get_repo_metadata_file_path(base: str, library_id: str, is_mono_repo: bool):

"""Constructs the full path to the .repo-metadata.json file.

Args:

base (str): The base directory where the library is located.

library_id (str): The ID of the library.

is_mono_repo (bool): True if the current repository is a mono-repo.

Returns:

str: The absolute path to the .repo-metadata.json file.

"""

path_to_library = f"packages/{library_id}" if is_mono_repo else "."

return f"{base}/{path_to_library}/.repo-metadata.json"

@lru_cache(maxsize=None)

def _get_repo_name_from_repo_metadata(base: str, library_id: str, is_mono_repo: bool):

"""Retrieves the repository name from the .repo-metadata.json file.

This function is cached to avoid redundant file I/O.

Args:

base (str): The base directory where the library is located.

library_id (str): The ID of the library.

is_mono_repo (bool): True if the current repository is a mono-repo.

Returns:

str: The name of the repository (e.g., 'googleapis/google-cloud-python').

Raises:

ValueError: If the '.repo-metadata.json' file is missing the 'repo' field.

"""

if is_mono_repo:

return "googleapis/google-cloud-python"

file_path = _get_repo_metadata_file_path(base, library_id, is_mono_repo)

repo_metadata = _read_json_file(file_path)

repo_name = repo_metadata.get("repo")

if not repo_name:

raise ValueError("`.repo-metadata.json` file is missing required 'repo' field.")

return repo_name

def _run_nox_sessions(library_id: str, repo: str, is_mono_repo: bool):

"""Calls nox for all specified sessions.

Args:

library_id(str): The library id under test.

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global files declared in

the config.yaml.

is_mono_repo(bool): True if the current repository is a mono-repo.

"""

session_runtime = "3.14"

# TODO(https://github.com/googleapis/google-cloud-python/issues/14992): Switch the protobuf

# implementation back to upb once we identify the root cause of the crash that occurs during testing.

# It's not trivial to debug this since it only happens in cloud build.

sessions = [

f"unit-{session_runtime}(protobuf_implementation='python')",

]

current_session = None

try:

for nox_session in sessions:

current_session = nox_session

_run_individual_session(nox_session, library_id, repo, is_mono_repo)

except Exception as e:

raise ValueError(f"Failed to run the nox session: {current_session}") from e

def _run_individual_session(

nox_session: str, library_id: str, repo: str, is_mono_repo: bool

):

"""

Calls nox with the specified sessions.

Args:

nox_session(str): The nox session to run.

library_id(str): The library id under test.

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

is_mono_repo(bool): True if the current repository is a mono-repo.

"""

if is_mono_repo:

path_to_library = f"packages/{library_id}"

library_path = f"{repo}/{path_to_library}"

else:

library_path = repo

command = [

"nox",

"-s",

nox_session,

"-f",

f"{library_path}/noxfile.py",

]

# TODO(#14992): Revert to 600 seconds (10 minutes) after debugging is complete.

result = subprocess.run(command, text=True, check=True, timeout=1200)

logger.info(result)

def _determine_library_namespace(

gapic_parent_path: Path, package_root_path: Path

) -> str:

"""

Determines the namespace from the gapic file's parent path relative

to its package root.

Args:

gapic_parent_path (Path): The absolute path to the directory containing

gapic_version.py (e.g., .../google/cloud/language).

package_root_path (Path): The absolute path to the root of the package

(e.g., .../packages/google-cloud-language).

"""

# This robustly calculates the relative path, e.g., "google/cloud/language"

relative_path = gapic_parent_path.relative_to(package_root_path)

# relative_path.parts will be like: ('google', 'cloud', 'language')

# We want all parts *except* the last one (the service dir) to form the namespace.

namespace_parts = relative_path.parts[:-1]

if not namespace_parts and relative_path.parts:

# This handles the edge case where the parts are just ('google',).

# This implies the namespace is just "google".

return ".".join(relative_path.parts)

return ".".join(namespace_parts)

def _verify_library_namespace(library_id: str, repo: str, is_mono_repo: bool):

"""

Verifies that all found package namespaces are one of

the hardcoded `exception_namespaces` or

`valid_namespaces`.

Args:

library_id (str): The library id under test (e.g., "google-cloud-language").

repo (str): The path to the root of the repository.

is_mono_repo(bool): True if the current repository is a mono-repo.

"""

# TODO(https://github.com/googleapis/google-cloud-python/issues/14376): Update the list of namespaces which are exceptions.

exception_namespaces = [

"google.area120",

"google.api",

"google.apps.script",

"google.apps.script.type",

"google.cloud.alloydb",

"google.cloud.billing",

"google.cloud.devtools",

"google.cloud.gkeconnect",

"google.cloud.gkehub_v1",

"google.cloud.orchestration.airflow",

"google.cloud.orgpolicy",

"google.cloud.security",

"google.cloud.video",

"google.cloud.workflows",

"google.iam",

"google.gapic",

"google.identity.accesscontextmanager",

"google.logging",

"google.monitoring",

"google.rpc",

]

valid_namespaces = [

"google",

"google.ads",

"google.ai",

"google.analytics",

"google.apps",

"google.cloud",

"google.geo",

"google.maps",

"google.pubsub",

"google.shopping",

"grafeas",

*exception_namespaces,

]

gapic_version_file = "gapic_version.py"

proto_file = "*.proto"

if is_mono_repo:

path_to_library = f"packages/{library_id}"

library_path = Path(f"{repo}/{path_to_library}")

else:

library_path = Path(repo)

if not library_path.is_dir():

raise ValueError(f"Error: Path is not a directory: {library_path}")

# Use a set to store unique parent directories of relevant directories

relevant_dirs = set()

# Find all parent directories for 'gapic_version.py' files

for gapic_file in library_path.rglob(gapic_version_file):

relevant_dirs.add(gapic_file.parent)

# Find all parent directories for '*.proto' files

for proto_file in library_path.rglob(proto_file):

proto_path = str(proto_file.parent.relative_to(library_path))

# Exclude proto paths which are not intended to be used for code generation.

# Generally any protos under the `samples` or `tests` directories or in a

# directory called `proto` are not used for code generation.

if (

proto_path.startswith("tests")

or proto_path.startswith("samples")

or proto_path.endswith("proto")

):

continue

relevant_dirs.add(proto_file.parent)

if not relevant_dirs:

raise ValueError(

f"Error: namespace cannot be determined for {library_id}."

f" Library is missing a `{gapic_version_file}` or `{proto_file}` file."

)

for relevant_dir in relevant_dirs:

library_namespace = _determine_library_namespace(relevant_dir, library_path)

if library_namespace not in valid_namespaces:

raise ValueError(

f"The namespace `{library_namespace}` for `{library_id}` must be one of {valid_namespaces}."

)

def _get_library_dist_name(library_id: str, repo: str, is_mono_repo: bool) -> str:

"""

Gets the package name by programmatically building the metadata.

Args:

library_id: id of the library.

repo: This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

is_mono_repo(bool): True if the current repository is a mono-repo.

Returns:

str: The library name string if found, otherwise None.

"""

if is_mono_repo:

path_to_library = f"packages/{library_id}"

library_path = Path(f"{repo}/{path_to_library}")

else:

library_path = Path(repo)

metadata = build.util.project_wheel_metadata(library_path)

return metadata.get("name")

def _verify_library_dist_name(library_id: str, repo: str, is_mono_repo: bool):

"""Verifies the library distribution name against its config files.

This function ensures that:

1. At least one of `setup.py` or `pyproject.toml` exists and is valid.

2. Any existing config file's 'name' property matches the `library_id`.

Args:

library_id: id of the library.

repo: This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

is_mono_repo(bool): True if the current repository is a mono-repo.

Raises:

ValueError: If a name in an existing config file does not match the `library_id`.

"""

dist_name = _get_library_dist_name(library_id, repo, is_mono_repo)

if dist_name != library_id:

raise ValueError(

f"The distribution name `{dist_name}` does not match the folder `{library_id}`."

)

def handle_build(librarian: str = LIBRARIAN_DIR, repo: str = REPO_DIR):

"""The main coordinator for validating client library generation."""

try:

is_mono_repo = _is_mono_repo(repo)

request_data = _read_json_file(f"{librarian}/{BUILD_REQUEST_FILE}")

library_id = _get_library_id(request_data)

_verify_library_namespace(library_id, repo, is_mono_repo)

_verify_library_dist_name(library_id, repo, is_mono_repo)

_run_nox_sessions(library_id, repo, is_mono_repo)

except Exception as e:

raise ValueError("Build failed.") from e

logger.info("'build' command executed.")

def _get_libraries_to_prepare_for_release(library_entries: Dict) -> List[dict]:

"""Get libraries which should be prepared for release. Only libraries

which have the `release_triggered` field set to `True` will be returned.

Args:

library_entries(Dict): Dictionary containing all of the libraries to

evaluate.

Returns:

List[dict]: List of all libraries which should be prepared for release,

along with the corresponding information for the release.

"""

return [

library

for library in library_entries["libraries"]

if library.get("release_triggered")

]

def _update_global_changelog(

changelog_src: str, changelog_dest: str, all_libraries: List[dict]

):

"""Updates the versions of libraries in the main CHANGELOG.md.

Args:

changelog_src(str): Path to the changelog file to read.

changelog_dest(str): Path to the changelog file to write.

all_libraries(Dict): Dictionary containing all of the library versions to

modify.

"""

def replace_version_in_changelog(content):

new_content = content

for library in all_libraries:

library_id = library["id"]

version = library["version"]

# Find the entry for the given library in the format`<library_id>==<version>`

# Replace the `<version>` part of the string.

pattern = re.compile(f"(\\[{re.escape(library_id)})(==)([\\d\\.]+)(\\])")

replacement = f"\\g<1>=={version}\\g<4>"

new_content = pattern.sub(replacement, new_content)

return new_content

updated_content = replace_version_in_changelog(_read_text_file(changelog_src))

_write_text_file(changelog_dest, updated_content)

def _process_version_file(content, version, version_path) -> str:

"""This function searches for a version string in the

given content, replaces the version and returns the content.

Args:

content(str): The contents where the version string should be replaced.

version(str): The new version of the library.

version_path(str): The relative path to the version file

Raises: ValueError if the version string could not be found in the given content

Returns: A string with the modified content.

"""

if version_path.name.endswith("gapic_version.py") or version_path.name.endswith(

"version.py"

):

pattern = r"(__version__\s*=\s*[\"'])([^\"']+)([\"'].*)"

else:

pattern = r"(version\s*=\s*[\"'])([^\"']+)([\"'].*)"

replacement_string = f"\\g<1>{version}\\g<3>"

new_content, num_replacements = re.subn(pattern, replacement_string, content)

if num_replacements == 0:

raise ValueError(

f"Could not find version string in {version_path}. File was not modified."

)

# Optionally update the `__release_date__` date string, if it exists, in the format YYYY-MM-DD

date_pattern = r"(__release_date__\s*=\s*[\"'])([^\"']+)([\"'].*)"

today_iso = date.today().isoformat() # Get today's date in YYYY-MM-DD format

date_replacement_string = f"\\g<1>{today_iso}\\g<3>"

new_content, _ = re.subn(date_pattern, date_replacement_string, new_content)

return new_content

def _update_version_for_library(

repo: str, output: str, path_to_library: str, version: str

):

"""Updates the version string in `**/gapic_version.py`, `**/version.py`, `setup.py`,

`pyproject.toml` and `samples/**/snippet_metadata.json` for a

given library, if applicable.

Args:

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

output(str): Path to the directory in the container where modified

code should be placed.

path_to_library(str): Relative path to the library to update

version(str): The new version of the library

Raises: `ValueError` if a version string could not be located in `**/gapic_version.py`

or `**/version.py` within the given library.

"""

# Find and update version.py or gapic_version.py files

search_base = Path(f"{repo}/{path_to_library}")

version_files = []

patterns = ["**/gapic_version.py", "**/version.py"]

excluded_dirs = {

".nox",

".venv",

"venv",

"site-packages",

".git",

"build",

"dist",

"__pycache__",

"tests",

}

for pattern in patterns:

version_files.extend(

[

p

for p in search_base.rglob(pattern)

if not any(part in excluded_dirs for part in p.parts)

]

)

if not version_files:

# Fallback to `pyproject.toml`` or `setup.py``. Proto-only libraries have

# version information in `setup.py` or `pyproject.toml` instead of `gapic_version.py`.

pyproject_toml = Path(f"{repo}/{path_to_library}/pyproject.toml")

setup_py = Path(f"{repo}/{path_to_library}/setup.py")

version_files = [pyproject_toml if pyproject_toml.exists() else setup_py]

for version_file in version_files:

# Do not process version files in the types directory as some

# GAPIC libraries have `version.py` which are generated from

# `version.proto` and do not include SDK versions.

if version_file.parent.name == "types":

continue

updated_content = _process_version_file(

_read_text_file(version_file), version, version_file

)

output_path = f"{output}/{version_file.relative_to(repo)}"

_write_text_file(output_path, updated_content)

# Find and update snippet_metadata.json files

snippet_metadata_files = Path(f"{repo}/{path_to_library}/samples").rglob(

"**/*snippet*.json"

)

for metadata_file in snippet_metadata_files:

output_path = f"{output}/{metadata_file.relative_to(repo)}"

os.makedirs(Path(output_path).parent, exist_ok=True)

shutil.copy(metadata_file, output_path)

metadata_contents = _read_json_file(metadata_file)

metadata_contents["clientLibrary"]["version"] = version

_write_json_file(output_path, metadata_contents)

def _get_previous_version(library_id: str, librarian: str) -> str:

"""Gets the previous version of the library from state.yaml.

Args:

library_id(str): id of the library.

librarian(str): Path to the directory in the container which contains

the `state.yaml` file.

Returns:

str: The version for a given library in state.yaml

"""

state_yaml_path = f"{librarian}/{STATE_YAML_FILE}"

with open(state_yaml_path, "r") as state_yaml_file:

state_yaml = yaml.safe_load(state_yaml_file)

for library in state_yaml.get("libraries", []):

if library.get("id") == library_id:

return library.get("version")

raise ValueError(

f"Could not determine previous version for {library_id} from state.yaml"

)

def _create_main_version_header(

version: str,

previous_version: str,

library_id: str,

repo_name: str,

tag_format: str,

) -> str:

"""This function creates a header to be used in a changelog. The header has the following format:

`## [{version}](https://github.com/googleapis/google-cloud-python/compare/{tag_format}{previous_version}...{tag_format}{version}) (YYYY-MM-DD)`

Args:

version(str): The new version of the library.

previous_version(str): The previous version of the library.

library_id(str): The id of the library where the changelog should

be updated.

repo_name(str): The name of the repository (e.g., 'googleapis/google-cloud-python').

tag_format(str): The format of the git tag.

Returns:

A header to be used in the changelog.

"""

current_date = datetime.now().strftime("%Y-%m-%d")

# We will assume that version is always at the end of the tag.

tag_format = tag_format.replace("{version}", "")

if "{id}" in tag_format:

tag_format = tag_format.format(**{"id": library_id})

# Return the main version header

return (

f"## [{version}]({_GITHUB_BASE}/{repo_name}/compare/{tag_format}{previous_version}"

f"...{tag_format}{version}) ({current_date})"

)

def _process_changelog(

content: str,

library_changes: List[Dict],

version: str,

previous_version: str,

library_id: str,

repo_name: str,

tag_format: str,

):

"""This function searches the given content for the anchor pattern

`[1]: https://pypi.org/project/{library_id}/#history`

and adds an entry in the following format:

## [{version}](https://github.com/googleapis/google-cloud-python/compare/{tag_format}{previous_version}...{tag_format}{version}) (YYYY-MM-DD)

### Documentation

* Update import statement example in README ([868b006](https://github.com/googleapis/google-cloud-python/commit/868b0069baf1a4bf6705986e0b6885419b35cdcc))

Args:

content(str): The contents of an existing changelog.

library_changes(List[Dict]): List of dictionaries containing the changes

for a given library.

version(str): The new version of the library.

previous_version(str): The previous version of the library.

library_id(str): The id of the library where the changelog should

be updated.

repo_name(str): The name of the repository (e.g., 'googleapis/google-cloud-python').

tag_format(str): The format of the git tag.

Raises: ValueError if the anchor pattern string could not be found in the given content

Returns: A string with the modified content.

"""

entry_parts = []

entry_parts.append(

_create_main_version_header(

version=version,

previous_version=previous_version,

library_id=library_id,

repo_name=repo_name,

tag_format=tag_format,

)

)

# Group changes by type (e.g., feat, fix, docs)

type_key = "type"

commit_hash_key = "commit_hash"

subject_key = "subject"

library_changes.sort(key=lambda x: x[type_key])

grouped_changes = itertools.groupby(library_changes, key=lambda x: x[type_key])

change_type_map = {

"feat": "Features",

"fix": "Bug Fixes",

"docs": "Documentation",

}

for library_change_type, library_changes in grouped_changes:

# We only care about feat, fix, docs

adjusted_change_type = library_change_type.replace("!", "")

if adjusted_change_type in change_type_map:

entry_parts.append(f"\n\n### {change_type_map[adjusted_change_type]}\n")

for change in library_changes:

commit_link = f"([{change[commit_hash_key]}]({_GITHUB_BASE}/{repo_name}/commit/{change[commit_hash_key]}))"

entry_parts.append(f"* {change[subject_key]} {commit_link}")

new_entry_text = "\n".join(entry_parts)

anchor_pattern = re.compile(

rf"(\[1\]: https://pypi\.org/project/{library_id}/#history)",

re.MULTILINE,

)

replacement_text = f"\\g<1>\n\n{new_entry_text}"

updated_content, num_subs = anchor_pattern.subn(replacement_text, content, count=1)

if num_subs == 0:

raise ValueError("Changelog anchor '[1]: ...#history' not found.")

return updated_content

def _update_changelog_for_library(

repo: str,

output: str,

library_changes: List[Dict],

version: str,

previous_version: str,

library_id: str,

is_mono_repo: bool,

tag_format: str,

):

"""Prepends a new release entry with multiple, grouped changes, to a changelog.

Args:

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

output(str): Path to the directory in the container where modified

code should be placed.

library_changes(List[Dict]): List of dictionaries containing the changes

for a given library

version(str): The desired version

previous_version(str): The version in state.yaml for a given library

library_id(str): The id of the library where the changelog should

be updated.

is_mono_repo(bool): True if the current repository is a mono-repo.

tag_format(str): The format of the git tag.

"""

if is_mono_repo:

relative_path = f"packages/{library_id}/CHANGELOG.md"

docs_relative_path = f"packages/{library_id}/docs/CHANGELOG.md"

else:

relative_path = "CHANGELOG.md"

docs_relative_path = f"docs/CHANGELOG.md"

changelog_src = f"{repo}/{relative_path}"

changelog_dest = f"{output}/{relative_path}"

repo_name = _get_repo_name_from_repo_metadata(repo, library_id, is_mono_repo)

updated_content = _process_changelog(

_read_text_file(changelog_src),

library_changes,

version,

previous_version,

library_id,

repo_name,

tag_format,

)

_write_text_file(changelog_dest, updated_content)

docs_changelog_src = f"{repo}/{docs_relative_path}"

if os.path.lexists(docs_changelog_src):

docs_changelog_dst = f"{output}/{docs_relative_path}"

_write_text_file(docs_changelog_dst, updated_content)

def _is_mono_repo(repo: str) -> bool:

"""Determines if a library is generated or handwritten.

Args:

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

Returns: True if the library is generated, False otherwise.

"""

return Path(f"{repo}/packages").exists()

def handle_release_stage(

librarian: str = LIBRARIAN_DIR, repo: str = REPO_DIR, output: str = OUTPUT_DIR

):

"""The main coordinator for the release preparation process.

This function prepares for the release of client libraries by reading a

`librarian/release-stage-request.json` file. The primary responsibility is

to update all required files with the new version and commit information

for libraries that have the `release_triggered` field set to `True`.

See https://github.com/googleapis/librarian/blob/main/doc/container-contract.md#generate-container-command

Args:

librarian(str): Path to the directory in the container which contains

the `release-stage-request.json` file.

repo(str): This directory will contain all directories that make up a

library, the .librarian folder, and any global file declared in

the config.yaml.

output(str): Path to the directory in the container where modified

code should be placed.

Raises:

ValueError: if the version in `release-stage-request.json` is

the same as the version in state.yaml or if the

`release-stage-request.json` file in the given

librarian directory cannot be read.

"""

try:

is_mono_repo = _is_mono_repo(repo)

# Read a release-stage-request.json file

request_data = _read_json_file(f"{librarian}/{RELEASE_STAGE_REQUEST_FILE}")

libraries_to_prep_for_release = _get_libraries_to_prepare_for_release(

request_data

)

if is_mono_repo: