working python source #533

jrhizor · 2020-10-09T18:06:37Z

adds state handling
adds schema handling, including marking singer catalogs as selected based on the incoming airbyte catalog
logging
interface changes

recommended reading order:

base / helper files
source_stripe_py
rest

jrhizor · 2020-10-09T18:08:48Z

Working state example:

config.json:

{
  "base": "AUD",
  "start_date": "2020-10-02"
}

state.json:

{"start_date": "2020-10-09"}

→ docker run -v $(pwd):/mount --rm airbyte/source-exchangeratesapi-singer:dev read --config /mount/config.json --catalog /mount/catalog.json
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-02 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9530295627, "HKD": 5.5527730271, "ISK": 99.071585634, "PHP": 34.7324700709, "DKK": 4.5451991204, "HUF": 219.2035182018, "CZK": 16.5037869533, "GBP": 0.5538297093, "RON": 2.9755069631, "SEK": 6.3675787931, "IDR": 10655.5888101637, "INR": 52.5516125092, "BRL": 4.0336550208, "RUB": 56.2438920108, "HRK": 4.6188614708, "JPY": 75.3725873442, "THB": 22.5940630344, "CHF": 0.6590520401, "EUR": 0.610798925, "MYR": 2.9837527486, "BGN": 1.1946005375, "TRY": 5.5452602003, "CNY": 4.8653799169, "NOK": 6.6601514781, "NZD": 1.07952602, "ZAR": 11.8496213047, "USD": 0.716467139, "MXN": 15.6720009773, "SGD": 0.9765453213, "AUD": 1.0, "ILS": 2.4580991937, "KRW": 833.5450769607, "PLN": 2.7451746885, "date": "2020-10-02T00:00:00Z"}, "emitted_at": "2020-10-09 18:05:05.462186"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-03 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9532938519, "HKD": 5.5682886623, "ISK": 99.1513523414, "PHP": 34.7768484034, "DKK": 4.542401856, "HUF": 218.6763538678, "CZK": 16.5431345015, "GBP": 0.5544294523, "RON": 2.975761646, "SEK": 6.3889736858, "IDR": 10577.3307283717, "INR": 52.5593748092, "BRL": 4.0642285854, "RUB": 56.4963062458, "HRK": 4.6205507052, "JPY": 75.8593320716, "THB": 22.4958788693, "CHF": 0.6582208926, "EUR": 0.6105378839, "MYR": 2.9835154771, "BGN": 1.1940899933, "TRY": 5.5933817693, "CNY": 4.8921179559, "NOK": 6.6460101349, "NZD": 1.0808352158, "ZAR": 11.8106722022, "USD": 0.7184809817, "MXN": 15.4191953111, "SGD": 0.9772879907, "AUD": 1.0, "ILS": 2.4569876061, "KRW": 832.602722999, "PLN": 2.7461994017, "date": "2020-10-05T00:00:00Z"}, "emitted_at": "2020-10-09 18:05:05.765276"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-04 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9488829529, "HKD": 5.549538611, "ISK": 98.8343856241, "PHP": 34.6873482273, "DKK": 4.5172413793, "HUF": 218.3827100534, "CZK": 16.41270034, "GBP": 0.5528047596, "RON": 2.9601748422, "SEK": 6.3828314716, "IDR": 10551.195968917, "INR": 52.5567629917, "BRL": 3.9681277319, "RUB": 55.7139388052, "HRK": 4.5949490044, "JPY": 75.6313744536, "THB": 22.3379067508, "CHF": 0.6545046139, "EUR": 0.6070908208, "MYR": 2.9741986401, "BGN": 1.1873482273, "TRY": 5.5624696455, "CNY": 4.8634652744, "NOK": 6.593188441, "NZD": 1.0765541525, "ZAR": 11.8097377368, "USD": 0.7160636231, "MXN": 15.2762870325, "SGD": 0.9732880039, "AUD": 1.0, "ILS": 2.440080136, "KRW": 830.7977173385, "PLN": 2.724137931, "date": "2020-10-06T00:00:00Z"}, "emitted_at": "2020-10-09 18:05:05.874866"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-05 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9478787879, "HKD": 5.5284242424, "ISK": 98.6666666667, "PHP": 34.5327272727, "DKK": 4.5103636364, "HUF": 217.7272727273, "CZK": 16.4024242424, "GBP": 0.5540181818, "RON": 2.9544242424, "SEK": 6.3533333333, "IDR": 10520.9151515152, "INR": 52.2806060606, "BRL": 3.9665454545, "RUB": 55.7403030303, "HRK": 4.586969697, "JPY": 75.5939393939, "THB": 22.2775757576, "CHF": 0.6537575758, "EUR": 0.6060606061, "MYR": 2.9646666667, "BGN": 1.1853333333, "TRY": 5.6195151515, "CNY": 4.8444848485, "NOK": 6.625030303, "NZD": 1.0825454545, "ZAR": 11.8662424242, "USD": 0.7133333333, "MXN": 15.3096363636, "SGD": 0.9693939394, "AUD": 1.0, "ILS": 2.4269090909, "KRW": 825.8787878788, "PLN": 2.7189090909, "date": "2020-10-07T00:00:00Z"}, "emitted_at": "2020-10-09 18:05:05.990058"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-06 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9489677852, "HKD": 5.5527068997, "ISK": 99.1413434017, "PHP": 34.7140856221, "DKK": 4.532184398, "HUF": 217.6968515925, "CZK": 16.4983862128, "GBP": 0.5543815846, "RON": 2.9692466963, "SEK": 6.3618537239, "IDR": 10573.1685037452, "INR": 52.4852323245, "BRL": 4.0090128494, "RUB": 55.4579501857, "HRK": 4.6087327203, "JPY": 75.9332561963, "THB": 22.3640460386, "CHF": 0.6576335181, "EUR": 0.6089763108, "MYR": 2.9750928689, "BGN": 1.1910358687, "TRY": 5.6819925705, "CNY": 4.8653553377, "NOK": 6.6401558979, "NZD": 1.0873881006, "ZAR": 11.8747335729, "USD": 0.7164606297, "MXN": 15.3107606114, "SGD": 0.9730832471, "AUD": 1.0, "ILS": 2.4303635589, "KRW": 825.4552097923, "PLN": 2.7308324706, "date": "2020-10-08T00:00:00Z"}, "emitted_at": "2020-10-09 18:05:06.098758"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-07 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9473940872, "HKD": 5.5722035965, "ISK": 99.2380371838, "PHP": 34.7960987504, "DKK": 4.5365437367, "HUF": 217.1776897287, "CZK": 16.5254495581, "GBP": 0.5557269125, "RON": 2.9695214874, "SEK": 6.351722036, "IDR": 10569.954282231, "INR": 52.5440414508, "BRL": 4.0107284365, "RUB": 55.4351112466, "HRK": 4.6184090216, "JPY": 76.1658031088, "THB": 22.3279487961, "CHF": 0.6566900335, "EUR": 0.609570253, "MYR": 2.9751904907, "BGN": 1.1921975008, "TRY": 5.6860103627, "CNY": 4.8184699787, "NOK": 6.6213349589, "NZD": 1.0863151478, "ZAR": 11.8448643706, "USD": 0.7189881134, "MXN": 15.2900335264, "SGD": 0.9744590064, "AUD": 1.0, "ILS": 2.4296860713, "KRW": 823.2368180433, "PLN": 2.7249009448, "date": "2020-10-09T00:00:00Z"}, "emitted_at": "2020-10-09 18:05:06.199196"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-08 using base AUD"}}
{"type": "STATE", "state": {"data": {"start_date": "2020-10-09"}}}
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-09 using base AUD"}}

