Repository: apache/airflow
File: providers/google/src/airflow/providers/google/cloud/hooks/cloud_run.py (Python)

Code before:
from airflow.providers.google.common.hooks.base_google import (
    PROVIDE_PROJECT_ID,
    GoogleBaseAsyncHook,
    GoogleBaseHook,
)

if TYPE_CHECKING:
    from google.api_core import operation
    from google.api_core.operation_async import AsyncOperation
    from google.cloud.run_v2.services.jobs import pagers


class CloudRunHook(GoogleBaseHook):
    """
    Hook for the Google Cloud Run service.

    :param gcp_conn_id: The connection ID to use when fetching connection info.
    :param impersonation_chain: Optional service account to impersonate using short-term
        credentials, or chained list of accounts required to get the access_token
        of the last account in the list, which will be impersonated in the request.
        If set as a string, the account must grant the originating account
        the Service Account Token Creator IAM role.
        If set as a sequence, the identities from the list must grant
        Service Account Token Creator IAM role to the directly preceding identity, with first
        account from the list granting this role to the originating account.
    :param transport: Optional. The transport to use for API requests. Can be 'rest' or 'grpc'.
        Defaults to 'grpc'. Use 'rest' if gRPC is not available or fails in your environment
        (e.g., Docker containers with certain network configurations).
    """

    def __init__(
        self,
        gcp_conn_id: str = "google_cloud_default",
        impersonation_chain: str | Sequence[str] | None = None,
        transport: Literal["rest", "grpc"] = "grpc",
        **kwargs,
    ) -> None:
        super().__init__(gcp_conn_id=gcp_conn_id, impersonation_chain=impersonation_chain, **kwargs)
        self._client: JobsClient | None = None
        self.transport = transport

    def get_conn(self):
        """
        Retrieve connection to Cloud Run.

        :return: Cloud Run Jobs client object.
        """
        if self._client is None:
            client_kwargs = {
                "credentials": self.get_credentials(),
                "client_info": CLIENT_INFO,

Review comment (suggestion):
In the [docs](https://docs.cloud.google.com/python/docs/reference/run/latest/google.cloud.run_v2.services.jobs.JobsClient#google_cloud_run_v2_services_jobs_JobsClient) it is referenced as: 

> "Optional[Union[str,JobsTransport,Callable[..., JobsTransport]]]
> The transport to use, or a Callable that constructs and returns a new transport. If a Callable is given, it will be called with the same set of initialization arguments as used in the JobsTransport constructor. If set to None, a transport is chosen automatically.

I think it makes sense to replicate similar hook behavior, at least to some extent, like `transport: Literal['rest', 'grpc'] | None = None)` 

Code after:
from airflow.providers.google.common.hooks.base_google import (
    PROVIDE_PROJECT_ID,
    GoogleBaseAsyncHook,
    GoogleBaseHook,
)

if TYPE_CHECKING:
    from google.api_core import operation
    from google.api_core.operation_async import AsyncOperation
    from google.cloud.run_v2.services.jobs import pagers


class CloudRunHook(GoogleBaseHook):
    """
    Hook for the Google Cloud Run service.

    :param gcp_conn_id: The connection ID to use when fetching connection info.
    :param impersonation_chain: Optional service account to impersonate using short-term
        credentials, or chained list of accounts required to get the access_token
        of the last account in the list, which will be impersonated in the request.
        If set as a string, the account must grant the originating account
        the Service Account Token Creator IAM role.
        If set as a sequence, the identities from the list must grant
        Service Account Token Creator IAM role to the directly preceding identity, with first
        account from the list granting this role to the originating account.
    :param transport: Optional. The transport to use for API requests. Can be 'rest' or 'grpc'.
        If set to None, a transport is chosen automatically. Use 'rest' if gRPC is not available
        or fails in your environment (e.g., Docker containers with certain network configurations).
    """

    def __init__(
        self,
        gcp_conn_id: str = "google_cloud_default",
        impersonation_chain: str | Sequence[str] | None = None,
        transport: Literal["rest", "grpc"] | None = None,
        **kwargs,
    ) -> None:
        super().__init__(gcp_conn_id=gcp_conn_id, impersonation_chain=impersonation_chain, **kwargs)
        self._client: JobsClient | None = None
        self.transport = transport

    def get_conn(self):
        """
        Retrieve connection to Cloud Run.

        :return: Cloud Run Jobs client object.
        """
        if self._client is None:
            client_kwargs = {
                "credentials": self.get_credentials(),
                "client_info": CLIENT_INFO,