/
table.py
1306 lines (1043 loc) · 50 KB
/
table.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright 2015 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 Google Cloud Bigtable Table."""
from grpc import StatusCode
from google.api_core import timeout
from google.api_core.exceptions import RetryError
from google.api_core.exceptions import NotFound
from google.api_core.retry import if_exception_type
from google.api_core.retry import Retry
from google.api_core.gapic_v1.method import wrap_method
from google.cloud._helpers import _to_bytes
from google.cloud.bigtable.backup import Backup
from google.cloud.bigtable.column_family import _gc_rule_from_pb
from google.cloud.bigtable.column_family import ColumnFamily
from google.cloud.bigtable.batcher import MutationsBatcher
from google.cloud.bigtable.batcher import FLUSH_COUNT, MAX_ROW_BYTES
from google.cloud.bigtable.policy import Policy
from google.cloud.bigtable.row import AppendRow
from google.cloud.bigtable.row import ConditionalRow
from google.cloud.bigtable.row import DirectRow
from google.cloud.bigtable.row_data import PartialRowsData
from google.cloud.bigtable.row_data import DEFAULT_RETRY_READ_ROWS
from google.cloud.bigtable.row_set import RowSet
from google.cloud.bigtable.row_set import RowRange
from google.cloud.bigtable import enums
from google.cloud.bigtable_v2.proto import bigtable_pb2 as data_messages_v2_pb2
from google.cloud.bigtable_admin_v2.gapic.bigtable_table_admin_client import (
BigtableTableAdminClient,
)
from google.cloud.bigtable_admin_v2.proto import table_pb2 as admin_messages_v2_pb2
from google.cloud.bigtable_admin_v2.proto import (
bigtable_table_admin_pb2 as table_admin_messages_v2_pb2,
)
import warnings
# Maximum number of mutations in bulk (MutateRowsRequest message):
# (https://cloud.google.com/bigtable/docs/reference/data/rpc/
# google.bigtable.v2#google.bigtable.v2.MutateRowRequest)
_MAX_BULK_MUTATIONS = 100000
VIEW_NAME_ONLY = enums.Table.View.NAME_ONLY
class _BigtableRetryableError(Exception):
"""Retry-able error expected by the default retry strategy."""
DEFAULT_RETRY = Retry(
predicate=if_exception_type(_BigtableRetryableError),
initial=1.0,
maximum=15.0,
multiplier=2.0,
deadline=120.0, # 2 minutes
)
"""The default retry strategy to be used on retry-able errors.
Used by :meth:`~google.cloud.bigtable.table.Table.mutate_rows`.
"""
class TableMismatchError(ValueError):
"""Row from another table."""
class TooManyMutationsError(ValueError):
"""The number of mutations for bulk request is too big."""
class Table(object):
"""Representation of a Google Cloud Bigtable Table.
.. note::
We don't define any properties on a table other than the name.
The only other fields are ``column_families`` and ``granularity``,
The ``column_families`` are not stored locally and
``granularity`` is an enum with only one value.
We can use a :class:`Table` to:
* :meth:`create` the table
* :meth:`delete` the table
* :meth:`list_column_families` in the table
:type table_id: str
:param table_id: The ID of the table.
:type instance: :class:`~google.cloud.bigtable.instance.Instance`
:param instance: The instance that owns the table.
:type app_profile_id: str
:param app_profile_id: (Optional) The unique name of the AppProfile.
"""
def __init__(self, table_id, instance, mutation_timeout=None, app_profile_id=None):
self.table_id = table_id
self._instance = instance
self._app_profile_id = app_profile_id
self.mutation_timeout = mutation_timeout
@property
def name(self):
"""Table name used in requests.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_name]
:end-before: [END bigtable_table_name]
.. note::
This property will not change if ``table_id`` does not, but the
return value is not cached.
The table name is of the form
``"projects/../instances/../tables/{table_id}"``
:rtype: str
:returns: The table name.
"""
project = self._instance._client.project
instance_id = self._instance.instance_id
table_client = self._instance._client.table_data_client
return table_client.table_path(
project=project, instance=instance_id, table=self.table_id
)
def get_iam_policy(self):
"""Gets the IAM access control policy for this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_get_iam_policy]
:end-before: [END bigtable_table_get_iam_policy]
:rtype: :class:`google.cloud.bigtable.policy.Policy`
:returns: The current IAM policy of this table.
"""
table_client = self._instance._client.table_admin_client
resp = table_client.get_iam_policy(resource=self.name)
return Policy.from_pb(resp)
def set_iam_policy(self, policy):
"""Sets the IAM access control policy for this table. Replaces any
existing policy.
For more information about policy, please see documentation of
class `google.cloud.bigtable.policy.Policy`
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_set_iam_policy]
:end-before: [END bigtable_table_set_iam_policy]
:type policy: :class:`google.cloud.bigtable.policy.Policy`
:param policy: A new IAM policy to replace the current IAM policy
of this table.
:rtype: :class:`google.cloud.bigtable.policy.Policy`
:returns: The current IAM policy of this table.
"""
table_client = self._instance._client.table_admin_client
resp = table_client.set_iam_policy(resource=self.name, policy=policy.to_pb())
return Policy.from_pb(resp)
def test_iam_permissions(self, permissions):
"""Tests whether the caller has the given permissions for this table.
Returns the permissions that the caller has.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_test_iam_permissions]
:end-before: [END bigtable_table_test_iam_permissions]
:type permissions: list
:param permissions: The set of permissions to check for
the ``resource``. Permissions with wildcards (such as '*'
or 'storage.*') are not allowed. For more information see
`IAM Overview
<https://cloud.google.com/iam/docs/overview#permissions>`_.
`Bigtable Permissions
<https://cloud.google.com/bigtable/docs/access-control>`_.
:rtype: list
:returns: A List(string) of permissions allowed on the table.
"""
table_client = self._instance._client.table_admin_client
resp = table_client.test_iam_permissions(
resource=self.name, permissions=permissions
)
return list(resp.permissions)
def column_family(self, column_family_id, gc_rule=None):
"""Factory to create a column family associated with this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_column_family]
:end-before: [END bigtable_table_column_family]
:type column_family_id: str
:param column_family_id: The ID of the column family. Must be of the
form ``[_a-zA-Z0-9][-_.a-zA-Z0-9]*``.
:type gc_rule: :class:`.GarbageCollectionRule`
:param gc_rule: (Optional) The garbage collection settings for this
column family.
:rtype: :class:`.ColumnFamily`
:returns: A column family owned by this table.
"""
return ColumnFamily(column_family_id, self, gc_rule=gc_rule)
def row(self, row_key, filter_=None, append=False):
"""Factory to create a row associated with this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_row]
:end-before: [END bigtable_table_row]
.. warning::
At most one of ``filter_`` and ``append`` can be used in a
:class:`~google.cloud.bigtable.row.Row`.
:type row_key: bytes
:param row_key: The key for the row being created.
:type filter_: :class:`.RowFilter`
:param filter_: (Optional) Filter to be used for conditional mutations.
See :class:`.ConditionalRow` for more details.
:type append: bool
:param append: (Optional) Flag to determine if the row should be used
for append mutations.
:rtype: :class:`~google.cloud.bigtable.row.Row`
:returns: A row owned by this table.
:raises: :class:`ValueError <exceptions.ValueError>` if both
``filter_`` and ``append`` are used.
"""
warnings.warn(
"This method will be deprecated in future versions. Please "
"use Table.append_row(), Table.conditional_row() "
"and Table.direct_row() methods instead.",
PendingDeprecationWarning,
stacklevel=2,
)
if append and filter_ is not None:
raise ValueError("At most one of filter_ and append can be set")
if append:
return AppendRow(row_key, self)
elif filter_ is not None:
return ConditionalRow(row_key, self, filter_=filter_)
else:
return DirectRow(row_key, self)
def append_row(self, row_key):
"""Create a :class:`~google.cloud.bigtable.row.AppendRow` associated with this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_append_row]
:end-before: [END bigtable_table_append_row]
Args:
row_key (bytes): The key for the row being created.
Returns:
A row owned by this table.
"""
return AppendRow(row_key, self)
def direct_row(self, row_key):
"""Create a :class:`~google.cloud.bigtable.row.DirectRow` associated with this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_direct_row]
:end-before: [END bigtable_table_direct_row]
Args:
row_key (bytes): The key for the row being created.
Returns:
A row owned by this table.
"""
return DirectRow(row_key, self)
def conditional_row(self, row_key, filter_):
"""Create a :class:`~google.cloud.bigtable.row.ConditionalRow` associated with this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_table_conditional_row]
:end-before: [END bigtable_table_conditional_row]
Args:
row_key (bytes): The key for the row being created.
filter_ (:class:`.RowFilter`): (Optional) Filter to be used for
conditional mutations. See :class:`.ConditionalRow` for more details.
Returns:
A row owned by this table.
"""
return ConditionalRow(row_key, self, filter_=filter_)
def __eq__(self, other):
if not isinstance(other, self.__class__):
return NotImplemented
return other.table_id == self.table_id and other._instance == self._instance
def __ne__(self, other):
return not self == other
def create(self, initial_split_keys=[], column_families={}):
"""Creates this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_create_table]
:end-before: [END bigtable_create_table]
.. note::
A create request returns a
:class:`._generated.table_pb2.Table` but we don't use
this response.
:type initial_split_keys: list
:param initial_split_keys: (Optional) list of row keys in bytes that
will be used to initially split the table
into several tablets.
:type column_families: dict
:param column_families: (Optional) A map columns to create. The key is
the column_id str and the value is a
:class:`GarbageCollectionRule`
"""
table_client = self._instance._client.table_admin_client
instance_name = self._instance.name
families = {
id: ColumnFamily(id, self, rule).to_pb()
for (id, rule) in column_families.items()
}
table = admin_messages_v2_pb2.Table(column_families=families)
split = table_admin_messages_v2_pb2.CreateTableRequest.Split
splits = [split(key=_to_bytes(key)) for key in initial_split_keys]
table_client.create_table(
parent=instance_name,
table_id=self.table_id,
table=table,
initial_splits=splits,
)
def exists(self):
"""Check whether the table exists.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_check_table_exists]
:end-before: [END bigtable_check_table_exists]
:rtype: bool
:returns: True if the table exists, else False.
"""
table_client = self._instance._client.table_admin_client
try:
table_client.get_table(name=self.name, view=VIEW_NAME_ONLY)
return True
except NotFound:
return False
def delete(self):
"""Delete this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_delete_table]
:end-before: [END bigtable_delete_table]
"""
table_client = self._instance._client.table_admin_client
table_client.delete_table(name=self.name)
def list_column_families(self):
"""List the column families owned by this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_list_column_families]
:end-before: [END bigtable_list_column_families]
:rtype: dict
:returns: Dictionary of column families attached to this table. Keys
are strings (column family names) and values are
:class:`.ColumnFamily` instances.
:raises: :class:`ValueError <exceptions.ValueError>` if the column
family name from the response does not agree with the computed
name from the column family ID.
"""
table_client = self._instance._client.table_admin_client
table_pb = table_client.get_table(self.name)
result = {}
for column_family_id, value_pb in table_pb.column_families.items():
gc_rule = _gc_rule_from_pb(value_pb.gc_rule)
column_family = self.column_family(column_family_id, gc_rule=gc_rule)
result[column_family_id] = column_family
return result
def get_cluster_states(self):
"""List the cluster states owned by this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_get_cluster_states]
:end-before: [END bigtable_get_cluster_states]
:rtype: dict
:returns: Dictionary of cluster states for this table.
Keys are cluster ids and values are
:class: 'ClusterState' instances.
"""
REPLICATION_VIEW = enums.Table.View.REPLICATION_VIEW
table_client = self._instance._client.table_admin_client
table_pb = table_client.get_table(self.name, view=REPLICATION_VIEW)
return {
cluster_id: ClusterState(value_pb.replication_state)
for cluster_id, value_pb in table_pb.cluster_states.items()
}
def read_row(self, row_key, filter_=None):
"""Read a single row from this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_read_row]
:end-before: [END bigtable_read_row]
:type row_key: bytes
:param row_key: The key of the row to read from.
:type filter_: :class:`.RowFilter`
:param filter_: (Optional) The filter to apply to the contents of the
row. If unset, returns the entire row.
:rtype: :class:`.PartialRowData`, :data:`NoneType <types.NoneType>`
:returns: The contents of the row if any chunks were returned in
the response, otherwise :data:`None`.
:raises: :class:`ValueError <exceptions.ValueError>` if a commit row
chunk is never encountered.
"""
row_set = RowSet()
row_set.add_row_key(row_key)
result_iter = iter(self.read_rows(filter_=filter_, row_set=row_set))
row = next(result_iter, None)
if next(result_iter, None) is not None:
raise ValueError("More than one row was returned.")
return row
def read_rows(
self,
start_key=None,
end_key=None,
limit=None,
filter_=None,
end_inclusive=False,
row_set=None,
retry=DEFAULT_RETRY_READ_ROWS,
):
"""Read rows from this table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_read_rows]
:end-before: [END bigtable_read_rows]
:type start_key: bytes
:param start_key: (Optional) The beginning of a range of row keys to
read from. The range will include ``start_key``. If
left empty, will be interpreted as the empty string.
:type end_key: bytes
:param end_key: (Optional) The end of a range of row keys to read from.
The range will not include ``end_key``. If left empty,
will be interpreted as an infinite string.
:type limit: int
:param limit: (Optional) The read will terminate after committing to N
rows' worth of results. The default (zero) is to return
all results.
:type filter_: :class:`.RowFilter`
:param filter_: (Optional) The filter to apply to the contents of the
specified row(s). If unset, reads every column in
each row.
:type end_inclusive: bool
:param end_inclusive: (Optional) Whether the ``end_key`` should be
considered inclusive. The default is False (exclusive).
:type row_set: :class:`.RowSet`
:param row_set: (Optional) The row set containing multiple row keys and
row_ranges.
:type retry: :class:`~google.api_core.retry.Retry`
:param retry:
(Optional) Retry delay and deadline arguments. To override, the
default value :attr:`DEFAULT_RETRY_READ_ROWS` can be used and
modified with the :meth:`~google.api_core.retry.Retry.with_delay`
method or the :meth:`~google.api_core.retry.Retry.with_deadline`
method.
:rtype: :class:`.PartialRowsData`
:returns: A :class:`.PartialRowsData` a generator for consuming
the streamed results.
"""
request_pb = _create_row_request(
self.name,
start_key=start_key,
end_key=end_key,
filter_=filter_,
limit=limit,
end_inclusive=end_inclusive,
app_profile_id=self._app_profile_id,
row_set=row_set,
)
data_client = self._instance._client.table_data_client
return PartialRowsData(data_client.transport.read_rows, request_pb, retry)
def yield_rows(self, **kwargs):
"""Read rows from this table.
.. warning::
This method will be removed in future releases. Please use
``read_rows`` instead.
:type start_key: bytes
:param start_key: (Optional) The beginning of a range of row keys to
read from. The range will include ``start_key``. If
left empty, will be interpreted as the empty string.
:type end_key: bytes
:param end_key: (Optional) The end of a range of row keys to read from.
The range will not include ``end_key``. If left empty,
will be interpreted as an infinite string.
:type limit: int
:param limit: (Optional) The read will terminate after committing to N
rows' worth of results. The default (zero) is to return
all results.
:type filter_: :class:`.RowFilter`
:param filter_: (Optional) The filter to apply to the contents of the
specified row(s). If unset, reads every column in
each row.
:type row_set: :class:`.RowSet`
:param row_set: (Optional) The row set containing multiple row keys and
row_ranges.
:rtype: :class:`.PartialRowData`
:returns: A :class:`.PartialRowData` for each row returned
"""
warnings.warn(
"`yield_rows()` is deprecated; use `read_rows()` instead",
DeprecationWarning,
stacklevel=2,
)
return self.read_rows(**kwargs)
def mutate_rows(self, rows, retry=DEFAULT_RETRY):
"""Mutates multiple rows in bulk.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_mutate_rows]
:end-before: [END bigtable_mutate_rows]
The method tries to update all specified rows.
If some of the rows weren't updated, it would not remove mutations.
They can be applied to the row separately.
If row mutations finished successfully, they would be cleaned up.
Optionally, a ``retry`` strategy can be specified to re-attempt
mutations on rows that return transient errors. This method will retry
until all rows succeed or until the request deadline is reached. To
specify a ``retry`` strategy of "do-nothing", a deadline of ``0.0``
can be specified.
:type rows: list
:param rows: List or other iterable of :class:`.DirectRow` instances.
:type retry: :class:`~google.api_core.retry.Retry`
:param retry:
(Optional) Retry delay and deadline arguments. To override, the
default value :attr:`DEFAULT_RETRY` can be used and modified with
the :meth:`~google.api_core.retry.Retry.with_delay` method or the
:meth:`~google.api_core.retry.Retry.with_deadline` method.
:rtype: list
:returns: A list of response statuses (`google.rpc.status_pb2.Status`)
corresponding to success or failure of each row mutation
sent. These will be in the same order as the `rows`.
"""
retryable_mutate_rows = _RetryableMutateRowsWorker(
self._instance._client,
self.name,
rows,
app_profile_id=self._app_profile_id,
timeout=self.mutation_timeout,
)
return retryable_mutate_rows(retry=retry)
def sample_row_keys(self):
"""Read a sample of row keys in the table.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_sample_row_keys]
:end-before: [END bigtable_sample_row_keys]
The returned row keys will delimit contiguous sections of the table of
approximately equal size, which can be used to break up the data for
distributed tasks like mapreduces.
The elements in the iterator are a SampleRowKeys response and they have
the properties ``offset_bytes`` and ``row_key``. They occur in sorted
order. The table might have contents before the first row key in the
list and after the last one, but a key containing the empty string
indicates "end of table" and will be the last response given, if
present.
.. note::
Row keys in this list may not have ever been written to or read
from, and users should therefore not make any assumptions about the
row key structure that are specific to their use case.
The ``offset_bytes`` field on a response indicates the approximate
total storage space used by all rows in the table which precede
``row_key``. Buffering the contents of all rows between two subsequent
samples would require space roughly equal to the difference in their
``offset_bytes`` fields.
:rtype: :class:`~google.cloud.exceptions.GrpcRendezvous`
:returns: A cancel-able iterator. Can be consumed by calling ``next()``
or by casting to a :class:`list` and can be cancelled by
calling ``cancel()``.
"""
data_client = self._instance._client.table_data_client
response_iterator = data_client.sample_row_keys(
self.name, app_profile_id=self._app_profile_id
)
return response_iterator
def truncate(self, timeout=None):
"""Truncate the table
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_truncate_table]
:end-before: [END bigtable_truncate_table]
:type timeout: float
:param timeout: (Optional) The amount of time, in seconds, to wait
for the request to complete.
:raise: google.api_core.exceptions.GoogleAPICallError: If the
request failed for any reason.
google.api_core.exceptions.RetryError: If the request failed
due to a retryable error and retry attempts failed.
ValueError: If the parameters are invalid.
"""
client = self._instance._client
table_admin_client = client.table_admin_client
if timeout:
table_admin_client.drop_row_range(
self.name, delete_all_data_from_table=True, timeout=timeout
)
else:
table_admin_client.drop_row_range(
self.name, delete_all_data_from_table=True
)
def drop_by_prefix(self, row_key_prefix, timeout=None):
"""
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_drop_by_prefix]
:end-before: [END bigtable_drop_by_prefix]
:type row_key_prefix: bytes
:param row_key_prefix: Delete all rows that start with this row key
prefix. Prefix cannot be zero length.
:type timeout: float
:param timeout: (Optional) The amount of time, in seconds, to wait
for the request to complete.
:raise: google.api_core.exceptions.GoogleAPICallError: If the
request failed for any reason.
google.api_core.exceptions.RetryError: If the request failed
due to a retryable error and retry attempts failed.
ValueError: If the parameters are invalid.
"""
client = self._instance._client
table_admin_client = client.table_admin_client
if timeout:
table_admin_client.drop_row_range(
self.name, row_key_prefix=_to_bytes(row_key_prefix), timeout=timeout
)
else:
table_admin_client.drop_row_range(
self.name, row_key_prefix=_to_bytes(row_key_prefix)
)
def mutations_batcher(self, flush_count=FLUSH_COUNT, max_row_bytes=MAX_ROW_BYTES):
"""Factory to create a mutation batcher associated with this instance.
For example:
.. literalinclude:: snippets_table.py
:start-after: [START bigtable_mutations_batcher]
:end-before: [END bigtable_mutations_batcher]
:type flush_count: int
:param flush_count: (Optional) Maximum number of rows per batch. If it
reaches the max number of rows it calls finish_batch() to
mutate the current row batch. Default is FLUSH_COUNT (1000
rows).
:type max_row_bytes: int
:param max_row_bytes: (Optional) Max number of row mutations size to
flush. If it reaches the max number of row mutations size it
calls finish_batch() to mutate the current row batch.
Default is MAX_ROW_BYTES (5 MB).
"""
return MutationsBatcher(self, flush_count, max_row_bytes)
def backup(self, backup_id, cluster_id=None, expire_time=None):
"""Factory to create a Backup linked to this Table.
:type backup_id: str
:param backup_id: The ID of the Backup to be created.
:type cluster_id: str
:param cluster_id: (Optional) The ID of the Cluster. Required for
calling 'delete', 'exists' etc. methods.
:type expire_time: :class:`datetime.datetime`
:param expire_time: (Optional) The expiration time of this new Backup.
Required, if the `create` method needs to be called.
"""
return Backup(
backup_id,
self._instance,
cluster_id=cluster_id,
table_id=self.table_id,
expire_time=expire_time,
)
def list_backups(self, cluster_id=None, filter_=None, order_by=None, page_size=0):
"""List Backups for this Table.
:type cluster_id: str
:param cluster_id: (Optional) Specifies a single cluster to list
Backups from. If none is specified, the returned list
contains all the Backups in this Instance.
:type filter_: str
:param filter_: (Optional) A filter expression that filters backups
listed in the response. The expression must specify
the field name, a comparison operator, and the value
that you want to use for filtering. The value must be
a string, a number, or a boolean. The comparison
operator must be <, >, <=, >=, !=, =, or :. Colon ':'
represents a HAS operator which is roughly synonymous
with equality. Filter rules are case insensitive.
The fields eligible for filtering are:
- ``name``
- ``source_table``
- ``state``
- ``start_time`` (values of the format YYYY-MM-DDTHH:MM:SSZ)
- ``end_time`` (values of the format YYYY-MM-DDTHH:MM:SSZ)
- ``expire_time`` (values of the format YYYY-MM-DDTHH:MM:SSZ)
- ``size_bytes``
To filter on multiple expressions, provide each
separate expression within parentheses. By default,
each expression is an AND expression. However, you can
include AND, OR, and NOT expressions explicitly.
Some examples of using filters are:
- ``name:"exact"`` --> The Backup name is the string "exact".
- ``name:howl`` --> The Backup name contains the string "howl"
- ``source_table:prod`` --> The source table's name contains
the string "prod".
- ``state:CREATING`` --> The Backup is pending creation.
- ``state:READY`` --> The Backup is created and ready for use.
- ``(name:howl) AND (start_time < \"2020-05-28T14:50:00Z\")``
--> The Backup name contains the string "howl" and
the Backup start time is before 2020-05-28T14:50:00Z.
- ``size_bytes > 10000000000`` --> The Backup size is greater
than 10GB
:type order_by: str
:param order_by: (Optional) An expression for specifying the sort order
of the results of the request. The string value should
specify one or more fields in ``Backup``. The full
syntax is described at https://aip.dev/132#ordering.
Fields supported are: \\* name \\* source_table \\*
expire_time \\* start_time \\* end_time \\*
size_bytes \\* state
For example, "start_time". The default sorting order
is ascending. To specify descending order for the
field, a suffix " desc" should be appended to the
field name. For example, "start_time desc". Redundant
space characters in the syntax are insigificant. If
order_by is empty, results will be sorted by
``start_time`` in descending order starting from
the most recently created backup.
:type page_size: int
:param page_size: (Optional) The maximum number of resources contained
in the underlying API response. If page streaming is
performed per-resource, this parameter does not
affect the return value. If page streaming is
performed per-page, this determines the maximum
number of resources in a page.
:rtype: :class:`~google.api_core.page_iterator.Iterator`
:returns: Iterator of :class:`~google.cloud.bigtable.backup.Backup`
resources within the current Instance.
:raises: :class:`ValueError <exceptions.ValueError>` if one of the
returned Backups' name is not of the expected format.
"""
cluster_id = cluster_id or "-"
backups_filter = "source_table:{}".format(self.name)
if filter_:
backups_filter = "({}) AND ({})".format(backups_filter, filter_)
parent = BigtableTableAdminClient.cluster_path(
project=self._instance._client.project,
instance=self._instance.instance_id,
cluster=cluster_id,
)
client = self._instance._client.table_admin_client
backup_list_pb = client.list_backups(
parent=parent,
filter_=backups_filter,
order_by=order_by,
page_size=page_size,
)
result = []
for backup_pb in backup_list_pb:
result.append(Backup.from_pb(backup_pb, self._instance))
return result
def restore(self, new_table_id, cluster_id=None, backup_id=None, backup_name=None):
"""Creates a new Table by restoring from the Backup specified by either
`backup_id` or `backup_name`. The returned ``long-running operation``
can be used to track the progress of the operation and to cancel it.
The ``response`` type is ``Table``, if successful.
:type new_table_id: str
:param new_table_id: The ID of the Table to create and restore to.
This Table must not already exist.
:type cluster_id: str
:param cluster_id: The ID of the Cluster containing the Backup.
This parameter gets overriden by `backup_name`, if
the latter is provided.
:type backup_id: str
:param backup_id: The ID of the Backup to restore the Table from.
This parameter gets overriden by `backup_name`, if
the latter is provided.
:type backup_name: str
:param backup_name: (Optional) The full name of the Backup to restore
from. If specified, it overrides the `cluster_id`
and `backup_id` parameters even of such specified.
:return: An instance of
:class:`~google.cloud.bigtable_admin_v2.types._OperationFuture`.
:raises: google.api_core.exceptions.AlreadyExists: If the table
already exists.
:raises: google.api_core.exceptions.GoogleAPICallError: If the request
failed for any reason.
:raises: google.api_core.exceptions.RetryError: If the request failed
due to a retryable error and retry attempts failed.
:raises: ValueError: If the parameters are invalid.
"""
api = self._instance._client.table_admin_client
if not backup_name:
backup_name = BigtableTableAdminClient.backup_path(
project=self._instance._client.project,
instance=self._instance.instance_id,
cluster=cluster_id,
backup=backup_id,
)
return api.restore_table(self._instance.name, new_table_id, backup_name)
class _RetryableMutateRowsWorker(object):
"""A callable worker that can retry to mutate rows with transient errors.
This class is a callable that can retry mutating rows that result in
transient errors. After all rows are successful or none of the rows
are retryable, any subsequent call on this callable will be a no-op.
"""
# pylint: disable=unsubscriptable-object
RETRY_CODES = (
StatusCode.DEADLINE_EXCEEDED.value[0],
StatusCode.ABORTED.value[0],
StatusCode.UNAVAILABLE.value[0],
)
# pylint: enable=unsubscriptable-object
def __init__(self, client, table_name, rows, app_profile_id=None, timeout=None):
self.client = client
self.table_name = table_name
self.rows = rows
self.app_profile_id = app_profile_id
self.responses_statuses = [None] * len(self.rows)
self.timeout = timeout
def __call__(self, retry=DEFAULT_RETRY):
"""Attempt to mutate all rows and retry rows with transient errors.
Will retry the rows with transient errors until all rows succeed or
``deadline`` specified in the `retry` is reached.
:rtype: list
:returns: A list of response statuses (`google.rpc.status_pb2.Status`)
corresponding to success or failure of each row mutation
sent. These will be in the same order as the ``rows``.
"""
mutate_rows = self._do_mutate_retryable_rows
if retry:
mutate_rows = retry(self._do_mutate_retryable_rows)