Merge branch 'main' into lazebnyi/add-state-delegating-retriever

Author: lazebnyi

airbyte_cdk/sources/declarative/datetime/__init__.py

@@ -1,7 +1,3 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-
-from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
-
-__all__ = ["MinMaxDatetime"]

airbyte_cdk/sources/declarative/datetime/datetime_parser.py

@@ -29,6 +29,8 @@ def parse(self, date: Union[str, int], format: str) -> datetime.datetime:
             return datetime.datetime.fromtimestamp(int(date), tz=datetime.timezone.utc)
         elif format == "%s_as_float":
             return datetime.datetime.fromtimestamp(float(date), tz=datetime.timezone.utc)
+        elif format == "%epoch_microseconds":
+            return self._UNIX_EPOCH + datetime.timedelta(microseconds=int(date))
         elif format == "%ms":
             return self._UNIX_EPOCH + datetime.timedelta(milliseconds=int(date))
         elif "%_ms" in format:
@@ -46,6 +48,8 @@ def format(self, dt: datetime.datetime, format: str) -> str:
             return str(int(dt.timestamp()))
         if format == "%s_as_float":
             return str(float(dt.timestamp()))
+        if format == "%epoch_microseconds":
+            return str(int(dt.timestamp() * 1_000_000))
         if format == "%ms":
             # timstamp() returns a float representing the number of seconds since the unix epoch
             return str(int(dt.timestamp() * 1000))

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -151,6 +151,10 @@ def decode(
         self, response: requests.Response
     ) -> Generator[MutableMapping[str, Any], None, None]:
         if self.is_stream_response():
+            # urllib mentions that some interfaces don't play nice with auto_close [here](https://urllib3.readthedocs.io/en/stable/user-guide.html#using-io-wrappers-with-response-content)
+            # We have indeed observed some issues with CSV parsing. Hence, we will manage the closing of the file ourselves until we find a better solution.
+            response.raw.auto_close = False
             yield from self.parser.parse(data=response.raw)  # type: ignore[arg-type]
+            response.raw.close()
         else:
             yield from self.parser.parse(data=io.BytesIO(response.content))

airbyte_cdk/sources/declarative/interpolation/macros.py

@@ -12,6 +12,8 @@
 from dateutil import parser
 from isodate import parse_duration
 
+from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
+
 """
 This file contains macros that can be evaluated by a `JinjaInterpolation` object
 """
