Rename query file and use async method.

Update to latest version of google-cloud-bigquery.

Author: tswast

bigquery/cloud-client/sync_query_params.py renamed to bigquery/cloud-client/query_params.py

@@ -14,22 +14,34 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Command-line app to perform synchronous queries with parameters in BigQuery.
+"""Command-line app to perform queries with parameters in BigQuery.
 
 For more information, see the README.md under /bigquery.
 
 Example invocation:
-    $ python sync_query_params.py --use-named-params 'romeoandjuliet' 100
-    $ python sync_query_params.py --use-positional-params 'romeoandjuliet' 100
+    $ python query_params.py --use-named-params 'romeoandjuliet' 100
+    $ python query_params.py --use-positional-params 'romeoandjuliet' 100
 """
 
 import argparse
 import datetime
+import time
+import uuid
 
 from google.cloud import bigquery
 import pytz
 
 
+def wait_for_job(job):
+    while True:
+        job.reload()  # Refreshes the state via a GET request.
+        if job.state == 'DONE':
+            if job.error_result:
+                raise RuntimeError(job.errors)
+            return
+        time.sleep(1)
+
+
 def print_results(query_results):
     """Print the query results by requesting a page at a time."""
     page_token = None
@@ -46,15 +58,16 @@ def print_results(query_results):
             break
 
 
-def sync_query_positional_params(corpus, min_word_count):
+def query_positional_params(corpus, min_word_count):
     client = bigquery.Client()
     query = """SELECT word, word_count
     FROM `bigquery-public-data.samples.shakespeare`
     WHERE corpus = ?
     AND word_count >= ?
     ORDER BY word_count DESC;
     """
