Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 95 additions & 0 deletions scripts/microgenerator/templates/_helpers.py.j2
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
# -*- coding: utf-8 -*-
# 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.
#

from typing import Any, Dict, List, Optional, Type
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: I presume we'll need a license header here?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tswast Done, good sir.



def _create_request(
request_class: Type,
path_identifier: str,
expected_args: List[str],
default_project_id: Optional[str] = None,
) -> Any:
"""
Constructs a *Request object from a class, path_identifier, and expected args.

Args:
request_class: The class of the request object to create (e.g., GetDatasetRequest).
path_identifier: The dot-separated string of resource IDs.
expected_args: An ordered list of the argument names the request object
expects (e.g., ['project_id', 'dataset_id', 'table_id']).
default_project_id: The default project ID to use if needed.

Returns:
An instantiated request object.

Examples:
>>> # Example with project_id provided in path_identifier
>>> request = _create_request(
... request_class=GetDatasetRequest,
... path_identifier="my-project.my-dataset",
... expected_args=["project_id", "dataset_id"]
... )
>>> request.project_id
'my-project'
>>> request.dataset_id
'my-dataset'

>>> # Example with project_id omitted from path_identifier, using default_project_id
>>> request = _create_request(
... request_class=GetDatasetRequest,
... path_identifier="my-dataset",
... expected_args=["project_id", "dataset_id"],
... default_project_id="my-default-project"
... )
>>> request.project_id
'my-default-project'
>>> request.dataset_id
'my-dataset'

"""
# Start of inlined parse_path_to_request_inputs
segments = path_identifier.split(".")
num_segments = len(segments)
num_expected = len(expected_args)
project_id_is_expected = "project_id" in expected_args

# Validate the number of segments.
if not (
num_segments == num_expected
or (project_id_is_expected and num_segments == num_expected - 1)
):
raise ValueError(
f"Invalid path identifier '{path_identifier}'. Expected "
f"{num_expected} parts (or {num_expected - 1} if project_id is "
f"omitted), but got {num_segments}."
)

# If project_id is implicitly expected, use the default.
if project_id_is_expected and num_segments == num_expected - 1:
if not default_project_id:
raise ValueError(
f"Missing project_id in path '{path_identifier}' and no "
"default_project_id was provided."
)
# Prepend the default project_id to the segments.
segments.insert(0, default_project_id)

request_inputs = dict(zip(expected_args, segments))
# End of inlined parse_path_to_request_inputs

# Instantiate the request object.
return request_class(**request_inputs)