Add 'Table.upload_from_file'.

docs/bigquery-usage.rst

@@ -224,6 +224,21 @@ Update all writable metadata for a table
  ... SchemaField(name='age', type='int', mode='required)]
  >>> table.update() # API request
 
+Upload table data from a file:
+
+.. doctest::
+
+ >>> from gcloud import bigquery
+ >>> client = bigquery.Client()
+ >>> dataset = client.dataset('dataset_name')
+ >>> table = dataset.table(name='person_ages')
+ >>> table.schema = [
+ ... SchemaField(name='full_name', type='string', mode='required'),
+ ... SchemaField(name='age', type='int', mode='required)]
+ >>> with open('person_ages.csv', 'rb') as csv_file:
+ ... table.upload_from_file(csv_file, CSV,
+ ... create_disposition='CREATE_IF_NEEDED')
+
 Get rows from a table's data:
 
 .. doctest::
@@ -263,7 +278,7 @@ List jobs for a project:
 
  >>> from gcloud import bigquery
  >>> client = bigquery.Client()
- >>> jobs= client.jobs() # API request
+ >>> jobs, token = client.list_jobs() # API request
  >>> [(job.job_id, job.type, job.created, job.state) for job in jobs]
  ['e3344fba-09df-4ae0-8337-fddee34b3840', 'insert', (datetime.datetime(2015, 7, 23, 9, 30, 20, 268260, tzinfo=<UTC>), 'done')]
 

gcloud/bigquery/client.py

@@ -102,20 +102,18 @@ def dataset(self, dataset_name):
  """
  return Dataset(dataset_name, client=self)
 
- def _job_from_resource(self, resource):
+ def job_from_resource(self, resource):
  """Detect correct job type from resource and instantiate.
 
- Helper for :meth:`list_jobs`.
-
  :type resource: dict
  :param resource: one job resource from API response
 
  :rtype; One of:
  :class:`gcloud.bigquery.job.LoadTableFromStorageJob`,
- :class:`gcloud.bigquery.job.CopyJob,
- :class:`gcloud.bigquery.job.ExtractTableToStorageJob,
- :class:`gcloud.bigquery.job.QueryJob,
- :class:`gcloud.bigquery.job.RunSyncQueryJob
+ :class:`gcloud.bigquery.job.CopyJob`,
+ :class:`gcloud.bigquery.job.ExtractTableToStorageJob`,
+ :class:`gcloud.bigquery.job.QueryJob`,
+ :class:`gcloud.bigquery.job.RunSyncQueryJob`
  :returns: the job instance, constructed via the resource
  """
  config = resource['configuration']
@@ -176,8 +174,7 @@ def list_jobs(self, max_results=None, page_token=None, all_users=None,
  path = '/projects/%s/jobs' % (self.project,)
  resp = self.connection.api_request(method='GET', path=path,
  query_params=params)
- jobs = [self._job_from_resource(resource)
- for resource in resp['jobs']]
+ jobs = [self.job_from_resource(resource) for resource in resp['jobs']]
  return jobs, resp.get('nextPageToken')
 
  def load_table_from_storage(self, job_name, destination, *source_uris):

gcloud/bigquery/table.py

@@ -15,13 +15,19 @@
 """Define API Datasets."""
 
 import datetime
+import json
+import os
 
 import six
 
 from gcloud._helpers import _datetime_from_microseconds
 from gcloud._helpers import _microseconds_from_datetime
 from gcloud._helpers import _millis_from_datetime
 from gcloud.exceptions import NotFound
+from gcloud.streaming.http_wrapper import Request
+from gcloud.streaming.http_wrapper import make_api_request
+from gcloud.streaming.transfer import RESUMABLE_UPLOAD
+from gcloud.streaming.transfer import Upload
 from gcloud.bigquery._helpers import _rows_from_json
 
 
@@ -693,6 +699,225 @@ def insert_data(self,
 
  return errors
 
+ def upload_from_file(self, # pylint: disable=R0913,R0914
+ file_obj,
+ source_format,
+ rewind=False,
+ size=None,
+ num_retries=6,
+ allow_jagged_rows=None,
+ allow_quoted_newlines=None,
+ create_disposition=None,
+ encoding=None,
+ field_delimiter=None,
+ ignore_unknown_values=None,
+ max_bad_records=None,
+ quote_character=None,
+ skip_leading_rows=None,
+ write_disposition=None,
+ client=None):
+ """Upload the contents of this table from a file-like object.
+
+ The content type of the upload will either be
+ - The value passed in to the function (if any)
+ - ``text/csv``.
+
+ :type file_obj: file
+ :param file_obj: A file handle open for reading.
+
+ :type source_format: string
+ :param source_format: one of 'CSV' or 'NEWLINE_DELIMITED_JSON'.
+ job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type rewind: boolean
+ :param rewind: If True, seek to the beginning of the file handle before
+ writing the file to Cloud Storage.
+
+ :type size: int
+ :param size: The number of bytes to read from the file handle.
+ If not provided, we'll try to guess the size using
+ :func:`os.fstat`. (If the file handle is not from the
+ filesystem this won't be possible.)
+
+ :type num_retries: integer
+ :param num_retries: Number of upload retries. Defaults to 6.
+
+ :type allow_jagged_rows: boolean
+ :param allow_jagged_rows: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type allow_quoted_newlines: boolean
+ :param allow_quoted_newlines: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type create_disposition: string
+ :param create_disposition: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type encoding: string
+ :param encoding: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type field_delimiter: string
+ :param field_delimiter: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type ignore_unknown_values: boolean
+ :param ignore_unknown_values: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type max_bad_records: integer
+ :param max_bad_records: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type quote_character: string
+ :param quote_character: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type skip_leading_rows: integer
+ :param skip_leading_rows: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type write_disposition: string
+ :param write_disposition: job configuration option; see
+ :meth:`gcloud.bigquery.job.LoadJob`
+
+ :type client: :class:`gcloud.storage.client.Client` or ``NoneType``
+ :param client: Optional. The client to use. If not passed, falls back
+ to the ``client`` stored on the current dataset.
+
+ :rtype: :class:`gcloud.bigquery.jobs.LoadTableFromStorageJob`
+ :returns: the job instance used to load the data (e.g., for
+ querying status)
+ :raises: :class:`ValueError` if size is not passed in and can not be
+ determined
+ """
+ client = self._require_client(client)
+ connection = client.connection
+ content_type = 'application/octet-stream'
+
+ # Rewind the file if desired.
+ if rewind:
+ file_obj.seek(0, os.SEEK_SET)
+
+ # Get the basic stats about the file.
+ total_bytes = size
+ if total_bytes is None:
+ if hasattr(file_obj, 'fileno'):
+ total_bytes = os.fstat(file_obj.fileno()).st_size
+ else:
+ raise ValueError('total bytes could not be determined. Please '
+ 'pass an explicit size.')
+ headers = {
+ 'Accept': 'application/json',
+ 'Accept-Encoding': 'gzip, deflate',
+ 'User-Agent': connection.USER_AGENT,
+ 'content-type': 'application/json',
+ }
+
+ metadata = {
+ 'configuration': {
+ 'load': {
+ 'sourceFormat': source_format,
+ 'schema': {
+ 'fields': _build_schema_resource(self._schema),
+ },
+ 'destinationTable': {
+ 'projectId': self._dataset.project,
+ 'datasetId': self._dataset.name,
+ 'tableId': self.name,
+ }
+ }
+ }
+ }
+
+ _configure_job_metadata(metadata, allow_jagged_rows,
+ allow_quoted_newlines, create_disposition,
+ encoding, field_delimiter,
+ ignore_unknown_values, max_bad_records,
+ quote_character, skip_leading_rows,
+ write_disposition)
+
+ upload = Upload(file_obj, content_type, total_bytes,
+ auto_transfer=False)
+
+ url_builder = _UrlBuilder()
+ upload_config = _UploadConfig()
+
+ # Base URL may change once we know simple vs. resumable.
+ base_url = connection.API_BASE_URL + '/upload'
+ path = '/projects/%s/jobs' % (self._dataset.project,)
+ upload_url = connection.build_api_url(api_base_url=base_url, path=path)
+
+ # Use apitools 'Upload' facility.
+ request = Request(upload_url, 'POST', headers,
+ body=json.dumps(metadata))
+
+ upload.configure_request(upload_config, request, url_builder)
+ query_params = url_builder.query_params
+ base_url = connection.API_BASE_URL + '/upload'
+ request.url = connection.build_api_url(api_base_url=base_url,
+ path=path,
+ query_params=query_params)
+ upload.initialize_upload(request, connection.http)
+
+ if upload.strategy == RESUMABLE_UPLOAD:
+ http_response = upload.stream_file(use_chunks=True)
+ else:
+ http_response = make_api_request(connection.http, request,
+ retries=num_retries)
+ response_content = http_response.content
+ if not isinstance(response_content,
+ six.string_types): # pragma: NO COVER Python3
+ response_content = response_content.decode('utf-8')
+ return client.job_from_resource(json.loads(response_content))
+
+
+def _configure_job_metadata(metadata, # pylint: disable=R0913
+ allow_jagged_rows,
+ allow_quoted_newlines,
+ create_disposition,
+ encoding,
+ field_delimiter,
+ ignore_unknown_values,
+ max_bad_records,
+ quote_character,
+ skip_leading_rows,
+ write_disposition):
+ """Helper for :meth:`Table.upload_from_file`."""
+ load_config = metadata['configuration']['load']
+
+ if allow_jagged_rows is not None:
+ load_config['allowJaggedRows'] = allow_jagged_rows
+
+ if allow_quoted_newlines is not None:
+ load_config['allowQuotedNewlines'] = allow_quoted_newlines
+
+ if create_disposition is not None:
+ load_config['createDisposition'] = create_disposition
+
+ if encoding is not None:
+ load_config['encoding'] = encoding
+
+ if field_delimiter is not None:
+ load_config['fieldDelimiter'] = field_delimiter
+
+ if ignore_unknown_values is not None:
+ load_config['ignoreUnknownValues'] = ignore_unknown_values
+
+ if max_bad_records is not None:
+ load_config['maxBadRecords'] = max_bad_records
+
+ if quote_character is not None:
+ load_config['quote'] = quote_character
+
+ if skip_leading_rows is not None:
+ load_config['skipLeadingRows'] = skip_leading_rows
+
+ if write_disposition is not None:
+ load_config['writeDisposition'] = write_disposition
+
 
 def _parse_schema_resource(info):
  """Parse a resource fragment into a schema field.
@@ -739,3 +964,20 @@ def _build_schema_resource(fields):
  info['fields'] = _build_schema_resource(field.fields)
  infos.append(info)
  return infos
+
+
+class _UploadConfig(object):
+ """Faux message FBO apitools' 'configure_request'."""
+ accept = ['*/*']
+ max_size = None
+ resumable_multipart = True
+ resumable_path = u'/upload/bigquery/v2/projects/{project}/jobs'
+ simple_multipart = True
+ simple_path = u'/upload/bigquery/v2/projects/{project}/jobs'
+
+
+class _UrlBuilder(object):
+ """Faux builder FBO apitools' 'configure_request'"""
+ def __init__(self):
+ self.query_params = {}
+ self._relative_path = ''

gcloud/bigquery/test_client.py

@@ -128,12 +128,12 @@ def test_dataset(self):
  self.assertEqual(dataset.name, DATASET)
  self.assertTrue(dataset._client is client)
 
- def test__job_from_resource_unknown_type(self):
+ def test_job_from_resource_unknown_type(self):
  PROJECT = 'PROJECT'
  creds = _Credentials()
  client = self._makeOne(PROJECT, creds)
  with self.assertRaises(ValueError):
- client._job_from_resource({'configuration': {'nonesuch': {}}})
+ client.job_from_resource({'configuration': {'nonesuch': {}}})
 
  def test_list_jobs_defaults(self):
  from gcloud.bigquery.job import LoadTableFromStorageJob