As of January 1, 2020 this library no longer supports Python 2 on the latest released version.
Library versions released prior to that date will continue to be available. For more information please
visit Python 2 support on Google Cloud.
Source code for google.cloud.spanner_v1.table
# Copyright 2021 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.
"""User friendly container for Cloud Spanner Table."""
from google.cloud.exceptions import NotFound
from google.cloud.spanner_v1.types import (
Type,
TypeCode,
)
_EXISTS_TEMPLATE = """
SELECT EXISTS(
SELECT TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_NAME = @table_id
)
"""
_GET_SCHEMA_TEMPLATE = "SELECT * FROM {} LIMIT 0"
[docs]class Table(object):
"""Representation of a Cloud Spanner Table.
:type table_id: str
:param table_id: The ID of the table.
:type database: :class:`~google.cloud.spanner_v1.database.Database`
:param database: The database that owns the table.
"""
def __init__(self, table_id, database):
self._table_id = table_id
self._database = database
# Calculated properties.
self._schema = None
@property
def table_id(self):
"""The ID of the table used in SQL.
:rtype: str
:returns: The table ID.
"""
return self._table_id
[docs] def exists(self):
"""Test whether this table exists.
:rtype: bool
:returns: True if the table exists, else false.
"""
with self._database.snapshot() as snapshot:
return self._exists(snapshot)
def _exists(self, snapshot):
"""Query to check that the table exists.
:type snapshot: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`
:param snapshot: snapshot to use for database queries
:rtype: bool
:returns: True if the table exists, else false.
"""
results = snapshot.execute_sql(
_EXISTS_TEMPLATE,
params={"table_id": self.table_id},
param_types={"table_id": Type(code=TypeCode.STRING)},
)
return next(iter(results))[0]
@property
def schema(self):
"""The schema of this table.
:rtype: list of :class:`~google.cloud.spanner_v1.types.StructType.Field`
:returns: The table schema.
"""
if self._schema is None:
with self._database.snapshot() as snapshot:
self._schema = self._get_schema(snapshot)
return self._schema
def _get_schema(self, snapshot):
"""Get the schema of this table.
:type snapshot: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`
:param snapshot: snapshot to use for database queries
:rtype: list of :class:`~google.cloud.spanner_v1.types.StructType.Field`
:returns: The table schema.
"""
query = _GET_SCHEMA_TEMPLATE.format(self.table_id)
results = snapshot.execute_sql(query)
# Start iterating to force the schema to download.
try:
next(iter(results))
except StopIteration:
pass
return list(results.fields)
[docs] def reload(self):
"""Reload this table.
Refresh any configured schema into :attr:`schema`.
:raises NotFound: if the table does not exist
"""
with self._database.snapshot() as snapshot:
if not self._exists(snapshot):
raise NotFound("table '{}' does not exist".format(self.table_id))
self._schema = self._get_schema(snapshot)