Add optional --discover capability

.gitignore

@@ -1,4 +1,9 @@
+# CSV output files
+
+exchange_rate-*.csv
+
 # Byte-compiled / optimized / DLL files
+.mypy_cache
 __pycache__/
 *.py[cod]
 *$py.class

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "[ptyhon]": {
+        // Prevent auto-format until `black` auto-formatter is adopted:
+        "editor.formatOnSave": false
+    },
+    "editor.formatOnSave": false,
+}

tap_exchangeratesapi/__init__.py

@@ -5,18 +5,23 @@
 import argparse
 import time
 import requests
-import singer
 import backoff
 import copy
 
-from datetime import date, datetime, timedelta
+from datetime import datetime, timedelta
+from typing import List, Dict, Any
+
+import singer
+from .discovery import discover
+
 
 base_url = 'https://api.exchangeratesapi.io/'
 
 logger = singer.get_logger()
 session = requests.Session()
 
-DATE_FORMAT='%Y-%m-%d'
+DATE_FORMAT: str = '%Y-%m-%d'
+REQUIRED_CONFIG_KEYS: List[str] = []  # All config keys are optional
 
 def parse_response(r):
     flattened = r['rates']
@@ -45,12 +50,12 @@ def request(url, params):
     response = requests.get(url=url, params=params)
     response.raise_for_status()
     return response
-    
-def do_sync(base, start_date):
+
+def do_sync(base, start_date, catalog_override: Dict[str, Any] = None):
     state = {'start_date': start_date}
     next_date = start_date
