refactor(crm): static schemas + raw_data for salesforce/hubspot - #2212
Conversation
Discover built schemas from a live describe: bronze DDL needed credentials and column sets varied per org/portal. Schemas are now repository artifacts and discover issues no API call. - salesforce: per-stream schema files, auth on demand, stream availability resolved per read - hubspot: allowlist columns emitted unconditionally, all portal properties fetched - raw_data carries the whole record; custom_fields removed as a subset - bootstrap-db and the connectors-ddl gate run without credentials BREAKING CHANGE: silver.class_crm_* lose custom_fields (migration 20260805000000_crm-drop-custom-fields.sql; staging healed in apply-ch-migrations.sh). Refs #756 #757 Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
📝 WalkthroughWalkthroughHubSpot and Salesforce now use static stream schemas and preserve undeclared source fields in ChangesCRM generation and schema contract
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (5)
src/ingestion/connectors/crm/salesforce/source_salesforce/source.py (2)
181-186: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
_get_stream_typeignoresstream_name.The method returns the same pair for every stream. The
stream_nameparameter is now dead. Keep it only if a caller or test depends on the signature; otherwise inline the constant pair inprepare_stream.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/source_salesforce/source.py` around lines 181 - 186, Update _get_stream_type so it no longer accepts the unused stream_name parameter unless callers or tests require that signature; otherwise remove the method and inline the constant RestSalesforceStream, IncrementalRestSalesforceStream pair directly in prepare_stream, updating all call sites accordingly.
180-201: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd the missing type hints on the changed signatures.
_get_stream_typehas no return annotation.prepare_streamhas no parameter or return annotations. The repository guideline requires type hints on every function and method signature undersrc/ingestion/.♻️ Proposed annotations
`@classmethod` - def _get_stream_type(cls, stream_name: str): + def _get_stream_type(cls, stream_name: str) -> Tuple[Type[Stream], Type[Stream]]: """Get proper stream class: full_refresh or incremental. Every stream uses the REST ``/queryAll`` API. """ return RestSalesforceStream, IncrementalRestSalesforceStream - def prepare_stream(self, stream_name, json_schema, sobject_options, sf_object, authenticator, config): + def prepare_stream( + self, + stream_name: str, + json_schema: Mapping[str, Any], + sobject_options: Mapping[str, Any], + sf_object: Salesforce, + authenticator: SalesforceAuthenticator, + config: Mapping[str, Any], + ) -> Tuple[Type[Stream], MutableMapping[str, Any]]:
Typemust be added to thetypingimport list.As per coding guidelines: "Put type hints on every function and method signature; do not allow bare
Anyto escape."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/source_salesforce/source.py` around lines 180 - 201, Add complete type annotations to the changed _get_stream_type and prepare_stream signatures, including the return type for _get_stream_type and parameter plus return types for prepare_stream. Add Type to the typing imports as required, and use concrete existing stream/config/schema types rather than allowing untyped or bare Any values to escape.Source: Coding guidelines
src/ingestion/connectors/crm/salesforce/tests/conftest.py (1)
99-106: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAnnotate
make_streamto satisfy the signature requirement.
sf: Salesforce | Noneis safe because the module enablesfrom __future__ import annotationsand the connector requires Python >=3.10, butmake_streamstill needs a return annotation and all parameters should avoid bareAnyas required undersrc/ingestion/.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/tests/conftest.py` around lines 99 - 106, Update the make_stream signature with an explicit return annotation and replace bare Any annotations on sf_fields and extra with precise types consistent with the surrounding src/ingestion conventions. Preserve the existing defaults and parameter behavior.Source: Coding guidelines
src/ingestion/connectors/crm/salesforce/tests/test_envelope.py (1)
103-117: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd tests for two non-obvious truncation rules.
The truncation tests use
Description, which is outsideDECLARED. Two semantics stay uncovered:
- A declared field keeps its full value in the top-level column while
raw_dataholds the truncated copy. The envelope truncates only therawaccumulator, so the same field has two different lengths in one output row._truncateslices on bytes and decodes witherrors="ignore". A multi-byte character that straddles the 2034-byte boundary is dropped so the result stays valid UTF-8.💚 Proposed tests
def test_declared_column_keeps_full_value_while_raw_data_truncates(self): long_name = "x" * 5000 out = _wrap({"Id": "001", "Name": long_name}) assert out["Name"] == long_name assert json.loads(out["raw_data"])["Name"].endswith("…[truncated]") def test_multibyte_character_split_at_the_cap_is_dropped(self): out = _wrap({"Id": "001", "Description": "é" * 2000}) truncated = json.loads(out["raw_data"])["Description"] assert truncated.endswith("…[truncated]") assert len(truncated.encode("utf-8")) <= 2048As per path instructions: "Test non-obvious semantics with a test whose name states the rule rather than adding a comment."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/tests/test_envelope.py` around lines 103 - 117, Extend the truncation tests in test_envelope.py with cases named for both rules: verify a declared Name field remains full-length in out["Name"] while raw_data contains its truncated value, and verify truncating repeated multi-byte "é" characters drops any character split at the byte cap while preserving the suffix and the 2048-byte limit. Use _wrap and JSON decoding consistently with the existing tests.Source: Path instructions
src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py (1)
25-37: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winRaise a typed error for a malformed schema file.
_loadraisesUnknownStreamSchemaErroronly when the file is absent. Two other failure kinds escape untyped:
- Invalid JSON raises
json.JSONDecodeError.- A schema file without a
propertieskey makesdeclared_field_namesraiseKeyError.Both failures reach the caller during discovery and produce a message that does not name the offending file. Convert them at the failure site.
The coding guidelines require one typed exception per failure kind, with detailed context logged at the failure site.
♻️ Proposed refactor
class UnknownStreamSchemaError(Exception): pass +class InvalidStreamSchemaError(Exception): + pass + + `@lru_cache`(maxsize=None) def available_stream_names() -> FrozenSet[str]: return frozenset(p.name[: -len(_SCHEMA_SUFFIX)] for p in _SCHEMA_DIR.glob(f"*{_SCHEMA_SUFFIX}")) `@lru_cache`(maxsize=None) def _load(stream_name: str) -> Mapping[str, Any]: path = _SCHEMA_DIR / f"{stream_name}{_SCHEMA_SUFFIX}" if not path.is_file(): raise UnknownStreamSchemaError( f"No static schema for stream '{stream_name}'; expected {path.name}" ) - schema: Mapping[str, Any] = json.loads(path.read_text()) + + try: + schema: Mapping[str, Any] = json.loads(path.read_text()) + except json.JSONDecodeError as exc: + raise InvalidStreamSchemaError(f"Static schema {path.name} is not valid JSON") from exc + + if "properties" not in schema: + raise InvalidStreamSchemaError(f"Static schema {path.name} has no 'properties' object") + return schemaAs per coding guidelines: "Use named helpers for repetition, log detailed context at the failure site, and raise one typed exception per failure kind."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py` around lines 25 - 37, Update _load to catch JSON decoding failures, log detailed context including the offending schema filename, and raise a dedicated typed exception for invalid JSON. Update declared_field_names to validate the loaded schema contains properties, logging the filename and raising a separate typed exception when it is absent instead of allowing KeyError to escape. Preserve UnknownStreamSchemaError for missing files and use named helpers if needed to avoid repeating failure-context handling.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/domain/connector/specs/ADR/0004-static-stream-schemas-with-raw-data.md`:
- Around line 79-80: Update the raw_data string-value rule in the static stream
schemas ADR to specify the 2 KB maximum and clarify whether the limit is
measured in UTF-8 bytes or characters. Add a boundary test covering values at
the limit and just beyond it, preserving the guarantee that the serialized blob
is never truncated and remains parseable.
In `@src/ingestion/connectors/crm/hubspot/source_hubspot/envelope.py`:
- Around line 60-66: Complete the type annotations for the affected callables:
define and use a recursive JSON-value type in _truncate_deep instead of bare
Any; parameterize allowed_property_names as an immutable string set in
envelope.py; annotate the helper input and return in test_envelope.py; and
annotate the batch input and iterable return in test_base_stream.py. Update
every listed signature while preserving existing behavior.
In `@src/ingestion/connectors/crm/hubspot/tests/test_api.py`:
- Around line 126-156: Add return type annotations of None to every changed test
method in test_api.py, including the parameterized test’s object_type parameter
annotated as str. Also annotate FakeHubspot.__init__ in conftest.py with ->
None; make no other changes.
In
`@src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Account.schema.json`:
- Around line 105-110: The Salesforce compound address fields are declared as
strings but are returned as nested objects. Update Account.schema.json lines
105-110, Contact.schema.json lines 117-122, and Lead.schema.json lines 123-128
for BillingAddress, ShippingAddress, Address, MailingAddress, and OtherAddress
to use the shared compound address type consistently, or remove these declared
columns if they are exposed only through raw_data.
In
`@src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/User.schema.json`:
- Around line 123-128: Update the User schema’s Address property to match
Salesforce’s compound-object shape by changing its type union from string/null
to object/null, or remove the Address property if it should not be exposed as a
column; do not leave it declared as a string.
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py`:
- Around line 113-129: Cache the declared field set before the record loop in
read_records and reuse it when calling envelope, rather than resolving
self.declared_fields for each record. Update declared_field_names in
schema_loader.py to cache the frozenset derived from
_load(stream_name)["properties"], while leaving get_json_schema’s existing
single _load call unchanged.
In `@src/ingestion/connectors/crm/salesforce/tests/test_api.py`:
- Around line 35-44: Update every changed test method in
src/ingestion/connectors/crm/salesforce/tests/test_api.py ranges 35-44, 184-190,
213-238, and 261-277, and
src/ingestion/connectors/crm/salesforce/tests/test_source.py ranges 201-208,
233-255, 270-274, and 332-340, adding appropriate fixture parameter types and ->
None return annotations. In test_source.py range 233-255, also annotate _sf_stub
with -> Salesforce; ensure all affected function and method signatures are fully
typed without bare Any escaping.
In `@src/ingestion/scripts/bootstrap-db/README.md`:
- Around line 86-90: Update the fork pull request documentation near the
workflow description to state that fork PRs run validation and receive the
regenerated snapshot artifact without CRM secrets, while limiting regen-pr to
same-repository pull requests.
In `@src/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql`:
- Around line 12-20: Migrate existing bronze tables from custom_fields to
raw_data while preserving raw_data immediately after collected_at. In
src/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql
(lines 12-20), add ALTER statements for every bronze_hubspot and
bronze_salesforce relation; in src/ingestion/scripts/apply-ch-migrations.sh
(lines 102-114), include those relations in contract healing. Keep the target
declarations in src/ingestion/scripts/connectors-ddl/hubspot.sql (lines 31-619)
and src/ingestion/scripts/connectors-ddl/salesforce.sql (lines 77-736), adding
matching existing-table ALTER statements there.
---
Nitpick comments:
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py`:
- Around line 25-37: Update _load to catch JSON decoding failures, log detailed
context including the offending schema filename, and raise a dedicated typed
exception for invalid JSON. Update declared_field_names to validate the loaded
schema contains properties, logging the filename and raising a separate typed
exception when it is absent instead of allowing KeyError to escape. Preserve
UnknownStreamSchemaError for missing files and use named helpers if needed to
avoid repeating failure-context handling.
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/source.py`:
- Around line 181-186: Update _get_stream_type so it no longer accepts the
unused stream_name parameter unless callers or tests require that signature;
otherwise remove the method and inline the constant RestSalesforceStream,
IncrementalRestSalesforceStream pair directly in prepare_stream, updating all
call sites accordingly.
- Around line 180-201: Add complete type annotations to the changed
_get_stream_type and prepare_stream signatures, including the return type for
_get_stream_type and parameter plus return types for prepare_stream. Add Type to
the typing imports as required, and use concrete existing stream/config/schema
types rather than allowing untyped or bare Any values to escape.
In `@src/ingestion/connectors/crm/salesforce/tests/conftest.py`:
- Around line 99-106: Update the make_stream signature with an explicit return
annotation and replace bare Any annotations on sf_fields and extra with precise
types consistent with the surrounding src/ingestion conventions. Preserve the
existing defaults and parameter behavior.
In `@src/ingestion/connectors/crm/salesforce/tests/test_envelope.py`:
- Around line 103-117: Extend the truncation tests in test_envelope.py with
cases named for both rules: verify a declared Name field remains full-length in
out["Name"] while raw_data contains its truncated value, and verify truncating
repeated multi-byte "é" characters drops any character split at the byte cap
while preserving the suffix and the 2048-byte limit. Use _wrap and JSON decoding
consistently with the existing tests.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 957d53d4-632b-40f7-9bd1-56001b77d4cc
📒 Files selected for processing (63)
.claude/skills/connector/workflows/create.md.github/workflows/connectors-ddl.ymldocs/domain/connector/specs/ADR/0004-static-stream-schemas-with-raw-data.mddocs/domain/ingestion-data-flow/specs/DESIGN.mdsrc/ingestion/connectors/crm/hubspot/README.mdsrc/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_accounts.sqlsrc/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_activities.sqlsrc/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_contacts.sqlsrc/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_deals.sqlsrc/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_users.sqlsrc/ingestion/connectors/crm/hubspot/dbt/schema.ymlsrc/ingestion/connectors/crm/hubspot/descriptor.yamlsrc/ingestion/connectors/crm/hubspot/source_hubspot/api.pysrc/ingestion/connectors/crm/hubspot/source_hubspot/constants.pysrc/ingestion/connectors/crm/hubspot/source_hubspot/envelope.pysrc/ingestion/connectors/crm/hubspot/source_hubspot/streams.pysrc/ingestion/connectors/crm/hubspot/tests/conftest.pysrc/ingestion/connectors/crm/hubspot/tests/test_api.pysrc/ingestion/connectors/crm/hubspot/tests/test_archived_stream.pysrc/ingestion/connectors/crm/hubspot/tests/test_base_stream.pysrc/ingestion/connectors/crm/hubspot/tests/test_envelope.pysrc/ingestion/connectors/crm/hubspot/tests/test_owners_streams.pysrc/ingestion/connectors/crm/hubspot/tests/test_search_stream.pysrc/ingestion/connectors/crm/hubspot/tests/test_source.pysrc/ingestion/connectors/crm/salesforce/README.mdsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_accounts.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_activities.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_contacts.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_deals.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_users.sqlsrc/ingestion/connectors/crm/salesforce/dbt/schema.ymlsrc/ingestion/connectors/crm/salesforce/descriptor.yamlsrc/ingestion/connectors/crm/salesforce/pyproject.tomlsrc/ingestion/connectors/crm/salesforce/source_salesforce/api.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/constants.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/envelope.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/exceptions.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/source.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Account.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Case.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Contact.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Event.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Lead.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Opportunity.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/OpportunityContactRole.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/OpportunityHistory.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Task.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/User.schema.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/streams.pysrc/ingestion/connectors/crm/salesforce/tests/conftest.pysrc/ingestion/connectors/crm/salesforce/tests/test_api.pysrc/ingestion/connectors/crm/salesforce/tests/test_envelope.pysrc/ingestion/connectors/crm/salesforce/tests/test_source.pysrc/ingestion/connectors/crm/salesforce/tests/test_streams.pysrc/ingestion/scripts/apply-ch-migrations.shsrc/ingestion/scripts/bootstrap-db/.env.bootstrap.examplesrc/ingestion/scripts/bootstrap-db/README.mdsrc/ingestion/scripts/bootstrap-db/connectors-config.yamlsrc/ingestion/scripts/connectors-ddl/hubspot.sqlsrc/ingestion/scripts/connectors-ddl/salesforce.sqlsrc/ingestion/scripts/connectors-ddl/silver.sqlsrc/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql
💤 Files with no reviewable changes (14)
- src/ingestion/connectors/crm/hubspot/tests/test_source.py
- src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_accounts.sql
- src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_contacts.sql
- src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_accounts.sql
- src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_users.sql
- src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_deals.sql
- src/ingestion/connectors/crm/salesforce/source_salesforce/exceptions.py
- src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_contacts.sql
- src/ingestion/connectors/crm/hubspot/dbt/hubspot__crm_activities.sql
- src/ingestion/scripts/bootstrap-db/.env.bootstrap.example
- src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_deals.sql
- src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_activities.sql
- src/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_users.sql
- src/ingestion/scripts/connectors-ddl/silver.sql
| 4. String values inside `raw_data` are capped per value. The serialized blob is | ||
| never truncated, so it always parses. |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Document the raw_data value limit.
The rule says that string values are capped, but it does not define the limit. The PR contract specifies a 2 KB cap per value. State the numeric limit and the unit, such as UTF-8 bytes or characters. Add a boundary test for this contract.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/domain/connector/specs/ADR/0004-static-stream-schemas-with-raw-data.md`
around lines 79 - 80, Update the raw_data string-value rule in the static stream
schemas ADR to specify the 2 KB maximum and clarify whether the limit is
measured in UTF-8 bytes or characters. Add a boundary test covering values at
the limit and just beyond it, preserving the guarantee that the serialized blob
is never truncated and remains parseable.
| def _truncate_deep(value: Any) -> Any: | ||
| """Copy ``value``, truncating every string it contains.""" | ||
| if isinstance(value, Mapping): | ||
| return {k: _truncate_deep(v) for k, v in value.items()} | ||
| if isinstance(value, (list, tuple)): | ||
| return [_truncate_deep(v) for v in value] | ||
| return _truncate(value) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add complete types to the changed callable interfaces.
Replace bare Any in _truncate_deep() with a recursive JSON-value type. Parameterize allowed_property_names as a string set. Add parameter and return types to wrap() and spy().
src/ingestion/connectors/crm/hubspot/source_hubspot/envelope.py#L60-L66: use a recursive JSON-value type instead ofAny.src/ingestion/connectors/crm/hubspot/source_hubspot/envelope.py#L73-L80: use a parameterized immutable string set type.src/ingestion/connectors/crm/hubspot/tests/test_envelope.py#L14-L15: annotate the helper input and return value.src/ingestion/connectors/crm/hubspot/tests/test_base_stream.py#L164-L166: annotate the batch input and iterable return value.
As per coding guidelines, “Put type hints on every function and method signature; do not allow bare Any to escape.”
📍 Affects 3 files
src/ingestion/connectors/crm/hubspot/source_hubspot/envelope.py#L60-L66(this comment)src/ingestion/connectors/crm/hubspot/source_hubspot/envelope.py#L73-L80src/ingestion/connectors/crm/hubspot/tests/test_envelope.py#L14-L15src/ingestion/connectors/crm/hubspot/tests/test_base_stream.py#L164-L166
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/connectors/crm/hubspot/source_hubspot/envelope.py` around lines
60 - 66, Complete the type annotations for the affected callables: define and
use a recursive JSON-value type in _truncate_deep instead of bare Any;
parameterize allowed_property_names as an immutable string set in envelope.py;
annotate the helper input and return in test_envelope.py; and annotate the batch
input and iterable return in test_base_stream.py. Update every listed signature
while preserving existing behavior.
Source: Coding guidelines
| def test_every_named_portal_property_is_requested(self): | ||
| hs = make_client([FakeResponse({"results": self.DESCRIPTORS})]) | ||
| assert hs.property_names("deals") == ("amount", "my_custom") | ||
| assert hs.property_names("deals") == ("amount", "uncurated_std", "my_custom") | ||
|
|
||
| def test_custom_property_names(self): | ||
| hs = make_client([FakeResponse({"results": self.DESCRIPTORS})]) | ||
| assert hs.custom_property_names("deals") == frozenset({"my_custom"}) | ||
|
|
||
| def test_unknown_object_has_no_curated_names(self): | ||
| def test_property_names_independent_of_allowlist(self): | ||
| hs = make_client([FakeResponse({"results": [prop("anything")]})]) | ||
| assert hs.property_names("unknown_object") == () | ||
| assert hs.property_names("unknown_object") == ("anything",) | ||
|
|
||
|
|
||
| class TestGenerateSchema: | ||
| def test_curated_props_added_with_string_type(self): | ||
| hs = make_client( | ||
| [ | ||
| FakeResponse( | ||
| { | ||
| "results": [ | ||
| prop("amount", type_="number"), | ||
| prop("uncurated_std"), | ||
| prop("my_custom", hubspot_defined=False), | ||
| ] | ||
| } | ||
| ) | ||
| ] | ||
| ) | ||
| schema = hs.generate_schema("deals") | ||
| props = schema["properties"] | ||
| # Base record fields always present. | ||
| def test_allowlist_is_the_whole_property_column_set(self): | ||
| props = make_client().generate_schema("deals")["properties"] | ||
| expected = {f"properties_{name}" for name in ALLOWED_PROPERTIES_BY_OBJECT["deals"]} | ||
| assert {k for k in props if k.startswith("properties_")} == expected | ||
| assert all(props[k] == {"type": ["string", "null"]} for k in expected) | ||
|
|
||
| def test_schema_needs_no_portal_describe(self): | ||
| hs = make_client() # no queued responses: any HTTP call raises | ||
| hs.generate_schema("deals") | ||
| assert hs._http_client.calls == [] | ||
|
|
||
| def test_base_record_fields_always_present(self): | ||
| props = make_client().generate_schema("leads")["properties"] | ||
| assert props["id"] == {"type": ["string", "null"]} | ||
| assert props["archived"] == {"type": ["boolean", "null"]} | ||
| assert props["archivedAt"]["format"] == "date-time" | ||
| # number maps to string on purpose (Bronze stays Nullable(String)). | ||
| assert props["properties_amount"] == {"type": ["string", "null"]} | ||
| assert "properties_uncurated_std" not in props | ||
| assert "properties_my_custom" not in props # customs ride in custom_fields | ||
|
|
||
| def test_unknown_type_warns_once(self, caplog): | ||
| hs = make_client( | ||
| [ | ||
| FakeResponse( | ||
| { | ||
| "results": [ | ||
| {"name": "amount", "hubspotDefined": True, "type": "alien"}, | ||
| {"name": "closedate", "hubspotDefined": True, "type": "alien"}, | ||
| ] | ||
| } | ||
| ) | ||
| ] | ||
| ) | ||
| with caplog.at_level(logging.WARNING, logger="airbyte"): | ||
| schema = hs.generate_schema("deals") | ||
| assert schema["properties"]["properties_amount"] == {"type": ["string", "null"]} | ||
| assert caplog.text.count("Unknown HubSpot property type") == 1 | ||
|
|
||
| def test_prop_to_json_schema_format_passthrough(self, monkeypatch): | ||
| # No current mapping carries a format — patch one in to cover the | ||
| # format branch. | ||
| monkeypatch.setitem(api_mod.HUBSPOT_TYPE_TO_JSON_SCHEMA, "datetime", ("string", "date-time")) | ||
| out = _prop_to_json_schema({"name": "x", "type": "datetime"}, set()) | ||
| assert out == {"type": ["string", "null"], "format": "date-time"} | ||
|
|
||
| def test_prop_to_json_schema_defaults_missing_type_to_string(self): | ||
| assert _prop_to_json_schema({"name": "x"}, set()) == {"type": ["string", "null"]} | ||
|
|
||
| @pytest.mark.parametrize("object_type", ["leads", "unknown_object"]) | ||
| def test_empty_allowlist_yields_base_fields_only(self, object_type): | ||
| props = make_client().generate_schema(object_type)["properties"] | ||
| assert not [k for k in props if k.startswith("properties_")], f"object: {object_type}" |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Apply the signature-typing rule to both changed Python files.
src/ingestion/connectors/crm/hubspot/tests/test_api.py#L126-L156: add-> Noneto each changed test method and annotateobject_type: str.src/ingestion/connectors/crm/hubspot/tests/conftest.py#L98-L99: add-> NonetoFakeHubspot.__init__.
As per coding guidelines, src/ingestion/**/*.{py,pyi} requires type hints on every function and method signature.
📍 Affects 2 files
src/ingestion/connectors/crm/hubspot/tests/test_api.py#L126-L156(this comment)src/ingestion/connectors/crm/hubspot/tests/conftest.py#L98-L99
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/connectors/crm/hubspot/tests/test_api.py` around lines 126 -
156, Add return type annotations of None to every changed test method in
test_api.py, including the parameterized test’s object_type parameter annotated
as str. Also annotate FakeHubspot.__init__ in conftest.py with -> None; make no
other changes.
Source: Coding guidelines
| "BillingAddress": { | ||
| "type": [ | ||
| "string", | ||
| "null" | ||
| ] | ||
| }, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Check compound-value coercion and find every address-typed property in the static schemas.
set -euo pipefail
fd -t f 'envelope.py' src/ingestion/connectors/crm/salesforce --exec cat -n {}
# Every property whose name ends in Address across the static schemas.
fd -t f -e json . src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas \
--exec rg -n -A4 '"[A-Za-z]*Address":' {}Repository: constructorfabric/insight
Length of output: 13548
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Candidate files:\n'
fd -t f -i 'schema|describe|salesforce|source.py' src ing* 2>/dev/null | sed -n '1,120p'
printf '\nKey symbol search:\n'
rg -n --glob '!**/stream_schemas/*.json' 'schema|schema_generator|describe|address|Address|queryAll|connector|stream' src/ingestion/connectors/crm/salesforce source* 2>/dev/null | sed -n '1,240p'Repository: constructorfabric/insight
Length of output: 38401
🏁 Script executed:
#!/bin/bash
set -euo pipefail
source_dir="src/ingestion/connectors/crm/salesforce"
printf 'Files under Salesforce connector:\n'
git ls-files "$source_dir" | sed -n '1,200p'
printf '\nPackage metadata:\n'
for f in pyproject.toml setup.py setup.cfg requirements*.txt; do
git ls-files | grep -E "(^$f$|/$f$)" || true
done
fd -t f '(^pyproject.toml$|^setup.py$|^setup.cfg$|^requirements.*\.txt$)' . | sed -n '1,200p'Repository: constructorfabric/insight
Length of output: 4097
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'schema_loader.py outline and relevant contents:\n'
ast-grep outline src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py --view expanded || true
cat -n src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py | sed -n '1,260p'
printf '\nconftest.py contents relevant to Salesforce fields:\n'
cat -n src/ingestion/connectors/crm/salesforce/tests/conftest.py | sed -n '1,260p'Repository: constructorfabric/insight
Length of output: 8152
Apply the Salesforce compound address mapping consistently. queryAll returns these address fields as nested JSON objects, while envelope() preserves compound values in raw_data without stringifying them. Update Account.BillingAddress, Account.ShippingAddress, Lead.Address, and Contact.MailingAddress / Contact.OtherAddress to match the shared address type, or remove them from the declared columns if the connector only exposes them via raw_data.
📍 Affects 3 files
src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Account.schema.json#L105-L110(this comment)src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Contact.schema.json#L117-L122src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Lead.schema.json#L123-L128
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/Account.schema.json`
around lines 105 - 110, The Salesforce compound address fields are declared as
strings but are returned as nested objects. Update Account.schema.json lines
105-110, Contact.schema.json lines 117-122, and Lead.schema.json lines 123-128
for BillingAddress, ShippingAddress, Address, MailingAddress, and OtherAddress
to use the shared compound address type consistently, or remove these declared
columns if they are exposed only through raw_data.
| "Address": { | ||
| "type": [ | ||
| "string", | ||
| "null" | ||
| ] | ||
| }, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Find compound address/location fields declared as strings in the static Salesforce schemas.
set -euo pipefail
fd -e json . src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas --exec \
jq -r --arg f '{}' '
.properties
| to_entries[]
| select(.key | test("^(Address|BillingAddress|ShippingAddress|MailingAddress|OtherAddress|Location)$"))
| "\($f): \(.key) -> \(.value.type | tostring)"
' {}Repository: constructorfabric/insight
Length of output: 1044
🌐 Web query:
airbyte-cdk TypeTransformer DefaultSchemaNormalization object value declared string coercion behavior
💡 Result:
In the Airbyte Python CDK, DefaultSchemaNormalization is a transformation configuration applied via the Transformer class to automatically mutate record data to align with a stream's declared JSON schema [1][2]. When DefaultSchemaNormalization is enabled for a stream, the CDK attempts to cast incoming data values to match the types defined in the stream's schema [1][2]. Regarding the behavior for specific types: - Automatic Casting: If the source API returns a value that does not match the schema type (e.g., a non-string value for a field declared as a "string"), the CDK automatically casts it to the correct type [1][2]. - Object Value Behavior: The DefaultSchemaNormalization supports transformation of complex types, including nested objects and arrays [1][2]. If a field is defined as an "object" in the schema, the transformer maintains the structure while ensuring the nested fields align with the declared schema [1]. - Failure to Cast: If a value cannot be successfully cast to the target schema type (for example, attempting to cast a non-numeric string like "asdf" to an integer), the field will retain its original value rather than being coerced or set to null [1][2]. - Limitations: Default normalization does not apply to types defined as an array of multiple types (except for "null") or fields governed by oneOf or anyOf keywords [1][2]. To use this feature, you configure it within your stream class by setting the transformer property: transformer = Transformer(TransformConfig.DefaultSchemaNormalization) This transformation is disabled by default and must be explicitly enabled on a per-stream basis [1][2]. If both default and custom transformations are needed, they can be combined using bitwise OR (e.g., TransformConfig.DefaultSchemaNormalization | TransformConfig.CustomSchemaNormalization), in which case the default normalization is always executed before the custom transformation [1][2].
Citations:
- 1: https://docs.airbyte.com/platform/connector-development/cdk-python/schemas
- 2: https://github.com/airbytehq/airbyte/blob/master/docs/platform/connector-development/cdk-python/schemas.md
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "## Candidate files"
fd -e json User.schema.json src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas || true
fd -e py 'source\.py$' src/ingestion/connectors/crm/salesforce/source_salesforce || true
echo
echo "## User schema address context"
file="$(fd -e json User.schema.json src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas | head -n 1)"
if [ -n "${file:-}" ]; then
sed -n '90,145p' "$file" | cat -n
fi
echo
echo "## Salesforce source schema normalization references"
rg -n "DefaultSchemaNormalization|AddressTransform|TypeTransformer|billingAddress|shippingAddress|mailingAddress|otherAddress|location|address" src/ingestion/connectors/crm/salesforce/source_salesforce -S || trueRepository: constructorfabric/insight
Length of output: 1933
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "## source.py outline"
ast-grep outline src/ingestion/connectors/crm/salesforce/source_salesforce/source.py --view compact || true
echo
echo "## source.py relevant imports and stream declarations"
rg -n "class .*\(.*Source|Salesforce|streams\s*=|stream_schemas|TypeTransformer|DefaultSchemaNormalization|User|Account|Contact|Lead|Event" src/ingestion/connectors/crm/salesforce/source_salesforce -S
echo
echo "## User stream/schema references"
rg -n "\bUser\b|stream_schemas/User|stream_schema|User\.schema" src/ingestion/connectors/crm/salesforce/source_salesforce -SRepository: constructorfabric/insight
Length of output: 36228
🌐 Web query:
Salesforce User object Address field compound field SOQL returns address object Street City
💡 Result:
In Salesforce, the Address field on the User object is a compound field [1][2]. While you can select the Address compound field directly in a SOQL SELECT clause using API version 30.0 or later, it is subject to several important limitations [2][3][4]. Key behaviors for querying the User Address compound field: 1. SELECT Clause Support: You can retrieve the Address field directly in a SOQL query (e.g., SELECT Address FROM User) [2][4]. The returned value is a structured data type (an Address object) that encapsulates components such as Street, City, State, PostalCode, Country, and Geolocation [2][5]. 2. Limitations in WHERE Clauses: You cannot use the compound Address field in a SOQL WHERE clause [3][6]. To filter records based on address data, you must query the individual component fields (e.g., WHERE Street!= null OR City!= null) [7]. 3. API and Tooling Restrictions: Compound fields are generally accessible via the SOAP and REST APIs [2][4]. However, they are not supported by the Bulk API [8]. If you are working with tools or integrations that rely on the Bulk API, you must explicitly query the individual component fields (Street, City, etc.) instead of the compound Address field [8]. 4. Read-Only Behavior: The Address compound field is read-only [1][4]. Updates to address information must be performed by writing to the individual component fields (e.g., Street, City, State, etc.) rather than the compound field itself [3][4]. If your application requires backward compatibility with API versions earlier than 30.0, or if you need to support environments where compound fields are not supported (like the Bulk API), you should always write your queries to select and filter by the individual component fields [2][8].
Citations:
- 1: https://developer.salesforce.com/docs/atlas.en-us.object_reference.meta/api/sforce_api_objects_user.htm
- 2: https://developer.salesforce.com/docs/atlas.en-us.object_reference.meta/object_reference/compound_fields_address.htm
- 3: https://developer.salesforce.com/docs/atlas.en-us.object_reference.meta/object_reference/compound_fields_limitations.htm
- 4: https://developer.salesforce.com/docs/atlas.en-us.object_reference.meta/object_reference/compound_fields.htm
- 5: https://developer.salesforce.com/docs/atlas.en-us.sfFieldRef.meta/sfFieldRef/salesforce_field_reference_Address.htm
- 6: https://developer.sf.tumee.com/docs/atlas.en-us.object_reference.meta/object_reference/compound_fields_limitations.htm
- 7: https://studyx.ai/questions/4mkildl/how-to-query-address-field-not-empty-in-user-object-in-salesforce
- 8: https://salesforce.stackexchange.com/questions/136844/get-all-salesforces-users
🌐 Web query:
Airbyte Python CDK TypeTransformer DefaultSchemaNormalization string type union array oneOf cast behavior source record values
💡 Result:
In the Airbyte Python CDK, the DefaultSchemaNormalization transformation (enabled via the TypeTransformer) performs automatic type casting to align record values with the stream's declared JSON schema [1][2]. However, this normalization mechanism has specific limitations regarding complex types [1][2]. Types described as an array of multiple types (except "null") and types defined under oneOf or anyOf keywords are explicitly excluded from this automatic transformation [1][2]. Consequently, if a field's schema utilizes oneOf or union types to allow for multiple possible data structures or types, the DefaultSchemaNormalization will not attempt to cast or mutate those values [1][2]. The source record values for these fields will remain in their original state [1][2]. If you require specific transformation logic for these complex types, you must implement a custom schema type transformation by using TransformConfig.CustomSchemaNormalization and registering a custom transformation function [1][2].
Citations:
- 1: https://docs.airbyte.com/platform/connector-development/cdk-python/schemas
- 2: https://github.com/airbytehq/airbyte/blob/master/docs/platform/connector-development/cdk-python/schemas.md
Declare Address as an object, or remove it.
User.Address is a Salesforce compound object, so ["string","null"] does not match the API shape. Default schema normalization does not transform non-string values back into strings for fields declared as a union type, so raw_data keeps the object but Address remains mismatched. Remove Address to avoid a misleading column, or declare it as ["object", "null"] if the column must stay.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/ingestion/connectors/crm/salesforce/source_salesforce/stream_schemas/User.schema.json`
around lines 123 - 128, Update the User schema’s Address property to match
Salesforce’s compound-object shape by changing its type union from string/null
to object/null, or remove the Address property if it should not be exposed as a
column; do not leave it declared as a string.
| for record in super().read_records(sync_mode, cursor_field, stream_slice, stream_state): | ||
| if isinstance(record, Mapping): | ||
| yield envelope( | ||
| record, | ||
| tenant_id=self._tenant_id, | ||
| source_id=self._source_id, | ||
| custom_field_names=self._custom_field_names, | ||
| declared_fields=self.declared_fields, | ||
| collision_seen=self._envelope_collisions_seen, | ||
| ) | ||
| else: | ||
| # State / log / trace messages pass through untouched. | ||
| yield record | ||
|
|
||
| def _sf_properties(self) -> Mapping[str, Any]: | ||
| @property | ||
| def declared_fields(self) -> FrozenSet[str]: | ||
| """SF fields this stream emits as Bronze columns; the rest ride in ``raw_data``.""" | ||
| return declared_field_names(self.name) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Find get_json_schema call sites in the installed airbyte-cdk read path.
set -euo pipefail
fd -t d '^airbyte_cdk$' / 2>/dev/null | head -1 | while read -r cdk; do
rg -n -C3 'get_json_schema\s*\(' "$cdk/sources" | head -60
doneRepository: constructorfabric/insight
Length of output: 163
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Locate relevant files =="
fd -a 'streams.py|schema_loader.py|pyproject.toml' . | sed 's#^\./##' | sort
echo
echo "== Current streams.py context =="
if [ -f src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py ]; then
nl -ba src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py | sed -n '1,180p'
fi
echo
echo "== Current schema_loader.py context =="
if [ -f src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py ]; then
nl -ba src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py | sed -n '1,220p'
fi
echo
echo "== Dependency references =="
for f in pyproject.toml src/ingestion/connectors/crm/salesforce source_salesforce; do
[ -f "$f" ] || continue
echo "--- $f ---"
rg -n 'airbyte|cdc|salesforce|source_salesforce|requirements|uv|pdm|poetry' "$f" || true
done
echo
echo "== Schema property counts for User and related files =="
fd 'User\.schema\.json|schema.*json' src/ingestion/connectors/crm/salesforce/source_salesforce -t f -x sh -c 'echo "--- $1 ---"; python3 - "$1" <<"PY"
import json, pathlib
p=pathlib.Path(sys.argv[1])
data=json.loads(p.read_text())
if "properties" in data:
print("properties:", len(data["properties"]))
for k in data.get("required", []):
print(k)
PY
' sh {} \; | sed -n '1,220p'Repository: constructorfabric/insight
Length of output: 1494
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path("src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py")
print(f"streams.py exists: {p.exists()}")
if p.exists():
for i, line in enumerate(p.read_text().splitlines(), 1):
if 1 <= i <= 180:
print(f"{i:4d}\t{line}")
PY
echo
python3 - <<'PY'
from pathlib import Path
p = Path("src/ingestion/connectors/crm/salesforce/source_salesforce/schema_loader.py")
print(f"schema_loader.py exists: {p.exists()}")
if p.exists():
for i, line in enumerate(p.read_text().splitlines(), 1):
if 1 <= i <= 260:
print(f"{i:4d}\t{line}")
PY
echo
echo "== Salesforce connector dependency references =="
for f in pyproject.toml src/ingestion/connectors/crm/salesforce/pyproject.toml src/ingestion/connectors/crm/salesforce/source_salesforce; do
[ -f "$f" ] || continue
echo "--- $f ---"
rg -n 'airbyte|cdc|salesforce|source_salesforce|requirements|uv|pdm|poetry' "$f" || true
done
echo
echo "== Schema property counts for relevant schema files =="
python3 - <<'PY'
from pathlib import Path
import json
for p in Path("src/ingestion/connectors/crm/salesforce/source_salesforce").rglob("*.schema.json"):
data = json.loads(p.read_text())
props = data.get("properties", {})
print(f"{p}: {len(props)}")
PYRepository: constructorfabric/insight
Length of output: 11375
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import subprocess
for cmd in [
["uv", "pip", "show", "airbyte-cdk"],
["python", "-m", "pip", "show", "airbyte-cdk"],
["pip", "show", "airbyte-cdk"],
]:
result = subprocess.run(cmd, capture_output=True, text=True)
print("CMD"," ".join(cmd))
print("rc", result.returncode)
print(result.stdout.splitlines()[:12] or result.stderr.splitlines()[:12])
print()
PYRepository: constructorfabric/insight
Length of output: 679
🌐 Web query:
airbyte-cdk source_streams.py read_records get_json_schema call path
💡 Result:
In the Airbyte Python CDK, the get_json_schema() method is the primary mechanism for defining a stream's schema [1][2]. It is not strictly part of the read_records call path for fetching data, but rather a configuration method used by the Airbyte platform to understand the data structure of the stream before and during the synchronization process [1][2][3]. The call path for read_records and get_json_schema can be summarized as follows: 1. Initialization and Catalog Processing: When an Airbyte sync starts, the source reads a ConfiguredAirbyteCatalog [3]. The Airbyte platform uses the get_json_schema() method of each Stream instance to validate or determine the schema for the requested streams [1][4]. 2. Stream Definition: By default, get_json_schema() looks for a JSON file in the schemas/ directory matching the snake-cased name of the stream class (e.g., schemas/employee_benefits.json for a class named EmployeeBenefits) [1][2]. Developers can override this method to return a dictionary directly if they need to generate the schema dynamically [2][5]. 3. The read_records Execution: The read_records method is responsible for fetching and emitting data [1][6]. It is distinct from get_json_schema(). During a sync, the source class iterates through the catalog, identifies the stream instance, and invokes read_records to perform the actual data extraction (e.g., hitting an API endpoint) [6][3]. 4. Interaction: While read_records typically does not call get_json_schema() itself to fetch data, custom implementations may sometimes invoke get_json_schema() within internal logic to inspect properties for filtering or building requests [4]. However, these are implementation-specific uses rather than a required part of the standard CDK data-reading call path [1][4]. In summary, get_json_schema() informs the system about the data structure, while read_records() implements the logic to stream the data, and they function as independent components of the Stream lifecycle [1][6][7].
Citations:
- 1: https://docs.airbyte.com/platform/connector-development/cdk-python/full-refresh-stream
- 2: https://docs.airbyte.com/platform/connector-development/cdk-python/schemas
- 3: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/abstract_source.py
- 4: feat(airbyte-cdk): add
json_schemafromConfiguredCatalogtoStreamairbytehq/airbyte#39522 - 5: https://github.com/airbytehq/airbyte/blob/master/docs/platform/connector-development/cdk-python/schemas.md
- 6: Understanding the Airbyte CDK #3: `read` command airbytehq/airbyte#33814
- 7: https://github.com/airbytehq/airbyte/blob/master/docs/platform/connector-development/cdk-python/full-refresh-stream.md
Cache declared_fields before iterating records.
read_records calls self.declared_fields for each JSON record; this performs declared_field_names(self.name), which currently rebuilds the frozenset over _load(stream_name)["properties"] every time. User has 185 schema properties, so hoist self.declared_fields outside the loop and cache declared_field_names() in schema_loader.py.
get_json_schema() is part of catalog/schema discovery, not the per-record read path, and already calls _load(stream_name) once before deepcopy/injection.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py` around
lines 113 - 129, Cache the declared field set before the record loop in
read_records and reuse it when calling envelope, rather than resolving
self.declared_fields for each record. Update declared_field_names in
schema_loader.py to cache the frozenset derived from
_load(stream_name)["properties"], while leaving get_json_schema’s existing
single _load call unchanged.
| def test_first_use_authenticates(self, sf): | ||
| sf.login = Mock(side_effect=lambda: setattr(sf, "access_token", "tok")) | ||
| assert sf._token_provider.get_token() == "tok" | ||
| sf.login.assert_called_once() | ||
|
|
||
| def test_first_login_failure_surfaces(self, sf): | ||
| sf.login = Mock(side_effect=RequestException("no creds")) | ||
| with pytest.raises(RequestException): | ||
| sf._token_provider.get_token() | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add type hints to the changed test definitions.
The changed test methods need typed fixture parameters and -> None. _sf_stub needs a -> Salesforce return annotation.
src/ingestion/connectors/crm/salesforce/tests/test_api.py#L35-L44: add fixture parameter types and-> None.src/ingestion/connectors/crm/salesforce/tests/test_api.py#L184-L190: add fixture parameter types and-> None.src/ingestion/connectors/crm/salesforce/tests/test_api.py#L213-L238: add fixture parameter types and-> None.src/ingestion/connectors/crm/salesforce/tests/test_api.py#L261-L277: add fixture parameter types and-> None.src/ingestion/connectors/crm/salesforce/tests/test_source.py#L201-L208: add fixture parameter types and-> None.src/ingestion/connectors/crm/salesforce/tests/test_source.py#L233-L255: add-> Noneto the test method and-> Salesforceto_sf_stub.src/ingestion/connectors/crm/salesforce/tests/test_source.py#L270-L274: add fixture parameter types and-> None.src/ingestion/connectors/crm/salesforce/tests/test_source.py#L332-L340: add fixture parameter types and-> None.
As per coding guidelines, "src/ingestion/**/*.{py,pyi}: Put type hints on every function and method signature; do not allow bare Any to escape."
📍 Affects 2 files
src/ingestion/connectors/crm/salesforce/tests/test_api.py#L35-L44(this comment)src/ingestion/connectors/crm/salesforce/tests/test_api.py#L184-L190src/ingestion/connectors/crm/salesforce/tests/test_api.py#L213-L238src/ingestion/connectors/crm/salesforce/tests/test_api.py#L261-L277src/ingestion/connectors/crm/salesforce/tests/test_source.py#L201-L208src/ingestion/connectors/crm/salesforce/tests/test_source.py#L233-L255src/ingestion/connectors/crm/salesforce/tests/test_source.py#L270-L274src/ingestion/connectors/crm/salesforce/tests/test_source.py#L332-L340
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/connectors/crm/salesforce/tests/test_api.py` around lines 35 -
44, Update every changed test method in
src/ingestion/connectors/crm/salesforce/tests/test_api.py ranges 35-44, 184-190,
213-238, and 261-277, and
src/ingestion/connectors/crm/salesforce/tests/test_source.py ranges 201-208,
233-255, 270-274, and 332-340, adding appropriate fixture parameter types and ->
None return annotations. In test_source.py range 233-255, also annotate _sf_stub
with -> Salesforce; ensure all affected function and method signatures are fully
typed without bare Any escaping.
Source: Coding guidelines
| The full cycle — throwaway ClickHouse, fresh `.env`, bootstrap, snapshot re-dump, field-parity audit, cleanup — as a single copy-paste. No credentials needed: every connector discovers on fake config values. **Overwrites `.env`** next to the scripts. | ||
|
|
||
| ```bash | ||
| cd src/ingestion/scripts/bootstrap-db | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the fork pull request documentation.
The workflow now runs validation for fork pull requests without CRM secrets. Line 132 still says that fork pull requests are skipped because the lane needs repository secrets. Update Line 132 to state that forks receive validation and the regenerated snapshot artifact, while regen-pr remains same-repository only.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/scripts/bootstrap-db/README.md` around lines 86 - 90, Update
the fork pull request documentation near the workflow description to state that
fork PRs run validation and receive the regenerated snapshot artifact without
CRM secrets, while limiting regen-pr to same-repository pull requests.
| ALTER TABLE silver.class_crm_accounts DROP COLUMN IF EXISTS custom_fields; | ||
|
|
||
| ALTER TABLE silver.class_crm_activities DROP COLUMN IF EXISTS custom_fields; | ||
|
|
||
| ALTER TABLE silver.class_crm_contacts DROP COLUMN IF EXISTS custom_fields; | ||
|
|
||
| ALTER TABLE silver.class_crm_deals DROP COLUMN IF EXISTS custom_fields; | ||
|
|
||
| ALTER TABLE silver.class_crm_users DROP COLUMN IF EXISTS custom_fields; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Migrate existing bronze tables to raw_data.
CREATE TABLE IF NOT EXISTS does not modify existing bronze tables. This deploy path drops custom_fields only from staging and silver tables. Existing bronze_hubspot.* and bronze_salesforce.* tables retain custom_fields and lack raw_data.
Before connectors emit the new envelope, alter every existing bronze table to drop custom_fields and add raw_data after collected_at. This preserves the required positional column order.
src/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql#L12-L20: add the bronze schema upgrade to the deploy migration path.src/ingestion/scripts/apply-ch-migrations.sh#L102-L114: include HubSpot and Salesforce bronze relations in contract healing.src/ingestion/scripts/connectors-ddl/hubspot.sql#L31-L619: keep these declarations as the target schema and provide matchingALTER TABLEstatements for existing tables.src/ingestion/scripts/connectors-ddl/salesforce.sql#L77-L736: keep these declarations as the target schema and provide matchingALTER TABLEstatements for existing tables.
🧰 Tools
🪛 Squawk (2.61.0)
[warning] 12-12: Dropping a column may break existing clients.
(ban-drop-column)
[warning] 14-14: Dropping a column may break existing clients.
(ban-drop-column)
[warning] 16-16: Dropping a column may break existing clients.
(ban-drop-column)
[warning] 18-18: Dropping a column may break existing clients.
(ban-drop-column)
[warning] 20-20: Dropping a column may break existing clients.
(ban-drop-column)
📍 Affects 4 files
src/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql#L12-L20(this comment)src/ingestion/scripts/apply-ch-migrations.sh#L102-L114src/ingestion/scripts/connectors-ddl/hubspot.sql#L31-L619src/ingestion/scripts/connectors-ddl/salesforce.sql#L77-L736
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql`
around lines 12 - 20, Migrate existing bronze tables from custom_fields to
raw_data while preserving raw_data immediately after collected_at. In
src/ingestion/scripts/migrations/20260805000000_crm-drop-custom-fields.sql
(lines 12-20), add ALTER statements for every bronze_hubspot and
bronze_salesforce relation; in src/ingestion/scripts/apply-ch-migrations.sh
(lines 102-114), include those relations in contract healing. Keep the target
declarations in src/ingestion/scripts/connectors-ddl/hubspot.sql (lines 31-619)
and src/ingestion/scripts/connectors-ddl/salesforce.sql (lines 77-736), adding
matching existing-table ALTER statements there.
3 participants
Problem
Salesforce and HubSpot built their advertised schemas from a live
describe. Two costs: bronze DDL could not be generated without credentials, and column sets varied per org/portal. HubSpot also dropped data — standard properties outside the allowlist were never requested.Fix
Schemas are repository artifacts.
discoverissues no API call for either connector.source_salesforce/stream_schemas/, frozen from the committed DDL so dbt reads nothing new. Auth happens on demand, not atstreams(). Stream availability is resolved per read, off the describe the stream already makes for its SOQL field list; a sobject the org lacks syncs empty andcheckreports it.raw_datacarries the whole record (per-value 2 KB cap).custom_fieldsremoved as a strict subset of it.bootstrap-dband theconnectors-ddlgate need no credentials; the fork skip is gone from the gate and now guards onlyregen-pr.Verify
discoveron fake config with the network blocked: SF 10 streams, HubSpot 19.bootstrap-dbon a throwaway ClickHouse: 218/218 dbt models, field parity 0 failures. Snapshot regenerated from that run.custom_fields→raw_data, so the frozen schemas reproduce the describe-derived ones.Notes
descriptor.yaml, andbump-descriptorsonly runs on push tomain. Until the new images publish, this PR'sconnectors-ddlrun uses the old describe-driven images and will fail on auth. Expected; it resolves onmainonce images land.custom_fieldsleaves the class contract: migration drops it fromsilver.class_crm_*,apply-ch-migrations.shfrom staging. Nothing read its content.bitbucket-cloud.sqlyields 10 tables against the committed 20 (pinned image exposes fewer streams, no config gates them). Left untouched here; worth checking onmain.Refs #756 #757
Summary by CodeRabbit
New Features
raw_datafield while exposing only declared fields as columns.Bug Fixes
custom_fieldscolumns from CRM data models and added migration support for existing tables.Documentation
raw_datahandling.