anelendata · Jan 10, 2024 · Jan 10, 2024 · Jan 12, 2024 · Jan 12, 2024 · Jan 29, 2024
Showing with 23 additions and 4 deletions.

+2 −1 README.md

+1 −0 setup.py

+20 −3 tap_bigquery/sync_bigquery.py
diff --git a/README.md b/README.md
@@ -80,7 +80,8 @@ config.sample.json:
   "start_datetime": "2017-01-01T00:00:00Z", // This can be set at the command line argument
   "end_datetime": "2017-02-01T00:00:00Z", // end_datetime is optional
   "limit": 100,
-  "start_always_inclusive": false // default is false, optional
+  "start_always_inclusive": false, // default is false, optional
+  "schema_path": "/absolute/path/to/schemas/directory" // default None, will attempt to infer schema
 }
 ```
 

diff --git a/setup.py b/setup.py
@@ -25,6 +25,7 @@
         "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.10",
     ],
 
     install_requires=[

diff --git a/tap_bigquery/sync_bigquery.py b/tap_bigquery/sync_bigquery.py
@@ -1,4 +1,6 @@
 import copy, datetime, json, time
+import os.path
+
 import dateutil.parser
 from decimal import Decimal
 
@@ -76,8 +78,7 @@ def _build_query(keys, filters=[], inclusive_start=True, limit=None):
     return query
 
 
-def do_discover(config, stream, output_schema_file=None,
-                add_timestamp=True):
+def infer_schema(config, stream):
     client = get_bigquery_client()
 
     start_datetime = dateutil.parser.parse(
@@ -113,7 +114,23 @@ def do_discover(config, stream, output_schema_file=None,
     if not data:
         raise Exception("Cannot infer schema: No record returned.")
 
-    schema = getschema.infer_schema(data)
+    return getschema.infer_schema(data)
+
+
+def do_discover(config, stream, output_schema_file=None,
+                add_timestamp=True):
+    schema_file_path = os.path.join(config.get("schema_path", ""), f"{stream['table']}.json")
+    if not config.get("schema_path"):
+        LOGGER.info("No schema folder specified, attempting to infer schema")
+        schema = infer_schema(config, stream)
+    elif not os.path.exists(schema_file_path):
+        LOGGER.info(f"No schema file found at {schema_file_path}, attempting to infer schema")
+        schema = infer_schema(config, stream)
+    else:
+        with open(schema_file_path, "r") as fd:
+            schema = json.loads(fd.read())
+        LOGGER.info(f"Using static schema defined at {schema_file_path}")
+
     if add_timestamp:
         timestamp_format = {"type": ["null", "string"],
                             "format": "date-time"}