ENH: Add table_schema parameter for user-defined BigQuery schema (#46)

Author: mremes

docs/source/changelog.rst

@@ -6,6 +6,7 @@ Changelog
 
 - Resolve issue where the optional ``--noauth_local_webserver`` command line argument would not be propagated during the authentication process. (:issue:`35`)
 - Drop support for Python 3.4 (:issue:`40`)
+- Add support for a passed schema in :func:``to_gbq`` instead inferring the schema from the passed ``DataFrame`` with ``DataFrame.dtypes`` (:issue:`46`)
 
 0.1.6 / 2017-05-03
 ------------------

pandas_gbq/gbq.py

@@ -768,7 +768,8 @@ def read_gbq(query, project_id=None, index_col=None, col_order=None,
 
 
 def to_gbq(dataframe, destination_table, project_id, chunksize=10000,
-           verbose=True, reauth=False, if_exists='fail', private_key=None):
+           verbose=True, reauth=False, if_exists='fail', private_key=None,
+           table_schema=None):
     """Write a DataFrame to a Google BigQuery table.
 
     The main method a user calls to export pandas DataFrame contents to
@@ -815,6 +816,14 @@ def to_gbq(dataframe, destination_table, project_id, chunksize=10000,
         Service account private key in JSON format. Can be file path
         or string contents. This is useful for remote server
         authentication (eg. jupyter iPython notebook on remote host)
+    table_schema : list of dicts
+        List of BigQuery table fields to which according DataFrame columns
+        conform to, e.g. `[{'name': 'col1', 'type': 'STRING'},...]`. If
+        schema is not provided, it will be generated according to dtypes
+        of DataFrame columns. See BigQuery API documentation on available
+        names of a field.
+        
+        .. versionadded:: 0.2.0
     """
 
     if if_exists not in ('fail', 'replace', 'append'):
@@ -831,7 +840,10 @@ def to_gbq(dataframe, destination_table, project_id, chunksize=10000,
     table = _Table(project_id, dataset_id, reauth=reauth,
                    private_key=private_key)
 
-    table_schema = _generate_bq_schema(dataframe)
+    if not table_schema:
+        table_schema = _generate_bq_schema(dataframe)
+    else:
+        table_schema = dict(fields=table_schema)
 
     # If table exists, check if_exists parameter
     if table.exists(table_id):

pandas_gbq/tests/test_gbq.py

@@ -1258,6 +1258,48 @@ def test_verify_schema_ignores_field_mode(self):
         assert self.sut.verify_schema(
             self.dataset_prefix + "1", TABLE_ID + test_id, test_schema_2)
 
+    def test_upload_data_with_valid_user_schema(self):
+        # Issue #46; tests test scenarios with user-provided
+        # schemas
+        df = tm.makeMixedDataFrame()
+        test_id = "15"
+        test_schema = [{'name': 'A', 'type': 'FLOAT'},
+                       {'name': 'B', 'type': 'FLOAT'},
+                       {'name': 'C', 'type': 'STRING'},
+                       {'name': 'D', 'type': 'TIMESTAMP'}]
+        destination_table = self.destination_table + test_id
+        gbq.to_gbq(df, destination_table, _get_project_id(),
+                   private_key=_get_private_key_path(),
+                   table_schema=test_schema)
+        dataset, table = destination_table.split('.')
+        assert self.table.verify_schema(dataset, table,
+                                        dict(fields=test_schema))
+
+    def test_upload_data_with_invalid_user_schema_raises_error(self):
+        df = tm.makeMixedDataFrame()
+        test_id = "16"
+        test_schema = [{'name': 'A', 'type': 'FLOAT'},
+                       {'name': 'B', 'type': 'FLOAT'},
+                       {'name': 'C', 'type': 'FLOAT'},
+                       {'name': 'D', 'type': 'FLOAT'}]
+        destination_table = self.destination_table + test_id
+        with tm.assertRaises(gbq.StreamingInsertError):
+            gbq.to_gbq(df, destination_table, _get_project_id(),
+                       private_key=_get_private_key_path(),
+                       table_schema=test_schema)
+
+    def test_upload_data_with_missing_schema_fields_raises_error(self):
+        df = tm.makeMixedDataFrame()
+        test_id = "16"
+        test_schema = [{'name': 'A', 'type': 'FLOAT'},
+                       {'name': 'B', 'type': 'FLOAT'},
+                       {'name': 'C', 'type': 'FLOAT'}]
+        destination_table = self.destination_table + test_id
+        with tm.assertRaises(gbq.StreamingInsertError):
+            gbq.to_gbq(df, destination_table, _get_project_id(),
+                       private_key=_get_private_key_path(),
+                       table_schema=test_schema)
+
     def test_list_dataset(self):
         dataset_id = self.dataset_prefix + "1"
         assert dataset_id in self.dataset.datasets()