✨ Source Sendgrid: Move contacts stream to async declarative component

[maxi297]

What

Addresses https://github.com/airbytehq/airbyte-internal-issues/issues/9186

This is a temporary PR until the CDK version from #45178 is released The release has happened, this PR is ready to be reviewed! 🎉

How

Remove the Python code and move everything to the manifest

Review guide

1. airbyte-integrations/connectors/source-sendgrid/source_sendgrid/manifest.yaml
2. airbyte-integrations/connectors/source-sendgrid/unit_tests/unit_test.py Baz did revamp some test there but we made sure to keep the current ones of compressed/uncompressed files

User Impact

None as the behavior will be the same. It'll be easier for us to maintain though

Can this PR be safely reverted and rolled back?

• YES 💚
• NO ❌

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Sep 10, 2024 2:45pm

[maxi297]

CI is expected to fail for now as CDK version has not been released yet

[girarda]

🔥

[brianjlai]

seems good to me pending a live test run against the contacts stream. But I don't need to block approval. once thats ✅ then looks good to merge

[maxi297]

@brianjlai I unfortunately can't run live test (example) because the cache does not support encoding x-gzip which leads to this error:

+ Exception Group Traceback (most recent call last):
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/runner.py", line 341, in from_call
  |     result: TResult | None = func()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/runner.py", line 242, in 
  |     lambda: runtest_hook(item=item, **kwds), when=when, reraise=reraise
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_hooks.py", line 513, in __call__
  |     return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_manager.py", line 120, in _hookexec
  |     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 139, in _multicall
  |     raise exception.with_traceback(exception.__traceback__)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/unraisableexception.py", line 90, in pytest_runtest_setup
  |     yield from unraisable_exception_runtest_hook()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/unraisableexception.py", line 70, in unraisable_exception_runtest_hook
  |     yield
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/logging.py", line 842, in pytest_runtest_setup
  |     yield from self._runtest_for(item, "setup")
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/logging.py", line 831, in _runtest_for
  |     yield
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/capture.py", line 874, in pytest_runtest_setup
  |     return (yield)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 122, in _multicall
  |     teardown.throw(exception)  # type: ignore[union-attr]
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/threadexception.py", line 87, in pytest_runtest_setup
  |     yield from thread_exception_runtest_hook()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/threadexception.py", line 68, in thread_exception_runtest_hook
  |     yield
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 103, in _multicall
  |     res = hook_impl.function(*args)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/runner.py", line 160, in pytest_runtest_setup
  |     item.session._setupstate.setup(item)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/runner.py", line 514, in setup
  |     col.setup()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/python.py", line 1630, in setup
  |     self._request._fillfixtures()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/fixtures.py", line 696, in _fillfixtures
  |     item.funcargs[argname] = self.getfixturevalue(argname)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/fixtures.py", line 531, in getfixturevalue
  |     fixturedef = self._get_active_fixturedef(argname)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/fixtures.py", line 616, in _get_active_fixturedef
  |     fixturedef.execute(request=subrequest)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/fixtures.py", line 1090, in execute
  |     result = ihook.pytest_fixture_setup(fixturedef=self, request=request)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_hooks.py", line 513, in __call__
  |     return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_manager.py", line 120, in _hookexec
  |     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 182, in _multicall
  |     return outcome.get_result()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_result.py", line 100, in get_result
  |     raise exc.with_traceback(exc.__traceback__)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 167, in _multicall
  |     teardown.throw(outcome._exception)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/setuponly.py", line 36, in pytest_fixture_setup
  |     return (yield)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 103, in _multicall
  |     res = hook_impl.function(*args)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/fixtures.py", line 1139, in pytest_fixture_setup
  |     result = call_fixture_func(fixturefunc, request, kwargs)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/_pytest/fixtures.py", line 890, in call_fixture_func
  |     fixture_result = next(generator)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/anyio/pytest_plugin.py", line 77, in wrapper
  |     yield runner.run_fixture(func, kwargs)
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1987, in run_fixture
  |     retval = self.get_loop().run_until_complete(
  |   File "/usr/local/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
  |     return future.result()
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1957, in _call_in_runner_task
  |     return await future
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1932, in _run_tests_and_fixtures
  |     retval = await coro
  |   File "/app/src/live_tests/conftest.py", line 812, in read_control_execution_result
  |     execution_result = await read_control_connector_runner.run()
  |   File "/app/src/live_tests/commons/connector_runner.py", line 110, in run
  |     async with asyncer.create_task_group() as task_group:
  |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
  |     raise BaseExceptionGroup(
  | exceptiongroup.ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/mitmproxy/net/encoding.py", line 62, in decode
    |     decoded = custom_decode[encoding](encoded)
    | KeyError: 'x-gzip'
    | 
    | During handling of the above exception, another exception occurred:
    | 
    | Traceback (most recent call last):
    |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/mitmproxy/net/encoding.py", line 64, in decode
    |     decoded = codecs.decode(encoded, encoding, errors)  # type: ignore
    | LookupError: unknown encoding: x-gzip
    | 
    | During handling of the above exception, another exception occurred:
    | 
    | Traceback (most recent call last):
    |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/asyncer/_main.py", line 164, in value_wrapper
    |     value = await partial_f()
    |   File "/app/src/live_tests/commons/connector_runner.py", line 187, in _run
    |     await execution_result.save_artifacts(self.output_dir, self.duckdb_path)
    |   File "/app/src/live_tests/commons/models.py", line 428, in save_artifacts
    |     await self.save_http_dump(output_dir)
    |   File "/app/src/live_tests/commons/models.py", line 400, in save_http_dump
    |     mitm_http_stream_to_har(http_dump_file_path, har_file_path)
    |   File "/app/src/live_tests/commons/utils.py", line 153, in mitm_http_stream_to_har
    |     SaveHar().export_har(flows, str(har_file_path))
    |   File "/root/.cache/pypoetry/virtualenvs/live-tests-9TtSrW0h-py3.10/lib/python3.10/site-packages/mitmproxy/command.py", line 322, in wrapper

I've tested locally using the output of some connection. For documentation purposes, this is the script I wrote to test it:

import json

latest = "new.jsonl"
dev = "old.jsonl"


def _get_records(file_path):
    records = []
    with open(file_path, "r") as file:
        for line in file.readlines():
            message = json.loads(line)
            if message["type"] == "RECORD":
                records.append(message["record"]["data"])
    return records


def _get_distinct_ids(records):
    return set(map(lambda message: message["record"]["data"]["id"], records))


import dictdiffer

if __name__ == "__main__":
    latest_records = _get_records(latest)
    dev_records = _get_records(dev)

    for latest_record in latest_records:
        matching_dev_records = list(filter(lambda x: x["contact_id"] == latest_record["contact_id"], dev_records))
        if len(matching_dev_records) != 1:
            print("OOPS!!")
            continue

        matching_dev_record = {k: v for k, v in matching_dev_records[0].items() if v is not None}
        for diff in list(dictdiffer.diff(latest_record, matching_dev_record)):
            print(diff)
    difference = _get_distinct_ids(concurrency_records) - _get_distinct_ids(records)
    print(difference)

Are we good with this for a release or is there more we should do?