-    query_results = client.run_sync_query(
+    query_job = client.run_async_query(
+        str(uuid.uuid4()),
         query,
         query_parameters=(
             bigquery.ScalarQueryParameter(
@@ -68,33 +81,40 @@ def sync_query_positional_params(corpus, min_word_count):
 
     # Only standard SQL syntax supports parameters in queries.
     # See: https://cloud.google.com/bigquery/sql-reference/
-    query_results.use_legacy_sql = False
-    query_results.run()
-    print_results(query_results)
+    query_job.use_legacy_sql = False
+
+    # Start the query and wait for the job to complete.
+    query_job.begin()
+    wait_for_job(query_job)
+    print_results(query_job.results())
 
 
-def sync_query_named_params(corpus, min_word_count):
+def query_named_params(corpus, min_word_count):
     client = bigquery.Client()
     query = """SELECT word, word_count
     FROM `bigquery-public-data.samples.shakespeare`
     WHERE corpus = @corpus
     AND word_count >= @min_word_count
     ORDER BY word_count DESC;
     """
-    query_results = client.run_sync_query(
+    query_job = client.run_async_query(
+        str(uuid.uuid4()),
         query,
         query_parameters=(
             bigquery.ScalarQueryParameter('corpus', 'STRING', corpus),
             bigquery.ScalarQueryParameter(
                 'min_word_count',
                 'INT64',
                 min_word_count)))
-    query_results.use_legacy_sql = False
-    query_results.run()
-    print_results(query_results)
+    query_job.use_legacy_sql = False
 
+    # Start the query and wait for the job to complete.
+    query_job.begin()
+    wait_for_job(query_job)
+    print_results(query_job.results())
 
-def sync_query_array_params(gender, states):
+
+def query_array_params(gender, states):
     client = bigquery.Client()
     query = """SELECT name, sum(number) as count
     FROM `bigquery-public-data.usa_names.usa_1910_2013`
@@ -104,45 +124,57 @@ def sync_query_array_params(gender, states):
     ORDER BY count DESC
     LIMIT 10;
     """
-    query_results = client.run_sync_query(
+    query_job = client.run_async_query(
+        str(uuid.uuid4()),
         query,
         query_parameters=(
             bigquery.ScalarQueryParameter('gender', 'STRING', gender),
             bigquery.ArrayQueryParameter('states', 'STRING', states)))
-    query_results.use_legacy_sql = False
-    query_results.run()
-    print_results(query_results)
+    query_job.use_legacy_sql = False
+
+    # Start the query and wait for the job to complete.
+    query_job.begin()
+    wait_for_job(query_job)
+    print_results(query_job.results())
 
 
-def sync_query_timestamp_params(year, month, day, hour, minute):
+def query_timestamp_params(year, month, day, hour, minute):
     client = bigquery.Client()
     query = 'SELECT TIMESTAMP_ADD(@ts_value, INTERVAL 1 HOUR);'
-    query_results = client.run_sync_query(
+    query_job = client.run_async_query(
+        str(uuid.uuid4()),
         query,
         query_parameters=[
             bigquery.ScalarQueryParameter(
                 'ts_value',
                 'TIMESTAMP',
                 datetime.datetime(
                     year, month, day, hour, minute, tzinfo=pytz.UTC))])
-    query_results.use_legacy_sql = False
-    query_results.run()
-    print_results(query_results)
+    query_job.use_legacy_sql = False
 
+    # Start the query and wait for the job to complete.
+    query_job.begin()
+    wait_for_job(query_job)
+    print_results(query_job.results())
 
-def sync_query_struct_params(x, y):
+
+def query_struct_params(x, y):
     client = bigquery.Client()
     query = 'SELECT @struct_value AS s;'
-    query_results = client.run_sync_query(
+    query_job = client.run_async_query(
+        str(uuid.uuid4()),
         query,
         query_parameters=[
             bigquery.StructQueryParameter(
                 'struct_value',
                 bigquery.ScalarQueryParameter('x', 'INT64', x),
                 bigquery.ScalarQueryParameter('y', 'STRING', y))])
-    query_results.use_legacy_sql = False
-    query_results.run()
-    print_results(query_results)
+    query_job.use_legacy_sql = False
+
+    # Start the query and wait for the job to complete.
+    query_job.begin()
+    wait_for_job(query_job)
+    print_results(query_job.results())
 
 
 if __name__ == '__main__':
@@ -197,15 +229,15 @@ def sync_query_struct_params(x, y):
     args = parser.parse_args()
 
     if args.sample == 'named':
-        sync_query_named_params(args.corpus, args.min_word_count)
+        query_named_params(args.corpus, args.min_word_count)
     elif args.sample == 'positional':
-        sync_query_positional_params(args.corpus, args.min_word_count)
+        query_positional_params(args.corpus, args.min_word_count)
     elif args.sample == 'array':
-        sync_query_array_params(args.gender, args.states)
+        query_array_params(args.gender, args.states)
     elif args.sample == 'timestamp':
-        sync_query_timestamp_params(
+        query_timestamp_params(
                 args.year, args.month, args.day, args.hour, args.minute)
     elif args.sample == 'struct':
-        sync_query_struct_params(args.x, args.y)
+        query_struct_params(args.x, args.y)
     else:
         print('Unexpected value for sample')

bigquery/cloud-client/sync_query_params_test.py renamed to bigquery/cloud-client/query_params_test.py

@@ -12,33 +12,41 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import sync_query_params
+import query_params
 
 
-def test_sync_query_named_params(capsys):
-    sync_query_params.sync_query_array_params(
+def test_query_array_params(capsys):
+    query_params.query_array_params(
         gender='M',
         states=['WA', 'WI', 'WV', 'WY'])
     out, _ = capsys.readouterr()
     assert 'James' in out
 
 
-def test_sync_query_positional_params(capsys):
-    sync_query_params.sync_query_positional_params(
+def test_query_named_params(capsys):
+    query_params.query_named_params(
         corpus='romeoandjuliet',
         min_word_count=100)
     out, _ = capsys.readouterr()
     assert 'love' in out
 
 
-def test_sync_query_struct_params(capsys):
-    sync_query_params.sync_query_struct_params(765, "hello world")
+def test_query_positional_params(capsys):
+    query_params.query_positional_params(
+        corpus='romeoandjuliet',
+        min_word_count=100)
+    out, _ = capsys.readouterr()
+    assert 'love' in out
+
+
+def test_query_struct_params(capsys):
+    query_params.query_struct_params(765, "hello world")
     out, _ = capsys.readouterr()
     assert '765' in out
     assert 'hello world' in out
 
 
-def test_sync_query_timestamp_params(capsys):
-    sync_query_params.sync_query_timestamp_params(2016, 12, 7, 8, 0)
+def test_query_timestamp_params(capsys):
+    query_params.query_timestamp_params(2016, 12, 7, 8, 0)
     out, _ = capsys.readouterr()
-    assert '2016-12-07 09:00:00' in out
+    assert '2016, 12, 7, 9, 0' in out

bigquery/cloud-client/requirements.txt

@@ -1,2 +1,2 @@
-google-cloud-bigquery==0.22.1
+google-cloud-bigquery==0.24.0
 pytz==2016.10

scripts/prepare-testing-project.sh

@@ -31,13 +31,13 @@ gcloud alpha bigtable clusters create bigtable-test \
     --zone=us-central1-c
 
 echo "Creating bigquery resources."
-gcloud alpha bigquery datasets create test_dataset
-gcloud alpha bigquery datasets create ephemeral_test_dataset
+bq mk test_dataset
+bq mk ephemeral_test_dataset
 gsutil cp bigquery/api/resources/data.csv gs://$GCLOUD_PROJECT/data.csv
-gcloud alpha bigquery import \
+bq load \
+    test_dataset.test_table \
     gs://$GCLOUD_PROJECT/data.csv \
-    test_dataset/test_table \
-    --schema-file bigquery/api/resources/schema.json
+    bigquery/api/resources/schema.json
 
 echo "Creating datastore indexes."
 gcloud app deploy -q datastore/api/index.yaml