Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

working python source #533

Merged
merged 34 commits into from
Oct 14, 2020
Merged
Changes from 1 commit
Commits
Show all changes
34 commits
Select commit Hold shift + click to select a range
5f8f787
use log messages for check and add todos
jrhizor Oct 9, 2020
5a1c16e
Merge branch 'master' into jrhizor/py-schema-converter
jrhizor Oct 9, 2020
13eea10
working state
jrhizor Oct 9, 2020
92b3bd7
handle stderr in discoer and use airbyte catalog as output
jrhizor Oct 13, 2020
d732de6
Merge branch 'master' into jrhizor/py-schema-converter
jrhizor Oct 13, 2020
f701fa4
misc restructure
jrhizor Oct 13, 2020
7cfb17c
inject logger again
jrhizor Oct 13, 2020
07e7e4d
use config container (still at the base level)
jrhizor Oct 13, 2020
d394766
add tests and fix logging for check command
jrhizor Oct 13, 2020
cde03b9
move michel's branch into this one
jrhizor Oct 13, 2020
f926838
fmt
jrhizor Oct 13, 2020
4d5c0c9
fix invalid credential test
jrhizor Oct 13, 2020
84c7db9
restructure catalog helpers
jrhizor Oct 13, 2020
09a442c
terrible hack to get stripe to not fail completely
jrhizor Oct 13, 2020
4ad1db2
fix discovery test now that a single message contains the catalog
jrhizor Oct 13, 2020
dda65c3
all working tests except sync
jrhizor Oct 13, 2020
bbe34ad
partially working mapping (in wrong location)
jrhizor Oct 13, 2020
b4a6e92
working but untested reads
jrhizor Oct 13, 2020
5fe3a1e
split logic out
jrhizor Oct 14, 2020
da2d155
fmt
jrhizor Oct 14, 2020
8fc3b1a
fix deadlock bug
jrhizor Oct 14, 2020
9a23ff3
convert catalog in test to airbyte catalog format
jrhizor Oct 14, 2020
f485b29
fix stripe tests
jrhizor Oct 14, 2020
8bf9d50
fmt
jrhizor Oct 14, 2020
76a97e5
Merge branch 'master' into jrhizor/py-schema-converter
jrhizor Oct 14, 2020
9342e08
remoce unused log command
jrhizor Oct 14, 2020
75ddaa4
output credentials
jrhizor Oct 14, 2020
4cfaf8e
clean up optional state command
jrhizor Oct 14, 2020
4aa95e0
use integration launcher
jrhizor Oct 14, 2020
0751013
DRY up read eval
jrhizor Oct 14, 2020
df15670
logging fixes and remove break
jrhizor Oct 14, 2020
d380330
rename
jrhizor Oct 14, 2020
bef954b
fmt
jrhizor Oct 14, 2020
8f673ee
fix inverted if statement
jrhizor Oct 14, 2020
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
use config container (still at the base level)
jrhizor committed Oct 13, 2020

Verified

This commit was signed with the committer’s verified signature.
jrhizor Jared Rhizor
commit 07e7e4d5b5d874206122bab7636a071a98754543
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
from typing import Generator
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why are there two airbyte_protocol directories? It's not clear to me what the difference between the two is?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems like standard structure for python packages:

pkg
pkg/setup.py
pkg/pkg
pkg/pkg/__init__.py
pkg/pkg/*

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SGTM, thanks for bearing with my un-pythonic ways :P

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,33 +38,30 @@ def __init__(self, successful, field_to_error):
self.field_to_error = field_to_error


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, logger, rendered_config_path) -> AirbyteCheckResponse:
def write_config(self, config_object, path):
jrhizor marked this conversation as resolved.
Show resolved Hide resolved
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, logger, rendered_config_path) -> AirbyteCatalog:
def discover(self, logger, config_container) -> AirbyteCatalog:
raise Exception("Not Implemented")


@@ -71,7 +70,7 @@ def __init__(self):
pass

# Iterator<AirbyteMessage>
def read(self, logger, rendered_config_path, catalog_path, state=None) -> Generator[AirbyteMessage, None, None]:
def read(self, logger, config_container, catalog_path, state=None) -> Generator[AirbyteMessage, None, None]:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will come in a future PR?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, this is expected to be implemented by specific sources. There is no default implementation.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah, makes sense

raise Exception("Not Implemented")


@@ -95,3 +94,10 @@ def log_line(line, default_level):
log_record = AirbyteLogMessage(level=log_level, message=rendered_line)
log_message = AirbyteMessage(type="LOG", log=log_record)
print(log_message.serialize())

@dataclass
class ConfigContainer:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe add a comment to explain what these fields are.

raw_config: object
rendered_config: object
raw_config_path: str
rendered_config_path: str
19 changes: 13 additions & 6 deletions airbyte-integrations/base-python/base.py
Original file line number Diff line number Diff line change
@@ -5,6 +5,7 @@
import os.path
import importlib

from airbyte_protocol import ConfigContainer
from airbyte_protocol import Source
from airbyte_protocol import AirbyteLogMessage
from airbyte_protocol import AirbyteMessage
@@ -76,24 +77,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(log_line, rendered_config_path)
check_result = source.check(log_line, config_container)
if check_result.successful:
log("INFO", "Check succeeded")
sys.exit(0)
else:
log("ERROR", "Check failed")
sys.exit(1)
elif cmd == "discover":
catalog = source.discover(log_line, rendered_config_path)
catalog = source.discover(log_line, config_container)
print(catalog.serialize())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is print the standard way of logging in python or is there a "slf4j" equivalent?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since we want to dictate the precise format of what we're outputting, we don't want an slf4j/logger equivalent.

It's debatable if we should use sys.stdout.write() or print, but print just seemed easier (we do want newlines after everthing, and we it's easy to use the string conversions in some cases).

sys.exit(0)
elif cmd == "read":
generator = source.read(log_line, rendered_config_path, parsed_args.catalog, parsed_args.state)
generator = source.read(log_line, config_container, parsed_args.catalog, parsed_args.state)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we have have a similar "container" for catalog and state?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will do this in a separate PR

for message in generator:
print(message.serialize())
sys.exit(0)
Original file line number Diff line number Diff line change
@@ -20,15 +20,15 @@ def __init__(self):
def spec(self) -> AirbyteSpec:
return SingerHelper.spec_from_file("/airbyte/exchangeratesapi-files/spec.json")

def check(self, logger, rendered_config_path) -> AirbyteCheckResponse:
def check(self, logger, config_container) -> AirbyteCheckResponse:
code = urllib.request.urlopen("https://api.exchangeratesapi.io/").getcode()
return AirbyteCheckResponse(code == 200, {})

def discover(self, logger, rendered_config_path) -> AirbyteCatalog:
def discover(self, logger, config_container) -> AirbyteCatalog:
return get_catalogs(logger).airbyte_catalog

def read(self, logger, rendered_config_path, catalog_path, state=None) -> Generator[AirbyteMessage, None, None]:
def read(self, logger, config_container, catalog_path, state=None) -> Generator[AirbyteMessage, None, None]:
if state:
return SingerHelper.read(logger, f"tap-exchangeratesapi --config {rendered_config_path} --state {state}")
return SingerHelper.read(logger, f"tap-exchangeratesapi --config {config_container.rendered_config_path} --state {state}")
jrhizor marked this conversation as resolved.
Show resolved Hide resolved
else:
return SingerHelper.read(logger, f"tap-exchangeratesapi --config {rendered_config_path}")
return SingerHelper.read(logger, f"tap-exchangeratesapi --config {config_container.rendered_config_path}")