working python source

airbyte-integrations/base-python/airbyte_protocol/airbyte_protocol/__init__.py

@@ -1,8 +1,10 @@
 from typing import Generator
 import yaml
+import json
 import pkgutil
 import warnings
 import python_jsonschema_objects as pjs
+from dataclasses import dataclass
 
 
 def _load_classes(yaml_path: str):
@@ -36,38 +38,30 @@ def __init__(self, successful, field_to_error):
         self.field_to_error = field_to_error
 
 
-class AirbyteSchema(object):
-    def __init__(self, schema):
-        self.schema = schema
-
-
-class AirbyteConfig(object):
-    def __init__(self, config_string):
-        self.config_string = config_string
-
-
 class Integration(object):
     def __init__(self):
         pass
 
     def spec(self) -> AirbyteSpec:
         raise Exception("Not Implemented")
 
-    # default version reads the config_path to a string
-    # this will often be overwritten to add fields for easy consumption or to modify the string for delegating to singer
-    def read_config(self, config_path) -> AirbyteConfig:
+    def read_config(self, config_path):
         with open(config_path, 'r') as file:
             contents = file.read()
-        return AirbyteConfig(contents)
+        return json.loads(contents)
 
-    def render_config(self, config_object, rendered_config_path):
-        with open(rendered_config_path, 'w') as fh:
-            fh.write(config_object.config_string)
+    # can be overridden to change an input file config
+    def transform_config(self, raw_config):
+        return raw_config
 
-    def check(self, config_object, rendered_config_path) -> AirbyteCheckResponse:
+    def write_config(self, config_object, path):
+        with open(path, 'w') as fh:
+            fh.write(json.dumps(config_object))
+
+    def check(self, logger, config_container) -> AirbyteCheckResponse:
         raise Exception("Not Implemented")
 
-    def discover(self, config_object, rendered_config_path) -> AirbyteSchema:
+    def discover(self, logger, config_container) -> AirbyteCatalog:
         raise Exception("Not Implemented")
 
 
@@ -76,10 +70,56 @@ def __init__(self):
         pass
 
     # Iterator<AirbyteMessage>
-    def read(self, config_object, rendered_config_path, state=None) -> Generator[AirbyteMessage, None, None]:
+    def read(self, logger, config_container, catalog_path, state=None) -> Generator[AirbyteMessage, None, None]:
         raise Exception("Not Implemented")
 
 
 class Destination(Integration):
     def __init__(self):
         pass
+
+
+class AirbyteLogger:
+    def __init__(self):
+        self.valid_log_types = ["FATAL", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"]
+
+    def log_by_prefix(self, message, default_level):
+        split_line = message.split()
+        first_word = next(iter(split_line), None)
+        if first_word in self.valid_log_types:
+            log_level = first_word
+            rendered_message = " ".join(split_line[1:])
+        else:
+            log_level = default_level
+            rendered_message = message
+        self.log(log_level, rendered_message)
+
+    def log(self, level, message):
+        log_record = AirbyteLogMessage(level=level, message=message)
+        log_message = AirbyteMessage(type="LOG", log=log_record)
+        print(log_message.serialize())
+
+    def fatal(self, message):
+        self.log("FATAL", message)
+
+    def error(self, message):
+        self.log("ERROR", message)
+
+    def warn(self, message):
+        self.log("WARN", message)
+
+    def info(self, message):
+        self.log("INFO", message)
+
+    def debug(self, message):
+        self.log("DEBUG", message)
+
+    def trace(self, message):
+        self.log("TRACE", message)
+
+@dataclass
+class ConfigContainer:
+    raw_config: object
+    rendered_config: object
+    raw_config_path: str
+    rendered_config_path: str

airbyte-integrations/base-python/base.py

@@ -1,18 +1,23 @@
 import argparse
-import logging
 import sys
 import tempfile
 import os.path
 import importlib
 
+from airbyte_protocol import ConfigContainer
 from airbyte_protocol import Source
+from airbyte_protocol import AirbyteLogger
+from airbyte_protocol import AirbyteLogMessage
+from airbyte_protocol import AirbyteMessage
 
 impl_module = os.environ['AIRBYTE_IMPL_MODULE']
 impl_class = os.environ['AIRBYTE_IMPL_PATH']
 
 module = importlib.import_module(impl_module)
 impl = getattr(module, impl_class)
 
+logger = AirbyteLogger()
+
 class AirbyteEntrypoint(object):
     def __init__(self, source):
         self.source = source
@@ -43,8 +48,8 @@ def start(self):
         # read
         read_parser = subparsers.add_parser("read", help="reads the source and outputs messages to STDOUT",
                                             parents=[parent_parser])
-        # todo: re-add state handling
-        # read_parser.add_argument('--state', type=str, required=False, help='path to the json-encoded state file')
+
+        read_parser.add_argument('--state', type=str, required=False, help='path to the json-encoded state file')
         required_read_parser = read_parser.add_argument_group('required named arguments')
         required_read_parser.add_argument('--config', type=str, required=True,
                                           help='path to the json configuration file')
@@ -66,25 +71,30 @@ def start(self):
                 sys.exit(0)
 
             rendered_config_path = os.path.join(temp_dir, 'config.json')
-            config_object = source.read_config(parsed_args.config)
-            source.render_config(config_object, rendered_config_path)
+            raw_config = source.read_config(parsed_args.config)
+            rendered_config = source.transform_config(raw_config)
+            source.write_config(rendered_config, rendered_config_path)
+
+            config_container = ConfigContainer(
+                raw_config=raw_config,
+                rendered_config=rendered_config,
+                raw_config_path=parsed_args.config,
+                rendered_config_path=rendered_config_path)
 
-            # todo: output message for check
             if cmd == "check":
-                check_result = source.check(logging, rendered_config_path)
+                check_result = source.check(logger, config_container)
                 if check_result.successful:
-                    print("Check succeeded")
+                    logger.info("Check succeeded")
                     sys.exit(0)
                 else:
-                    print("Check failed")
+                    logger.error("Check failed")
                     sys.exit(1)
             elif cmd == "discover":
-                schema = source.discover(logging, rendered_config_path)
-                print(schema.schema)
+                catalog = source.discover(logger, config_container)
+                print(catalog.serialize())
                 sys.exit(0)
             elif cmd == "read":
-                # todo: pass in state
-                generator = source.read(logging, rendered_config_path)
+                generator = source.read(logger, config_container, parsed_args.catalog, parsed_args.state)
                 for message in generator:
                     print(message.serialize())
                 sys.exit(0)

airbyte-integrations/base/base.sh

@@ -48,8 +48,9 @@ function main() {
     eval "$AIRBYTE_DISCOVER_CMD" --config "$CONFIG_FILE"
     ;;
   read)
-    # todo: state should be optional: --state "$STATE_FILE"
-    eval "$AIRBYTE_READ_CMD" --config "$CONFIG_FILE" --catalog "$CATALOG_FILE"
+    READ_STATEMENT="$AIRBYTE_READ_CMD --config $CONFIG_FILE --catalog $CATALOG_FILE"
+    if [[ ! -z "$STATE_FILE" ]]; then READ_STATEMENT="$READ_STATEMENT --state $STATE_FILE"; fi
+    eval "$READ_STATEMENT"
     ;;
   write)
     eval "$AIRBYTE_WRITE_CMD" --config "$CONFIG_FILE" --catalog "$CATALOG_FILE"