@@ -171,11 +173,7 @@ def format_datetime(
     dt_datetime = (
         datetime.datetime.strptime(dt, input_format) if input_format else str_to_datetime(dt)
     )
-    if format == "%s":
-        return str(int(dt_datetime.timestamp()))
-    elif format == "%ms":
-        return str(int(dt_datetime.timestamp() * 1_000_000))
-    return dt_datetime.strftime(format)
+    return DatetimeParser().format(dt=dt_datetime, format=format)
 
 
 _macros_list = [

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -57,7 +57,7 @@
 )
 from airbyte_cdk.sources.declarative.checks import CheckDynamicStream, CheckStream
 from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
-from airbyte_cdk.sources.declarative.datetime import MinMaxDatetime
+from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.declarative_stream import DeclarativeStream
 from airbyte_cdk.sources.declarative.decoders import (
     Decoder,

airbyte_cdk/sources/declarative/yaml_declarative_source.py

@@ -50,7 +50,6 @@ def _read_and_parse_yaml_file(self, path_to_yaml_file: str) -> ConnectionDefinit
 
     def _emit_manifest_debug_message(self, extra_args: dict[str, Any]) -> None:
         extra_args["path_to_yaml"] = self._path_to_yaml
-        self.logger.debug("declarative source created from parsed YAML manifest", extra=extra_args)
 
     @staticmethod
     def _parse(connection_definition_str: str) -> ConnectionDefinition:

unit_tests/sources/declarative/datetime/test_datetime_parser.py

@@ -10,102 +10,118 @@
 
 
 @pytest.mark.parametrize(
-    "test_name, input_date, date_format, expected_output_date",
+    "input_date, date_format, expected_output_date",
     [
         (
-            "test_parse_date_iso",
             "2021-01-01T00:00:00.000000+0000",
             "%Y-%m-%dT%H:%M:%S.%f%z",
             datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
         ),
         (
-            "test_parse_date_iso_with_timezone_not_utc",
             "2021-01-01T00:00:00.000000+0400",
             "%Y-%m-%dT%H:%M:%S.%f%z",
             datetime.datetime(
                 2021, 1, 1, 0, 0, tzinfo=datetime.timezone(datetime.timedelta(seconds=14400))
             ),
         ),
         (
-            "test_parse_timestamp",
             "1609459200",
             "%s",
             datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
         ),
         (
-            "test_parse_timestamp_as_float",
             "1675092508.873709",
             "%s_as_float",
             datetime.datetime(2023, 1, 30, 15, 28, 28, 873709, tzinfo=datetime.timezone.utc),
         ),
         (
-            "test_parse_ms_timestamp",
+            "1675092508873709",
+            "%epoch_microseconds",
+            datetime.datetime(2023, 1, 30, 15, 28, 28, 873709, tzinfo=datetime.timezone.utc),
+        ),
+        (
             "1609459200001",
             "%ms",
             datetime.datetime(2021, 1, 1, 0, 0, 0, 1000, tzinfo=datetime.timezone.utc),
         ),
         (
-            "test_parse_date_ms",
             "20210101",
             "%Y%m%d",
             datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
         ),
         (
-            "test_parse_format_datetime_with__ms",
             "2021-11-22T08:41:55.640Z",
             "%Y-%m-%dT%H:%M:%S.%_msZ",
             datetime.datetime(2021, 11, 22, 8, 41, 55, 640000, tzinfo=datetime.timezone.utc),
         ),
     ],
+    ids=[
+        "test_parse_date_iso",
+        "test_parse_date_iso_with_timezone_not_utc",
+        "test_parse_timestamp",
+        "test_parse_timestamp_as_float",
+        "test_parse_timestamp_microseconds",
+        "test_parse_ms_timestamp",
+        "test_parse_date_ms",
+        "test_parse_format_datetime_with__ms",
+    ],
 )
-def test_parse_date(test_name, input_date, date_format, expected_output_date):
+def test_parse_date(input_date: str, date_format: str, expected_output_date: datetime.datetime):
     parser = DatetimeParser()
     output_date = parser.parse(input_date, date_format)
     assert output_date == expected_output_date
 
 
 @pytest.mark.parametrize(
-    "test_name, input_dt, datetimeformat, expected_output",
+    "input_dt, datetimeformat, expected_output",
     [
         (
-            "test_format_timestamp",
             datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
             "%s",
             "1609459200",
         ),
         (
-            "test_format_timestamp_ms",
             datetime.datetime(2021, 1, 1, 0, 0, 0, 1000, tzinfo=datetime.timezone.utc),
             "%ms",
             "1609459200001",
         ),
         (
-            "test_format_timestamp_as_float",
             datetime.datetime(2023, 1, 30, 15, 28, 28, 873709, tzinfo=datetime.timezone.utc),
             "%s_as_float",
             "1675092508.873709",
         ),
         (
-            "test_format_string",
+            datetime.datetime(2023, 1, 30, 15, 28, 28, 873709, tzinfo=datetime.timezone.utc),
+            "%epoch_microseconds",
+            "1675092508873709",
+        ),
+        (
             datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
             "%Y-%m-%d",
             "2021-01-01",
         ),
         (
-            "test_format_to_number",
             datetime.datetime(2021, 1, 1, 0, 0, tzinfo=datetime.timezone.utc),
             "%Y%m%d",
             "20210101",
         ),
         (
-            "test_parse_format_datetime_with__ms",
             datetime.datetime(2021, 11, 22, 8, 41, 55, 640000, tzinfo=datetime.timezone.utc),
             "%Y-%m-%dT%H:%M:%S.%_msZ",
             "2021-11-22T08:41:55.640Z",
         ),
     ],
+    ids=[
+        "test_format_timestamp",
+        "test_format_timestamp_ms",
+        "test_format_timestamp_as_float",
+        "test_format_timestamp_microseconds",
+        "test_format_string",
+        "test_format_to_number",
+        "test_parse_format_datetime_with__ms",
+    ],
 )
-def test_format_datetime(test_name, input_dt, datetimeformat, expected_output):
+def test_format_datetime(input_dt: datetime.datetime, datetimeformat: str, expected_output: str):
     parser = DatetimeParser()
     output_date = parser.format(input_dt, datetimeformat)
     assert output_date == expected_output

unit_tests/sources/declarative/decoders/test_composite_decoder.py

@@ -4,7 +4,10 @@
 import csv
 import gzip
 import json
+import socket
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from io import BytesIO, StringIO
+from threading import Thread
 from unittest.mock import patch
 
 import pytest
@@ -20,6 +23,12 @@
 from airbyte_cdk.utils import AirbyteTracedException
 
 
+def find_available_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("localhost", 0))
+        return s.getsockname()[1]  # type: ignore  # this should return a int
+
+
 def compress_with_gzip(data: str, encoding: str = "utf-8"):
     """
     Compress the data using Gzip.
@@ -202,6 +211,39 @@ def test_composite_raw_decoder_csv_parser_values(requests_mock, encoding: str, d
     assert parsed_records == expected_data
 
 
+class TestServer(BaseHTTPRequestHandler):
+    def do_GET(self) -> None:
+        self.send_response(200)
+        self.end_headers()
+        self.wfile.write(bytes("col1,col2\nval1,val2", "utf-8"))
+
+
+def test_composite_raw_decoder_csv_parser_without_mocked_response():
+    """
+    This test reproduce a `ValueError: I/O operation on closed file` error we had with CSV parsing. We could not catch this with other tests because the closing of the mocked response from requests_mock was not the same as the one in requests.
+
+    We first identified this issue while working with the sample defined https://people.sc.fsu.edu/~jburkardt/data/csv/addresses.csv.
+    This should be reproducible by having the test server return the `self.wfile.write` statement as a comment below but it does not. However, it wasn't reproducible.
+
+    Currently we use `self.wfile.write(bytes("col1,col2\nval1,val2", "utf-8"))` to reproduce which we know is not a valid csv as it does not end with a newline character. However, this is the only we were able to reproduce locally.
+    """
+    # self.wfile.write(bytes('John,Doe,120 jefferson st.,Riverside, NJ, 08075\nJack,McGinnis,220 hobo Av.,Phila, PA,09119\n"John ""Da Man""",Repici,120 Jefferson St.,Riverside, NJ,08075\nStephen,Tyler,"7452 Terrace ""At the Plaza"" road",SomeTown,SD, 91234\n,Blankman,,SomeTown, SD, 00298\n"Joan ""the bone"", Anne",Jet,"9th, at Terrace plc",Desert City,CO,00123\n', "utf-8"))
+
+    # start server
+    port = find_available_port()
+    httpd = HTTPServer(("localhost", port), TestServer)
+    thread = Thread(target=httpd.serve_forever, args=())
+    thread.start()
+    try:
+        response = requests.get(f"http://localhost:{port}", stream=True)
+        result = list(CompositeRawDecoder(parser=CsvParser()).decode(response))
+
+        assert len(result) == 1
+    finally:
+        httpd.shutdown()  # release port and kill the thread
+        thread.join(timeout=5)  # ensure thread is cleaned up
+
+
 def test_given_response_already_consumed_when_decode_then_no_data_is_returned(requests_mock):
     requests_mock.register_uri(
         "GET", "https://airbyte.io/", content=json.dumps({"test": "test"}).encode()

unit_tests/sources/declarative/extractors/test_record_filter.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-from airbyte_cdk.sources.declarative.datetime import MinMaxDatetime
+from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.extractors.record_filter import (
     ClientSideIncrementalRecordFilterDecorator,
     RecordFilter,

unit_tests/sources/declarative/interpolation/test_macros.py

@@ -79,7 +79,7 @@ def test_macros_export(test_name, fn_name, found_in_macros):
         ),
         (
             "2022-01-01T01:01:01Z",
-            "%ms",
+            "%epoch_microseconds",
             "%Y-%m-%dT%H:%M:%SZ",
             "1640998861000000",
         ),