-    prev_schema = {}
-    
+    prev_schema: Dict[str, Any] = {}
+
     try:
         while datetime.strptime(next_date, DATE_FORMAT) <= datetime.utcnow():
             logger.info('Replicating exchange rate data from %s using base %s',
@@ -70,7 +75,38 @@ def do_sync(base, start_date):
                 singer.write_schema('exchange_rate', schema, 'date')
 
             if payload['date'] == next_date:
-                singer.write_records('exchange_rate', [parse_response(payload)])
+                if catalog_override:
+                    catalog_stream_override = [
+                        x for x in catalog_override["streams"]
+                        if x["tap_stream_id"] == "exchange_rate"
+                    ]
+                    if not catalog_stream_override:
+                        raise ValueError(
+                            "Stream 'exchange_rate' not found in "
+                            f"json: {catalog_override}"
+                        )
+                    # else:
+                    #     logger.info(catalog_stream_override)
+                    metadata_override = catalog_stream_override[0]["metadata"]
+                    if not metadata_override:
+                        raise ValueError(
+                            "Metadata not found in "
+                            f"json: {catalog_override}"
+                        )
+                    # else:
+                    #     logger.info(metadata_override)
+                    logger.info("Replicating with provided catalog file override")
+                    for record in [parse_response(payload)]:
+                        singer.write_record(
+                            'exchange_rate',
+                            singer.Transformer().transform(
+                                data=record,
+                                schema=catalog_stream_override[0]["schema"],
+                                # metadata=metadata_override, # TODO: debug (not working)
+                            )
+                        )
+                else:
+                    singer.write_records('exchange_rate', [parse_response(payload)])
 
             state = {'start_date': next_date}
             next_date = (datetime.strptime(next_date, DATE_FORMAT) + timedelta(days=1)).strftime(DATE_FORMAT)
@@ -90,13 +126,29 @@ def do_sync(base, start_date):
 def main():
     parser = argparse.ArgumentParser()
 
+    parser.add_argument(
+        '-d', '--discover', help='Do schema discovery', required=False, action='store_true')
     parser.add_argument(
         '-c', '--config', help='Config file', required=False)
     parser.add_argument(
         '-s', '--state', help='State file', required=False)
+    parser.add_argument(
+        '--catalog', help='Catalog file', required=False)
 
     args = parser.parse_args()
 
+    # If discover flag was passed, run discovery mode and dump output to stdout
+    if args.discover:
+        catalog = discover()
+        singer.catalog.write_catalog(catalog)
+        return
+
+    catalog_override: Dict = None
+    if args.catalog:
+        with open(args.catalog) as file:
+            catalog_override = json.load(file)
+
+    # Otherwise run in sync mode
     if args.config:
         with open(args.config) as file:
             config = json.load(file)
@@ -112,7 +164,7 @@ def main():
     start_date = state.get('start_date') or config.get('start_date') or datetime.utcnow().strftime(DATE_FORMAT)
     start_date = singer.utils.strptime_with_tz(start_date).date().strftime(DATE_FORMAT)
 
-    do_sync(config.get('base', 'USD'), start_date)
+    do_sync(config.get('base', 'USD'), start_date, catalog_override=catalog_override)
 
 
 if __name__ == '__main__':

tap_exchangeratesapi/discovery.py

@@ -0,0 +1,55 @@
+"""Discovery functions for the Singer.io tap"""
+
+import json
+import os
+
+import singer
+from singer import Catalog, metadata
+
+from typing import Dict, List, Any
+
+LOGGER = singer.get_logger()
+REPLICATION_METHOD = "INCREMENTAL"
+
+
+def _get_abs_path(path):
+    return os.path.join(os.path.dirname(os.path.realpath(__file__)), path)
+
+
+# Load schemas from schemas folder
+def _load_schemas():
+    schemas = {}
+    for filename in os.listdir(_get_abs_path("schemas")):
+        path = _get_abs_path("schemas") + "/" + filename
+        basename = filename.replace(".json", "")
+        with open(path) as file:
+            schemas[basename] = json.load(file)
+    return schemas
+
+
+def discover() -> Dict[str, Dict[str, Any]]:
+    """
+    Run discovery
+
+    Returns
+    -------
+    Dict[str, Dict[str, Any]]
+        The list of streams with their associated metadata
+    """
+    LOGGER.info("Starting discovery mode")
+    streams = []
+    for stream_name, schema in _load_schemas().items():
+        catalog_entry = {
+            "stream": stream_name,
+            "tap_stream_id": stream_name,
+            "schema": schema,
+            "metadata": metadata.get_standard_metadata(
+                schema=schema["schema"],
+                key_properties=["date"],
+                valid_replication_keys=["date"],
+                replication_method=REPLICATION_METHOD,
+            ),
+            "key_properties": ["date"],
+        }
+        streams.append(catalog_entry)
+    return Catalog.from_dict({"streams": streams})

tap_exchangeratesapi/schemas/exchange_rate.json

@@ -0,0 +1,56 @@
+{
+  "stream": "exchange_rate",
+  "tap_stream_id": "exchange_rate",
+  "schema": {
+    "type": "object",
+    "properties": {
+      "date": { "type": "string", "format": "date-time" },
+      "CAD": { "type": ["null", "number"] },
+      "HKD": { "type": ["null", "number"] },
+      "ISK": { "type": ["null", "number"] },
+      "PHP": { "type": ["null", "number"] },
+      "DKK": { "type": ["null", "number"] },
+      "HUF": { "type": ["null", "number"] },
+      "CZK": { "type": ["null", "number"] },
+      "GBP": { "type": ["null", "number"] },
+      "RON": { "type": ["null", "number"] },
+      "SEK": { "type": ["null", "number"] },
+      "IDR": { "type": ["null", "number"] },
+      "INR": { "type": ["null", "number"] },
+      "BRL": { "type": ["null", "number"] },
+      "RUB": { "type": ["null", "number"] },
+      "HRK": { "type": ["null", "number"] },
+      "JPY": { "type": ["null", "number"] },
+      "THB": { "type": ["null", "number"] },
+      "CHF": { "type": ["null", "number"] },
+      "EUR": { "type": ["null", "number"] },
+      "MYR": { "type": ["null", "number"] },
+      "BGN": { "type": ["null", "number"] },
+      "TRY": { "type": ["null", "number"] },
+      "CNY": { "type": ["null", "number"] },
+      "NOK": { "type": ["null", "number"] },
+      "NZD": { "type": ["null", "number"] },
+      "ZAR": { "type": ["null", "number"] },
+      "USD": { "type": ["null", "number"] },
+      "MXN": { "type": ["null", "number"] },
+      "SGD": { "type": ["null", "number"] },
+      "AUD": { "type": ["null", "number"] },
+      "ILS": { "type": ["null", "number"] },
+      "KRW": { "type": ["null", "number"] },
+      "PLN": { "type": ["null", "number"] }
+    }
+  },
+  "metadata": [
+    {
+      "metadata": {
+        "table-key-properties": ["date"],
+        "valid-replication-keys": ["date"],
+        "is-view": false,
+        "selected": true,
+        "replication-method": "INCREMENTAL"
+      },
+      "breadcrumb": []
+    }
+  ],
+  "key_properties": ["date"]
+}