from google.cloud.spanner_v1 import __version__

from google.cloud.spanner_v1 import ExecuteSqlRequest

from google.cloud.spanner_v1 import DefaultTransactionOptions

from google.cloud.spanner_v1._helpers import _merge_query_options

from google.cloud.spanner_v1._helpers import (

_create_experimental_host_transport,

_merge_query_options,

)

from google.cloud.spanner_v1._helpers import _metadata_with_prefix

from google.cloud.spanner_v1.instance import Instance

from google.cloud.spanner_v1.metrics.constants import (

:raises: :class:`ValueError <exceptions.ValueError>` if both ``read_only``

and ``admin`` are :data:`True`

:type use_plain_text: bool

:param use_plain_text: (Optional) Whether to use plain text for the connection.

This is intended only for experimental host spanner endpoints.

If set, this will override the `api_endpoint` in `client_options`.

If not set, the default behavior is to use TLS.

:type ca_certificate: str

:param ca_certificate: (Optional) The path to the CA certificate file used for TLS connection.

This is intended only for experimental host spanner endpoints.

If set, this will override the `api_endpoint` in `client_options`.

This is mandatory if the experimental_host requires a TLS connection.

:type client_certificate: str

:param client_certificate: (Optional) The path to the client certificate file used for mTLS connection.

This is intended only for experimental host spanner endpoints.

If set, this will override the `api_endpoint` in `client_options`.

This is mandatory if the experimental_host requires a mTLS connection.

:type client_key: str

:param client_key: (Optional) The path to the client key file used for mTLS connection.

This is intended only for experimental host spanner endpoints.

If set, this will override the `api_endpoint` in `client_options`.

This is mandatory if the experimental_host requires a mTLS connection.

"""

_instance_admin_api = None

default_transaction_options: Optional[DefaultTransactionOptions] = None,

experimental_host=None,

disable_builtin_metrics=False,

use_plain_text=False,

ca_certificate=None,

client_certificate=None,

client_key=None,

):

self._emulator_host = _get_spanner_emulator_host()

self._experimental_host = experimental_host

if self._emulator_host:

credentials = AnonymousCredentials()

elif self._experimental_host:

# For all experimental host endpoints project is default

project = "default"

self._use_plain_text = use_plain_text

self._ca_certificate = ca_certificate

self._client_certificate = client_certificate

self._client_key = client_key

credentials = AnonymousCredentials()

elif isinstance(credentials, AnonymousCredentials):

self._emulator_host = self._client_options.api_endpoint

transport=transport,

)

elif self._experimental_host:

transport = InstanceAdminGrpcTransport(

channel=grpc.insecure_channel(target=self._experimental_host)

transport = _create_experimental_host_transport(

InstanceAdminGrpcTransport,

self._experimental_host,

self._use_plain_text,

self._ca_certificate,

self._client_certificate,

self._client_key,

)

self._instance_admin_api = InstanceAdminClient(

client_info=self._client_info,

transport=transport,

)

elif self._experimental_host:

transport = DatabaseAdminGrpcTransport(

channel=grpc.insecure_channel(target=self._experimental_host)

transport = _create_experimental_host_transport(

DatabaseAdminGrpcTransport,

self._experimental_host,

self._use_plain_text,

self._ca_certificate,

self._client_certificate,

self._client_key,

)

self._database_admin_api = DatabaseAdminClient(

client_info=self._client_info,

self._emulator_host,

labels,

processing_units,

self._experimental_host,

)

def list_instances(self, filter_="", page_size=None):