client.py.j2

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

# Imports
import os

from typing import (
    Any,
    Dict,
    List,
    Optional,
    Sequence,
    Tuple,
    Union,
)

# Service Client Imports
{% for imp in service_imports %}{{ imp }}{% endfor %}

# Helper Imports
from . import _helpers

# Type Imports
{% for imp in type_imports %}{{ imp }}{% endfor %}

from google.api_core import client_options as client_options_lib
from google.api_core import gapic_v1
from google.api_core import retry as retries
from google.auth import credentials as auth_credentials
import google.auth

# Create type aliases
try:
    OptionalRetry = Union[retries.Retry, gapic_v1.method._MethodDefault, None]
except AttributeError:  # pragma: NO COVER
    OptionalRetry = Union[retries.Retry, object, None]  # type: ignore

DEFAULT_RETRY: OptionalRetry = gapic_v1.method.DEFAULT
DEFAULT_TIMEOUT: Union[float, object] = gapic_v1.method.DEFAULT
DEFAULT_METADATA: Sequence[Tuple[str, Union[str, bytes]]] = ()

{#- Create a mapping from the ServiceClient class name to its property name on the main client.
    e.g., {'DatasetServiceClient': 'dataset_service_client'}
-#}
{% set class_to_instance_map = {} %}
{% for service in services %}
    {% set _ = class_to_instance_map.update({service.service_client_class: service.property_name}) %}
{% endfor %}


class BigQueryClient:
    def __init__(
        self,
        project: Optional[str] = None,
        credentials: Optional[auth_credentials.Credentials] = None,
        client_options: Optional[client_options_lib.ClientOptions] = None,
    ):
        if credentials is None:
            credentials, project_id = google.auth.default()
        else:
            project_id = None  # project_id is not available from non-default credentials

        if project is None:
            project = project_id

        self.project = project
        self._credentials = credentials
        self._client_options = client_options
        self._clients: Dict[str, Any] = {}

    # --- *METHOD SECTION ---

{% for method in methods %}
{% if method.request_id_args is not none and method.request_id_args|length > 0 %}
{% filter indent(4, True) %}
    {% include 'partials/_method_with_request_builder.j2' %}
{% endfilter %}
{% else %}
{% filter indent(4, True) %}
    {% include 'partials/_simple_passthrough_method.j2' %}
{% endfilter %}
{% endif %}


{% endfor %}

{#- *ServiceClient Properties Section: methods to get/set service clients -#}
    # --- *SERVICECLIENT PROPERTIES ---
{% for service in services %}
    @property
    def {{ service.property_name }}(self):
        if "{{ service.service_name }}" not in self._clients:
            self._clients["{{ service.service_name }}"] = {{ service.service_module_name }}.{{ service.service_client_class }}(
                credentials=self._credentials, client_options=self._client_options
            )
        return self._clients["{{ service.service_name }}"]

    @{{ service.property_name }}.setter
    def {{ service.property_name }}(self, value):
        if not isinstance(value, {{ service.service_module_name }}.{{ service.service_client_class }}):
            raise TypeError(
                "Expected an instance of {{ service.service_client_class }}."
            )
        self._clients["{{ service.service_name }}"] = value

{% endfor %}

{#- Helper Section: methods included from partial template -#}
    {#- include "partials/_client_helpers.j2" #}