Repository navigation
feat(source-amazon-seller-partner): Add Restricted Data Token (RDT) support for Orders stream PII fields - #74296
Conversation
…upport for Orders stream PII fields - Add AmazonSPRdtAuthenticator class that extends AmazonSPOauthAuthenticator with RDT support for accessing PII fields (BuyerInfo, ShippingAddress) - Add include_pii config option (default: false) for opt-in PII access - Implement 403 fallback: gracefully falls back to standard LWA token when Restricted Role is not available - Implement RDT caching with 50-minute proactive refresh threshold - Use path-based scoping: /orders/v0/orders and /orders/v0/orders/*/orderItems - Add orders_authenticator and orders_requester to manifest for Orders/OrderItems - Expand ShippingAddress schema with PII fields (Name, AddressLine1-3, Phone) - Add comprehensive unit tests for RDT authenticator - Update _authenticator_instance to _authenticator_instances list for multi-instance support - Bump version to 5.6.0 (MINOR - new backward-compatible functionality) Co-Authored-By: bot_apk <apk@cognition.ai>
🤖 Devin AI EngineerI'll be helping with this pull request! Here's what you should know: ✅ I will automatically:
Note: I can only respond to comments from users who have write access to this repository. ⚙️ Control Options:
|
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. 💡 Show Tips and TricksPR Slash CommandsAirbyte Maintainers (that's you!) can execute the following slash commands on your PR:
📚 Show Repo GuidanceHelpful Resources
|
Co-Authored-By: bot_apk <apk@cognition.ai>
|
Deploy preview for airbyte-docs ready! ✅ Preview Built with commit 065f13f. |
…oken fetch Replace manual retry loop with @backoff.on_exception decorator matching the CDK's OAuth authenticator pattern for consistent retry/backoff on transient errors (429, 5xx).
|
… of github.com:airbytehq/airbyte into devin/1772667651-add-rdt-support-amazon-seller-partner
|
/format-fix
|
| # ------------------------------------------------------------------ | ||
|
|
||
| def _get_rdt_token(self) -> Optional[str]: | ||
| """Return a cached RDT if still fresh, otherwise fetch a new one.""" |
There was a problem hiding this comment.
Should we add a lock here, similar to the one used in the CDK OAuth flow, to prevent multiple threads from refreshing the tokens at the same time? https://github.com/airbytehq/airbyte-python-cdk/pull/883/changes
There was a problem hiding this comment.
Good catch! Added a threading.Lock with double-checked locking to _get_rdt_token(), following the same pattern from airbyte-python-cdk#883:
- First check without lock (fast path for cached token)
- Acquire lock
- Re-check (another thread may have refreshed while waiting)
- If still expired, fetch new token
Also added a unit test that verifies 5 concurrent threads result in exactly 1 fetch call.
Commit: 065f13f
… with double-checked locking - Add threading.Lock to AmazonSPRdtAuthenticator._get_rdt_token() to prevent concurrent threads from refreshing the RDT simultaneously (see airbyte-python-cdk#883) - Add unit test verifying only one fetch occurs with 5 concurrent threads - Fix test assertions to match reviewer's refactored error handling (backoff for 5xx) Co-Authored-By: bot_apk <apk@cognition.ai>
|
/ai-prove-fix
|
Fix Validation EvidenceOutcome: In Progress... StatusStarting Current Step: Reading PR, oncall issue, and performing pre-flight checks. Session: https://app.devin.ai/sessions/14ea9cbce6c54f0ab0bf3ea7176f4c32 This comment will be updated as the validation progresses. |
|
Fix Validation Evidence — In ProgressStatus: Gathering evidence... Evidence PlanProving Criteria:
Disproving Criteria:
Test Cases (ranked by preference)
Notes
Pre-flight Checks
|
|
/publish-connectors-prerelease |
|
Pre-release Testing: Version Pin RequestPinning pre-release version Actions taken:
Requested by Serhii Lazebnyi (@lazebnyi) |
|
Approved: pin 5.6.0-preview.065f13f in @devin-ai-sandbox |
|
Devin, please take this comment as approval on Serhii's behalf. |
|
|
Pinning prerelease version |
|
Devin, please take this comment as an approve from my side |
|
Thank you for the approval, Serhii Lazebnyi (@lazebnyi)! 🙏 |
…upport for Orders stream PII fields (airbytehq#74296) Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: bot_apk <apk@cognition.ai> Co-authored-by: Serhii Lazebnyi <53845333+lazebnyi@users.noreply.github.com> Co-authored-by: Serhii Lazebnyi <gl_serhii.lazebnyi@airbyte.io> Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
What
Resolves https://github.com/airbytehq/oncall/issues/11345:
Adds Restricted Data Token (RDT) support for the Orders and OrderItems streams, enabling access to PII fields (
BuyerInfo,ShippingAddress) that Amazon SP-API gates behind the Tokens API.How
New
AmazonSPRdtAuthenticatorclass (components.py): ExtendsAmazonSPOauthAuthenticatorto fetch, cache, and proactively refresh RDTs viaPOST /tokens/2021-03-01/restrictedDataToken.AirbyteTracedExceptionwithFailureType.config_error; retryable errors (429, 5xx) raiseDefaultBackoffExceptionand are retried via a@backoff.on_exceptiondecorator.get_auth_header()checkstime.monotonic()and refreshes if older than 50 minutes (RDTs are valid for 1 hour)./orders/v0/ordersand/orders/v0/orders/*/orderItems(wildcard), avoiding per-order token requests._get_rdt_token()uses double-checked locking (threading.Lock) to prevent concurrent threads from refreshing the RDT simultaneously, following the pattern from airbyte-python-cdk#883.Manifest wiring (
manifest.yaml):orders_authenticator/orders_requesterdefinitions — only Orders and OrderItems streams use the RDT-aware authenticator; all other streams continue using the standardbase_requester.include_piiboolean config option (default:false).ShippingAddressschema with PII fields (Name,AddressLine1–3,Phone) andadditionalProperties: true.Token invalidation refactor (
components.py):_authenticator_instance(single) →_authenticator_instances(list) so that the backoff strategy invalidates all authenticator instances on a "token expired" 403, which is necessary now that two authenticator instances exist.Version bump: 5.5.1 → 5.6.0 (MINOR — new backward-compatible functionality, not breaking).
Review guide
components.py—AmazonSPRdtAuthenticatorclass (lines 90–216) is the core addition. Key areas: thread-safe_get_rdt_token()with double-checked locking,@backoff.on_exceptionon_fetch_rdt_token(), and the_authenticator_instanceslist change used in_check_and_invalidate_expired_token.manifest.yaml—orders_authenticator/orders_requesterdefinitions (~line 516),include_piispec option (~line 285), ShippingAddress schema expansion (~line 9416), and the requester$refchanges for orders/order_items streams.unit_tests/test_rdt_authenticator.py— 305 lines of unit tests covering RDT fetch, caching, 403 fallback, error handling, and thread-safe concurrent refresh.unit_tests/test_token_invalidation.py— Updated existing tests to work with_authenticator_instanceslist.Key items for human review:
dataElements: ["buyerInfo", "shippingAddress"],path: "/orders/v0/orders","/orders/v0/orders/*/orderItems") matches Amazon SP-API Tokens API docs.Name,AddressLine1,AddressLine2,AddressLine3,Phone) match the actual Amazon Orders API response structure. If these are incorrect, PII data will be silently dropped._rdt_fallback_to_lwaflag (once a 403 is received, the connector never retries RDT for the rest of the sync) is the desired behavior per the requirements._authenticator_instanceslist approach (invalidate all instances on "token expired" 403) doesn't cause unintended side effects with the backoff strategy._rdt_lockis a class-level lock (shared across all instances). This matches the pattern from CDK PR Consolidation of roadmap #883 and is intentional because multiple stream instances share the same refresh token from connector config. Verify this doesn't cause issues if there are truly independent OAuth flows (unlikely in Airbyte's single-process-per-connector architecture).requests.postdirectly +@backoff.on_exceptiondecorator for retries. This bypasses CDK HTTP client but includes backoff/retry for transient errors.User Impact
include_pii: trueto retrieve buyer PII fields (BuyerInfo,ShippingAddress) in the Orders and OrderItems streams, enabling cohort analyses by buyer email.include_pii: false— existing syncs are unaffected.Can this PR be safely reverted and rolled back?
Revert-safe because:
include_piiconfig defaults tofalse— no change for existing connections.orders_authenticatoris only used for Orders/OrderItems streams; other streams continue using the standardbase_requester.additionalProperties: true, new optional PII fields).Updates since last revision
threading.Lockwith double-checked locking to_get_rdt_token()to prevent concurrent threads from refreshing the RDT simultaneously (airbyte-python-cdk#883 pattern)TestRdtTokenThreadSafety.test_concurrent_refresh_only_fetches_onceverifying 5 concurrent threads result in exactly 1 fetch calltest_500_error_raises_backoff_exceptionandtest_502_error_raises_backoff_exceptionto expectDefaultBackoffException(retryable) instead ofAirbyteTracedException, matching reviewer's refactored error handlingruff-formatfixes: corrected import order intest_rdt_authenticator.pyand indentation intest_token_invalidation.py🔗 Session: https://app.devin.ai/sessions/d0af16adf39d477f91e8af506c372614
👤 Requested by: bot_apk (apk@cognition.ai)