fix(low-code): support AsyncRetriever in DynamicSchemaLoader

[devin-ai-integration]

Summary

AsyncRetriever is allowed as DynamicSchemaLoader.retriever in the schema, but it never worked there. There were three separate failures, and fixing only the first one moves the crash one step further:

1. create_dynamic_schema_loader passed partition_router= only, while create_async_retriever requires stream_slicer >> TypeError at construction (the error reported in AsyncRetriever cannot be used in DynamicSchemaLoader - parameter name mismatch #766).
2. DynamicSchemaLoader reads with read_records({}), i.e. without a slice. AsyncRetriever.read_records dereferenced stream_slice.cursor_slice and never created any job >> AttributeError.
3. AsyncRetriever yields a slice log message before its records, and next() took that message as the schema record >> TypeError.

Changes:

• model_to_component_factory.py: pass stream_slicer to AsyncRetriever only, so other retrievers (including a CustomRetriever with its own stream_slicer) get exactly the kwargs they got before. Reject an AsyncRetriever with a partition_router inside DynamicSchemaLoader with a clear ValueError: only the first record is read, so jobs for other partitions would be orphaned and keep holding max_concurrent_async_job_count slots. That configuration already failed at construction before this change.
• async_retriever.py: when called without a slice, create and poll the jobs through stream_slicer.stream_slices() and read each job slice. Calls with a slice (every stream partition) are unchanged.
• dynamic_schema_loader.py: skip log messages when taking the first record. SimpleRetriever only yields records, so its behaviour is unchanged.

Known limits (documented in the DynamicSchemaLoader docstring)

• Each schema load creates one async job (retried on failure) and stops at the first record, so the job's delete endpoint is not called.
• During a read, the schema loads on the first partition read, and its job takes a slot from the source-wide max_concurrent_async_job_count, next to the stream's own jobs.

Tests

• End-to-end discover with an async schema loader (creation >> polling >> download): JSON download, CSV download with the header as the schema (the source-blackbaud-sky-nxt shape from source-blackbaud-sky-nxt contribution from allandelmare airbyte#66221), and an empty download.
• End-to-end read with an async stream retriever plus an async schema loader: records are emitted and cast to the loaded schema, and the schema job is created once.
• A failed schema job during discover raises AirbyteTracedException.
• The partition_router rejection, a CustomRetriever keeping its manifest stream_slicer, and log messages being skipped.
• Each production hunk was reverted on its own and turns at least one new test red. The construction test from the first commit is kept.

Out of scope (follow-up issue)

create_http_components_resolver has the same kwarg mismatch (its partition router request_option is dropped since 7.0, and AsyncRetriever cannot be used there). It changes resolver HTTP requests, so it is tracked separately: #1181.

Fixes #766
Related: https://github.com/airbytehq/oncall/issues/11848

Link to Devin session: https://app.devin.ai/sessions/1035eef80d1f48f6ab01e96589caedb1

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment and CI monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1775066638-fix-dynamic-schema-loader-async-retriever#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1775066638-fix-dynamic-schema-loader-async-retriever

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 922 tests  +9   4 910 ✅ +9   8m 49s ⏱️ - 2m 4s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit e6a5155. ± Comparison against base commit f243b0f.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 925 tests  +9   4 913 ✅ +9   15m 40s ⏱️ -10s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit e6a5155. ± Comparison against base commit f243b0f.

♻️ This comment has been updated with latest results.

[Baz (bazarnov)]

Approved. The kwarg change alone was not enough: with it, the loader still crashed on the slice-less read and then on the slice log message. On top of it I added slice-less reads in AsyncRetriever, log-message skipping in DynamicSchemaLoader, stream_slicer passed to AsyncRetriever only, and a clear error for a partition_router on the schema loader's AsyncRetriever. End-to-end discover and read tests cover JSON, CSV (the source-blackbaud-sky-nxt shape from airbytehq/airbyte#66221), empty and failed downloads. The resolver has the same kwarg mismatch; it is tracked in #1181. All required checks are green on this head. Non-required failures are environmental: Check: destination-motherduck fails the same way on recently merged PRs, and Check: source-intercom failed on an Intercom permission-scope response and then on a PyPI install error, while the same commit passed it in run 36409941804. When squash-merging, drop the Co-Authored-By: bot_apk trailer and the stale "passing both stream_slicer and partition_router" line from the first commit message.

[devin-ai-integration]

❌ Cannot revive Devin session - the session is too old. Please start a new session instead.

View session