→ docker run -v $(pwd):/mount --rm airbyte/source-exchangeratesapi-singer:dev read --config /mount/config.json --catalog /mount/catalog.json --state /mount/state.json
{"type": "LOG", "log": {"level": "INFO", "message": "Replicating exchange rate data from 2020-10-09 using base AUD"}}
{"type": "RECORD", "record": {"stream": "exchange_rate", "data": {"CAD": 0.9473940872, "HKD": 5.5722035965, "ISK": 99.2380371838, "PHP": 34.7960987504, "DKK": 4.5365437367, "HUF": 217.1776897287, "CZK": 16.5254495581, "GBP": 0.5557269125, "RON": 2.9695214874, "SEK": 6.351722036, "IDR": 10569.954282231, "INR": 52.5440414508, "BRL": 4.0107284365, "RUB": 55.4351112466, "HRK": 4.6184090216, "JPY": 76.1658031088, "THB": 22.3279487961, "CHF": 0.6566900335, "EUR": 0.609570253, "MYR": 2.9751904907, "BGN": 1.1921975008, "TRY": 5.6860103627, "CNY": 4.8184699787, "NOK": 6.6213349589, "NZD": 1.0863151478, "ZAR": 11.8448643706, "USD": 0.7189881134, "MXN": 15.2900335264, "SGD": 0.9744590064, "AUD": 1.0, "ILS": 2.4296860713, "KRW": 823.2368180433, "PLN": 2.7249009448, "date": "2020-10-09T00:00:00Z"}, "emitted_at": "2020-10-09 18:04:57.870061"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Tap exiting normally"}}

jrhizor · 2020-10-14T15:21:01Z

making this available for review even though stripe tests need updating.

michel-tricot · 2020-10-14T15:59:14Z

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

+    print(log_message.serialize())
+
+@dataclass
+class ConfigContainer:


Maybe add a comment to explain what these fields are.

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

michel-tricot · 2020-10-14T16:14:01Z

airbyte-integrations/base-python/base.py

                sys.exit(0)
            elif cmd == "read":
-                # todo: pass in state
-                generator = source.read(logging, rendered_config_path)
+                generator = source.read(log_line, config_container, parsed_args.catalog, parsed_args.state)


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

will do this in a separate PR

airbyte-integrations/base/base.sh

michel-tricot · 2020-10-14T16:42:33Z

airbyte-integrations/singer/base-singer/base_singer/singer_helpers.py

+            logger(line, "ERROR")
+
+        airbyte_streams = []
+        singer_catalog = singer_transform(json.loads(completed_process.stdout))


Should we only try to load the line if the line starts with { so we don't have the risk of reading some debug logs?

I think most singer taps write all logs to stderr for the discover command and the entire stdout is expected to be readable as the catalog. I think it's probably fine to expect that behavior for now and adapt if we run into a case that violates that expectation. I imagine for that case we should be doing custom work at that source level to handle it.

...ger/exchangeratesapi/source/source_exchangeratesapi_singer/source_exchangeratesapi_singer.py

...-integration/java/io/airbyte/integration_tests/sources/SingerExchangeRatesApiSourceTest.java

...te-integrations/singer/stripe_abprotocol/source/source_stripe_singer/source_stripe_singer.py

remove config

sherifnada

Did a first pass on the python files, will do a second pass later. The usage is looking clean so far! Very exciting

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

sherifnada · 2020-10-14T17:23:07Z

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

@@ -76,10 +70,34 @@ 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]:


This will come in a future PR?

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

ah, makes sense

sherifnada · 2020-10-14T17:36:19Z

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

@@ -1,8 +1,10 @@
 from typing import Generator


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

This seems like standard structure for python packages:

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

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

sherifnada · 2020-10-14T17:40:31Z

airbyte-integrations/base-python/base.py

-                schema = source.discover(logging, rendered_config_path)
-                print(schema.schema)
+                catalog = source.discover(log_line, config_container)
+                print(catalog.serialize())


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

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).

sherifnada · 2020-10-14T17:45:17Z

airbyte-integrations/singer/base-singer/base_singer/singer_helpers.py

+            schema = stream.get("schema").get("properties")
+
+            # todo: figure out how to serialize an object with an items key in python_jsonschema_objects
+            if name == "subscriptions":


does this mean we are blotting out parts of the catalog?

Updated #530 to show what it needs to do to fix this. I've only seen this on Stripe so far but this is definitely something we need to fix before release. We can do it outside of this PR.

sherifnada · 2020-10-14T17:45:19Z

airbyte-integrations/singer/base-singer/base_singer/singer_helpers.py

    @staticmethod
-    def discover(shell_command, transform=(lambda x: AirbyteSchema(x))) -> AirbyteSchema:
+    def get_catalogs(logger, shell_command, singer_transform=(lambda catalog: catalog), airbyte_transform=(lambda catalog: catalog)) -> Catalogs:


what is singer_transform and airbyte_transform used for?

sherifnada · 2020-10-14T17:53:20Z

airbyte-integrations/singer/base-singer/base_singer/singer_helpers.py

+
+            ok = True
+            while ok:
+                for key, val1 in sel.select():


does select break if the underlying stream has closed? From the select docs: If timeout is None, the call will block until a monitored file object becomes ready.

It doesn't break when the underlying process ends. The final iteration of this loop is when the file object p.stdout is already closed, and it exits gracefully by reaching the ok = False condition.

airbyte-integrations/singer/base-singer/base_singer/singer_helpers.py

sherifnada · 2020-10-14T17:57:56Z

airbyte-integrations/singer/base-singer/base_singer/singer_helpers.py

-
-                if err_line:
-                    log_line(err_line, "ERROR")
+    def combine_catalogs(masked_airbyte_catalog, discovered_singer_catalog) -> str:


is the output here expected to be an object or a Singer or Airbytecatalog? if so can the type signature reflect that? Also, this is more of making selections on the discovered catalog rather than "combining" the two catalogs right? Can we rename the method to reflect that?

also, can you add a unit test for this method?

I did the rename here. I'll do the documentation piece and type handling separately.

jrhizor · 2020-10-14T18:12:57Z

A few of these comments can be summarized as:

add type hints
add comments/docs
add unit tests

I'll address these in a separate PR after this is merged.

jrhizor added 3 commits October 9, 2020 09:51

use log messages for check and add todos

5f8f787

Merge branch 'master' into jrhizor/py-schema-converter

5a1c16e

working state

13eea10

jrhizor added 16 commits October 13, 2020 06:28

handle stderr in discoer and use airbyte catalog as output

92b3bd7

Merge branch 'master' into jrhizor/py-schema-converter

d732de6

misc restructure

f701fa4

inject logger again

7cfb17c

use config container (still at the base level)

07e7e4d

add tests and fix logging for check command

d394766

move michel's branch into this one

cde03b9

fmt

f926838

fix invalid credential test

4d5c0c9

restructure catalog helpers

84c7db9

terrible hack to get stripe to not fail completely

09a442c

fix discovery test now that a single message contains the catalog

4ad1db2

all working tests except sync

dda65c3

partially working mapping (in wrong location)

bbe34ad

working but untested reads

b4a6e92

split logic out

5fe3a1e

jrhizor requested review from sherifnada, cgardens and michel-tricot and removed request for sherifnada October 14, 2020 15:20

jrhizor marked this pull request as ready for review October 14, 2020 15:20

fmt

da2d155

jrhizor changed the title ~~state + schema handling in python source~~ working python source Oct 14, 2020

michel-tricot approved these changes Oct 14, 2020

View reviewed changes

jrhizor added 6 commits October 14, 2020 09:56

fix deadlock bug

8fc3b1a

convert catalog in test to airbyte catalog format

9a23ff3

fix stripe tests

f485b29

fmt

8bf9d50

Merge branch 'master' into jrhizor/py-schema-converter

76a97e5

remoce unused log command

9342e08

remove config

jrhizor force-pushed the jrhizor/py-schema-converter branch from 90110fb to 9342e08 Compare October 14, 2020 17:21

jrhizor added 4 commits October 14, 2020 10:24

output credentials

75ddaa4

clean up optional state command

4cfaf8e

use integration launcher

4aa95e0

DRY up read eval

0751013

sherifnada reviewed Oct 14, 2020

View reviewed changes

logging fixes and remove break

df15670

jrhizor added 3 commits October 14, 2020 11:32

rename

d380330

fmt

bef954b

fix inverted if statement

8f673ee

jrhizor merged commit 8d165a7 into master Oct 14, 2020

jrhizor deleted the jrhizor/py-schema-converter branch October 14, 2020 19:14

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

working python source #533

working python source #533

jrhizor commented Oct 9, 2020 •

edited

Loading

jrhizor commented Oct 9, 2020 •

edited

Loading

jrhizor commented Oct 14, 2020

michel-tricot Oct 14, 2020

michel-tricot Oct 14, 2020

jrhizor Oct 14, 2020

michel-tricot Oct 14, 2020

jrhizor Oct 14, 2020

sherifnada left a comment •

edited

Loading

sherifnada Oct 14, 2020

jrhizor Oct 14, 2020

sherifnada Oct 14, 2020

sherifnada Oct 14, 2020

jrhizor Oct 14, 2020

sherifnada Oct 14, 2020

sherifnada Oct 14, 2020

jrhizor Oct 14, 2020

sherifnada Oct 14, 2020

jrhizor Oct 14, 2020

sherifnada Oct 14, 2020

sherifnada Oct 14, 2020

jrhizor Oct 14, 2020

sherifnada Oct 14, 2020

sherifnada Oct 14, 2020

jrhizor Oct 14, 2020

jrhizor commented Oct 14, 2020

working python source #533

working python source #533

Conversation

jrhizor commented Oct 9, 2020 • edited Loading

jrhizor commented Oct 9, 2020 • edited Loading

jrhizor commented Oct 14, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sherifnada left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrhizor commented Oct 14, 2020

jrhizor commented Oct 9, 2020 •

edited

Loading

jrhizor commented Oct 9, 2020 •

edited

Loading

sherifnada left a comment •

edited

Loading