jprdonnelly/airbyte
1
0
Fork 0
mirror of synced 2025-12-25 02:09:19 -05:00
Code Issues Packages Projects Releases Wiki Activity

🚨🚨Source Hubspot: add property history streams for Deals and Companies (#33266)

Browse Source
This commit is contained in:
Roman Yermilov [GL]
2023-12-12 22:29:43 +01:00
committed by GitHub
parent 44ce0022d2
commit 8bd60a946e
17 changed files with 622 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
airbyte-integrations/connectors/source-hubspot/acceptance-test-config.yml
View File

@@ -20,7 +20,7 @@ acceptance_tests:
tests:
- config_path: secrets/config_oauth.json
backward_compatibility_tests_config:
disable_for_version: 1.4.1
disable_for_version: 1.9.0
basic_read:
tests:
- config_path: secrets/config_oauth.json

16
airbyte-integrations/connectors/source-hubspot/integration_tests/abnormal_state.json
View File

@@ -203,7 +203,21 @@
{
"type": "STREAM",
"stream": {
"stream_descriptor": { "name": "property_history" },
"stream_descriptor": { "name": "contacts_property_history" },
"stream_state": { "updatedAt": 7945393076000 }
}
},
{
"type": "STREAM",
"stream": {
"stream_descriptor": { "name": "companies_property_history" },
"stream_state": { "updatedAt": 7945393076000 }
}
},
{
"type": "STREAM",
"stream": {
"stream_descriptor": { "name": "deals_property_history" },
"stream_state": { "updatedAt": 7945393076000 }
}
},

13
airbyte-integrations/connectors/source-hubspot/integration_tests/expected_records.jsonl
View File

File diff suppressed because one or more lines are too long

9
airbyte-integrations/connectors/source-hubspot/metadata.yaml
View File

@@ -10,7 +10,7 @@ data:
connectorSubtype: api
connectorType: source
definitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
dockerImageTag: 1.9.0
dockerImageTag: 2.0.0
dockerRepository: airbyte/source-hubspot
documentationUrl: https://docs.airbyte.com/integrations/sources/hubspot
githubIssueLabel: source-hubspot
@@ -23,6 +23,13 @@ data:
oss:
enabled: true
releaseStage: generally_available
releases:
breakingChanges:
2.0.0:
message: >-
This version eliminates the Property History stream in favor of creating 3 different streams, Contacts, Companies, and Deals, which can now all fetch their property history.
It will affect only users who use Property History stream, who will need to fix schema conflicts and sync Contacts Property History stream instead of Property History.
upgradeDeadline: 2023-12-21
suggestedStreams:
streams:
- contacts

24
airbyte-integrations/connectors/source-hubspot/sample_files/basic_read_catalog.json
View File

@@ -245,7 +245,29 @@
},
{
"stream": {
"name": "property_history",
"name": "contacts_property_history",
"json_schema": {},
"supported_sync_modes": ["full_refresh", "incremental"],
"default_cursor_field": ["timestamp"]
},
"sync_mode": "full_refresh",
"cursor_field": ["timestamp"],
"destination_sync_mode": "overwrite"
},
{
"stream": {
"name": "companies_property_history",
"json_schema": {},
"supported_sync_modes": ["full_refresh", "incremental"],
"default_cursor_field": ["timestamp"]
},
"sync_mode": "full_refresh",
"cursor_field": ["timestamp"],
"destination_sync_mode": "overwrite"
},
{
"stream": {
"name": "deals_property_history",
"json_schema": {},
"supported_sync_modes": ["full_refresh", "incremental"],
"default_cursor_field": ["timestamp"]

26
airbyte-integrations/connectors/source-hubspot/sample_files/incremental_catalog.json
View File

@@ -278,7 +278,31 @@
},
{
"stream": {
"name": "property_history",
"name": "contacts_property_history",
"json_schema": {},
"supported_sync_modes": ["full_refresh", "incremental"],
"source_defined_cursor": true,
"default_cursor_field": ["timestamp"]
},
"sync_mode": "incremental",
"cursor_field": ["timestamp"],
"destination_sync_mode": "append"
},
{
"stream": {
"name": "companies_property_history",
"json_schema": {},
"supported_sync_modes": ["full_refresh", "incremental"],
"source_defined_cursor": true,
"default_cursor_field": ["timestamp"]
},
"sync_mode": "incremental",
"cursor_field": ["timestamp"],
"destination_sync_mode": "append"
},
{
"stream": {
"name": "deals_property_history",
"json_schema": {},
"supported_sync_modes": ["full_refresh", "incremental"],
"source_defined_cursor": true,

16
airbyte-integrations/connectors/source-hubspot/sample_files/sample_state.json
View File

@@ -87,7 +87,21 @@
"type": "STREAM",
"stream": {
"stream_state": { "timestamp": 1700681340514 },
"stream_descriptor": { "name": "property_history" }
"stream_descriptor": { "name": "contacts_property_history" }
}
},
{
"type": "STREAM",
"stream": {
"stream_state": { "timestamp": 1700681340514 },
"stream_descriptor": { "name": "companies_property_history" }
}
},
{
"type": "STREAM",
"stream": {
"stream_state": { "timestamp": 1700681340514 },
"stream_descriptor": { "name": "deals_property_history" }
}
},
{

21
airbyte-integrations/connectors/source-hubspot/source_hubspot/helpers.py
View File

@@ -70,6 +70,11 @@ class IURLPropertyRepresentation(abc.ABC):
def as_url_param(self):
""""""
@property
@abc.abstractmethod
def as_url_param_with_history(self) -> str:
""""""
@property
@abc.abstractmethod
def _term_representation(self):
@@ -105,9 +110,25 @@ class APIv1Property(IURLPropertyRepresentation):
def as_url_param(self):
return {"property": self.properties}
def as_url_param_with_history(self) -> str:
return "&".join(map(lambda prop: f"propertiesWithHistory={prop}", self.properties))
class APIv2Property(IURLPropertyRepresentation):
_term_representation = "property={property}&"
def as_url_param(self):
return {"property": self.properties}
def as_url_param_with_history(self) -> str:
return "&".join(map(lambda prop: f"propertiesWithHistory={prop}", self.properties))
class APIv3Property(IURLPropertyRepresentation):
_term_representation = "{property},"
def as_url_param(self):
return {"properties": ",".join(self.properties)}
def as_url_param_with_history(self) -> str:
raise NotImplementedError("Not implemented")

52
airbyte-integrations/connectors/source-hubspot/source_hubspot/schemas/companies_property_history.json Normal file
View File

@@ -0,0 +1,52 @@
{
"$schema": "http://json-schema.org/draft-07/schema",
"type": ["null", "object"],
"additionalProperties": true,
"properties": {
"updatedByUserId": {
"type": ["null", "number"]
},
"requestId": {
"type": ["null", "string"]
},
"source": {
"type": ["null", "string"]
},
"portalId": {
"type": ["null", "number"]
},
"isDeleted": {
"type": ["null", "boolean"]
},
"timestamp": {
"type": ["null", "number"]
},
"property": {
"type": ["null", "string"]
},
"persistenceTimestamp": {
"type": ["null", "number"]
},
"name": {
"type": ["null", "string"]
},
"sourceVid": {
"type": ["null", "array"]
},
"useTimestampAsPersistenceTimestamp": {
"type": ["null", "boolean"]
},
"sourceMetadata": {
"type": ["null", "string"]
},
"companyId": {
"type": ["null", "number"]
},
"sourceId": {
"type": ["null", "string"]
},
"value": {
"type": ["null", "string"]
}
}
}

10
airbyte-integrations/connectors/source-hubspot/source_hubspot/schemas/property_history.json → airbyte-integrations/connectors/source-hubspot/source_hubspot/schemas/contacts_property_history.json
View File

@@ -1,6 +1,7 @@
{
"$schema": "http://json-schema.org/draft-07/schema",
"type": ["null", "object"],
"additionalProperties": true,
"properties": {
"value": {
"type": ["null", "string"]
@@ -23,12 +24,21 @@
"selected": {
"type": ["null", "boolean"]
},
"is-contact": {
"type": ["null", "boolean"]
},
"property": {
"type": ["null", "string"]
},
"vid": {
"type": ["null", "integer"]
},
"canonical-vid": {
"type": ["null", "integer"]
},
"portal-id": {
"type": ["null", "integer"]
},
"source-vids": {
"type": ["array", "null"],
"items": {

52
airbyte-integrations/connectors/source-hubspot/source_hubspot/schemas/deals_property_history.json Normal file
View File

@@ -0,0 +1,52 @@
{
"$schema": "http://json-schema.org/draft-07/schema",
"type": ["null", "object"],
"additionalProperties": true,
"properties": {
"updatedByUserId": {
"type": ["null", "number"]
},
"requestId": {
"type": ["null", "string"]
},
"source": {
"type": ["null", "string"]
},
"portalId": {
"type": ["null", "number"]
},
"isDeleted": {
"type": ["null", "boolean"]
},
"timestamp": {
"type": ["null", "number"]
},
"property": {
"type": ["null", "string"]
},
"persistenceTimestamp": {
"type": ["null", "number"]
},
"name": {
"type": ["null", "string"]
},
"sourceVid": {
"type": ["null", "array"]
},
"useTimestampAsPersistenceTimestamp": {
"type": ["null", "boolean"]
},
"sourceMetadata": {
"type": ["null", "string"]
},
"dealId": {
"type": ["null", "number"]
},
"sourceId": {
"type": ["null", "string"]
},
"value": {
"type": ["null", "string"]
}
}
}

8
airbyte-integrations/connectors/source-hubspot/source_hubspot/source.py
View File

@@ -17,16 +17,19 @@ from source_hubspot.streams import (
API,
Campaigns,
Companies,
CompaniesPropertyHistory,
CompaniesWebAnalytics,
ContactLists,
Contacts,
ContactsListMemberships,
ContactsMergedAudit,
ContactsPropertyHistory,
ContactsWebAnalytics,
CustomObject,
DealPipelines,
Deals,
DealsArchived,
DealsPropertyHistory,
DealsWebAnalytics,
EmailEvents,
EmailSubscriptions,
@@ -52,7 +55,6 @@ from source_hubspot.streams import (
OwnersArchived,
Products,
ProductsWebAnalytics,
PropertyHistory,
SubscriptionChanges,
TicketPipelines,
Tickets,
@@ -136,7 +138,9 @@ class SourceHubspot(AbstractSource):
Owners(**common_params),
OwnersArchived(**common_params),
Products(**common_params),
PropertyHistory(**common_params),
ContactsPropertyHistory(**common_params),
CompaniesPropertyHistory(**common_params),
DealsPropertyHistory(**common_params),
SubscriptionChanges(**common_params),
Tickets(**common_params),
TicketPipelines(**common_params),

280
airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py
View File

@@ -31,7 +31,15 @@ from airbyte_cdk.utils import AirbyteTracedException
from requests import HTTPError, codes
from source_hubspot.constants import OAUTH_CREDENTIALS, PRIVATE_APP_CREDENTIALS
from source_hubspot.errors import HubspotAccessDenied, HubspotInvalidAuth, HubspotRateLimited, HubspotTimeout, InvalidStartDateConfigError
from source_hubspot.helpers import APIv1Property, APIv3Property, GroupByKey, IRecordPostProcessor, IURLPropertyRepresentation, StoreAsIs
from source_hubspot.helpers import (
APIv1Property,
APIv2Property,
APIv3Property,
GroupByKey,
IRecordPostProcessor,
IURLPropertyRepresentation,
StoreAsIs,
)
# we got this when provided API Token has incorrect format
CLOUDFLARE_ORIGIN_DNS_ERROR = 530
@@ -356,6 +364,7 @@ class Stream(HttpStream, ABC):
stream_state: Mapping[str, Any] = None,
stream_slice: Mapping[str, Any] = None,
next_page_token: Mapping[str, Any] = None,
properties: IURLPropertyRepresentation = None,
) -> str:
return self.url
@@ -364,6 +373,8 @@ class Stream(HttpStream, ABC):
properties = list(self.properties.keys())
if "v1" in self.url:
return APIv1Property(properties)
if "v2" in self.url:
return APIv2Property(properties)
return APIv3Property(properties)
def __init__(self, api: API, start_date: Union[str, pendulum.datetime], credentials: Mapping[str, Any] = None, **kwargs):
@@ -410,6 +421,10 @@ class Stream(HttpStream, ABC):
json_schema["properties"] = {**default_props, **properties, **unnested_properties}
return json_schema
def update_request_properties(self, params: Mapping[str, Any], properties: IURLPropertyRepresentation) -> None:
if properties:
params.update(properties.as_url_param())
@retry_token_expired_handler(max_tries=5)
def handle_request(
self,
@@ -420,11 +435,10 @@ class Stream(HttpStream, ABC):
) -> requests.Response:
request_headers = self.request_headers(stream_state=stream_state, stream_slice=stream_slice, next_page_token=next_page_token)
request_params = self.request_params(stream_state=stream_state, stream_slice=stream_slice, next_page_token=next_page_token)
if properties:
request_params.update(properties.as_url_param())
self.update_request_properties(request_params, properties)
request = self._create_prepared_request(
path=self.path(stream_state=stream_state, stream_slice=stream_slice, next_page_token=next_page_token),
path=self.path(stream_state=stream_state, stream_slice=stream_slice, next_page_token=next_page_token, properties=properties),
headers=dict(request_headers, **self.authenticator.get_auth_header()),
params=request_params,
json=self.request_body_json(stream_state=stream_state, stream_slice=stream_slice, next_page_token=next_page_token),
@@ -904,6 +918,7 @@ class AssociationsStream(Stream):
stream_state: Mapping[str, Any] = None,
stream_slice: Mapping[str, Any] = None,
next_page_token: Mapping[str, Any] = None,
properties: IURLPropertyRepresentation = None,
) -> str:
return f"/crm/v4/associations/{self.parent_stream.entity}/{stream_slice}/batch/read"
@@ -1669,6 +1684,7 @@ class FormSubmissions(ClientSideIncrementalStream):
stream_state: Mapping[str, Any] = None,
stream_slice: Mapping[str, Any] = None,
next_page_token: Mapping[str, Any] = None,
properties: IURLPropertyRepresentation = None,
) -> str:
return f"{self.url}/{stream_slice['form_id']}"
@@ -1769,23 +1785,76 @@ class PropertyHistory(ClientSideIncrementalStream):
Docs: https://legacydocs.hubspot.com/docs/methods/contacts/get_contacts
"""
more_key = "has-more"
url = "/contacts/v1/lists/all/contacts/all"
updated_at_field = "timestamp"
created_at_field = "timestamp"
entity = "contacts"
data_field = "contacts"
page_field = "vid-offset"
page_filter = "vidOffset"
primary_key = "vid"
denormalize_records = True
limit_field = "count"
limit = 100
scopes = {"crm.objects.contacts.read"}
properties_scopes = {"crm.schemas.contacts.read"}
@property
def cursor_field_datetime_format(self):
@abstractmethod
def page_field(self) -> str:
"""Page offset field"""
@property
@abstractmethod
def limit_field(self) -> str:
"""Limit query field"""
@property
@abstractmethod
def page_filter(self) -> str:
"""Query param name that indicates page offset"""
@property
@abstractmethod
def more_key(self) -> str:
"""Field that indicates that are more records"""
@property
@abstractmethod
def scopes(self) -> set:
"""Scopes needed to get access to CRM object"""
@property
@abstractmethod
def properties_scopes(self) -> set:
"""Scopes needed to get access to CRM object properies"""
@property
@abstractmethod
def entity(self) -> str:
"""
CRM object entity name.
This is usually a part of some URL or key that contains data in response
"""
@property
@abstractmethod
def primary_key(self) -> str:
"""Indicates a field name which is considered to be a primary key of the stream"""
@property
@abstractmethod
def additional_keys(self) -> list:
"""The root keys to be placed into each record while iterating through versions"""
@property
@abstractmethod
def last_modified_date_field_name(self) -> str:
"""Last modified date field name"""
@property
@abstractmethod
def data_field(self) -> str:
"""A key that contains data in response"""
@property
@abstractmethod
def url(self) -> str:
"""An API url"""
@property
def cursor_field_datetime_format(self) -> str:
"""Cursor value expected to be a timestamp in milliseconds"""
return "x"
@@ -1804,10 +1873,11 @@ class PropertyHistory(ClientSideIncrementalStream):
for record in records:
properties = record.get("properties")
primary_key = record.get(self.primary_key)
additional_keys = {additional_key: record.get(additional_key) for additional_key in self.additional_keys}
value_dict: Dict
for property_name, value_dict in properties.items():
versions = value_dict.get("versions")
if property_name == "lastmodifieddate":
if property_name == self.last_modified_date_field_name:
# Skipping the lastmodifieddate since it only returns the value
# when one field of a contact was changed no matter which
# field was changed. It therefore creates overhead, since for
@@ -1818,7 +1888,183 @@ class PropertyHistory(ClientSideIncrementalStream):
for version in versions:
version["property"] = property_name
version[self.primary_key] = primary_key
yield version
yield version | additional_keys
class ContactsPropertyHistory(PropertyHistory):
@property
def scopes(self):
return {"crm.objects.contacts.read"}
@property
def properties_scopes(self):
return {"crm.schemas.contacts.read"}
@property
def page_field(self) -> str:
return "vid-offset"
@property
def limit_field(self) -> str:
return "count"
@property
def page_filter(self) -> str:
return "vidOffset"
@property
def more_key(self) -> str:
return "has-more"
@property
def entity(self):
return "contacts"
@property
def primary_key(self) -> list:
return "vid"
@property
def additional_keys(self) -> list:
return ["portal-id", "is-contact", "canonical-vid"]
@property
def last_modified_date_field_name(self):
return "lastmodifieddate"
@property
def data_field(self):
return "contacts"
@property
def url(self):
return "/contacts/v1/lists/all/contacts/all"
class CompaniesPropertyHistory(PropertyHistory):
@property
def scopes(self) -> set:
return {"crm.objects.companies.read"}
@property
def properties_scopes(self) -> set:
return {"crm.schemas.companies.read"}
@property
def page_field(self) -> str:
return "offset"
@property
def limit_field(self) -> str:
return "limit"
@property
def page_filter(self) -> str:
return "offset"
@property
def more_key(self) -> str:
return "hasMore"
@property
def entity(self) -> str:
return "companies"
@property
def primary_key(self) -> list:
return "companyId"
@property
def additional_keys(self) -> list:
return ["portalId", "isDeleted"]
@property
def last_modified_date_field_name(self) -> str:
return "hs_lastmodifieddate"
@property
def data_field(self) -> str:
return "companies"
@property
def url(self) -> str:
return "/companies/v2/companies/paged"
def update_request_properties(self, params: Mapping[str, Any], properties: IURLPropertyRepresentation) -> None:
pass
def path(
self,
*,
stream_state: Mapping[str, Any] = None,
stream_slice: Mapping[str, Any] = None,
next_page_token: Mapping[str, Any] = None,
properties: IURLPropertyRepresentation = None,
) -> str:
return f"{self.url}?{properties.as_url_param_with_history()}"
class DealsPropertyHistory(PropertyHistory):
@property
def scopes(self) -> set:
return {"crm.objects.deals.read"}
@property
def properties_scopes(self):
return {"crm.schemas.deals.read"}
@property
def page_field(self) -> str:
return "offset"
@property
def limit_field(self) -> str:
return "limit"
@property
def page_filter(self) -> str:
return "offset"
@property
def more_key(self) -> str:
return "hasMore"
@property
def entity(self) -> set:
return "deals"
@property
def primary_key(self) -> list:
return "dealId"
@property
def additional_keys(self) -> list:
return ["portalId", "isDeleted"]
@property
def last_modified_date_field_name(self) -> str:
return "hs_lastmodifieddate"
@property
def data_field(self) -> str:
return "deals"
@property
def url(self) -> str:
return "/deals/v1/deal/paged"
def update_request_properties(self, params: Mapping[str, Any], properties: IURLPropertyRepresentation) -> None:
pass
def path(
self,
*,
stream_state: Mapping[str, Any] = None,
stream_slice: Mapping[str, Any] = None,
next_page_token: Mapping[str, Any] = None,
properties: IURLPropertyRepresentation = None,
) -> str:
return f"{self.url}?{properties.as_url_param_with_history()}"
class SubscriptionChanges(IncrementalStream):

45
airbyte-integrations/connectors/source-hubspot/unit_tests/test_source.py
View File

@@ -85,7 +85,7 @@ def test_streams(requests_mock, config):
streams = SourceHubspot().streams(config)
assert len(streams) == 30
assert len(streams) == 32
@mock.patch("source_hubspot.source.SourceHubspot.get_custom_object_streams")
@@ -93,7 +93,7 @@ def test_streams(requests_mock, config_experimental):
streams = SourceHubspot().streams(config_experimental)
assert len(streams) == 42
assert len(streams) == 44
def test_custom_streams(config_experimental):
@@ -478,7 +478,11 @@ def test_search_based_stream_should_not_attempt_to_get_more_than_10k_records(req
requests_mock.register_uri("POST", test_stream.url, responses)
test_stream._sync_mode = None
requests_mock.register_uri("GET", "/properties/v2/company/properties", properties_response)
requests_mock.register_uri("POST", "/crm/v4/associations/company/contacts/batch/read", [{"status_code": 200, "json": {"results": []}}])
requests_mock.register_uri(
"POST",
"/crm/v4/associations/company/contacts/batch/read",
[{"status_code": 200, "json": {"results": [{"from": {"id": "1"}, "to": [{"toObjectId": "2"}]}]}}]
)
records, _ = read_incremental(test_stream, {})
# The stream should not attempt to get more than 10K records.
@@ -700,3 +704,38 @@ def test_pagination_marketing_emails_stream(requests_mock, common_params):
records = read_full_refresh(test_stream)
# The stream should handle pagination correctly and output 600 records.
assert len(records) == 600
def test_get_granted_scopes(requests_mock, mocker):
authenticator = mocker.Mock()
authenticator.get_access_token.return_value = "the-token"
expected_scopes = ["a", "b", "c"]
response = [
{"json": {"scopes": expected_scopes}, "status_code": 200},
]
requests_mock.register_uri("GET", "https://api.hubapi.com/oauth/v1/access-tokens/the-token", response)
actual_scopes = SourceHubspot().get_granted_scopes(authenticator)
assert expected_scopes == actual_scopes
def test_streams_oauth_2_auth_no_suitable_scopes(requests_mock, mocker, config):
authenticator = mocker.Mock()
authenticator.get_access_token.return_value = "the-token"
mocker.patch("source_hubspot.streams.API.is_oauth2", return_value=True)
mocker.patch("source_hubspot.streams.API.get_authenticator", return_value=authenticator)
requests_mock.get("https://api.hubapi.com/crm/v3/schemas", json={}, status_code=200)
expected_scopes = ["no.scopes.granted"]
response = [
{"json": {"scopes": expected_scopes}, "status_code": 200},
]
requests_mock.register_uri("GET", "https://api.hubapi.com/oauth/v1/access-tokens/the-token", response)
streams = SourceHubspot().streams(config)
assert len(streams) == 0

51
airbyte-integrations/connectors/source-hubspot/unit_tests/test_streams.py
View File

@@ -10,7 +10,9 @@ from source_hubspot.streams import (
Companies,
ContactLists,
Contacts,
ContactsListMemberships,
ContactsMergedAudit,
ContactsPropertyHistory,
ContactsWebAnalytics,
CustomObject,
DealPipelines,
@@ -31,7 +33,6 @@ from source_hubspot.streams import (
Owners,
OwnersArchived,
Products,
PropertyHistory,
RecordUnnester,
TicketPipelines,
Tickets,
@@ -551,7 +552,7 @@ def test_web_analytics_latest_state(common_params, mocker):
def test_property_history_transform(common_params):
stream = PropertyHistory(**common_params)
stream = ContactsPropertyHistory(**common_params)
versions = [
{
"value": "Georgia",
@@ -561,10 +562,54 @@ def test_property_history_transform(common_params):
records = [
{
"vid": 1,
"canonical-vid": 1,
"portal-id": 1,
"is-contact": True,
"properties": {
"hs_country": {"versions": versions},
"lastmodifieddate": {"value": 1645135236625}
}
}
]
assert [{"vid": 1, "property": "hs_country", **version} for version in versions] == list(stream._transform(records=records))
assert [
{
"vid": 1,
"canonical-vid": 1,
"portal-id": 1,
"is-contact": True,
"property": "hs_country",
**version
} for version in versions
] == list(stream._transform(records=records))
def test_contacts_membership_transform(common_params):
stream = ContactsListMemberships(**common_params)
versions = [
{
"value": "Georgia",
"timestamp": 1645135236625
}
]
memberships = [
{"membership": 1}
]
records = [
{
"vid": 1,
"canonical-vid": 1,
"portal-id": 1,
"is-contact": True,
"properties": {
"hs_country": {"versions": versions},
"lastmodifieddate": {"value": 1645135236625}
},
"list-memberships": memberships
}
]
assert [
{
"membership": 1,
"canonical-vid": 1
} for _ in versions
] == list(stream._transform(records=records))

28
docs/integrations/sources/hubspot-migrations.md Normal file
View File

@@ -0,0 +1,28 @@
# HubSpot Migration Guide
## Upgrading to 2.0.0
Note: this change is only breaking if you are using the PropertyHistory stream.
With this update, you can now access historical property changes for Deals and Companies, in addition to Contacts. Property History stream has been renamed to Contacts Property History (since it historically contained historical property changes from Contacts) and two new streams were added: Deals Property History and Companies Property History.
This is a breaking change because Property History has been replaced with Contacts Property History, so please follow the instructions below to migrate to version 2.0.0:
1. Select **Connections** in the main navbar.
1.1 Select the connection(s) affected by the update.
2. Select the **Replication** tab.
2.1 Select **Refresh source schema**.
```note
Any detected schema changes will be listed for your review.
```
2.2 Select **OK**.
3. Select **Save changes** at the bottom of the page.
3.1 Ensure the **Reset affected streams** option is checked.
```note
Depending on destination type you may not be prompted to reset your data
```
4. Select **Save connection**.
```note
This will reset the data in your destination and initiate a fresh sync.
```
For more information on resetting your data in Airbyte, see [this page](https://docs.airbyte.com/operator-guides/reset).

7
docs/integrations/sources/hubspot.md
View File

@@ -185,7 +185,9 @@ The HubSpot source connector supports the following streams:
- [Marketing Emails](https://legacydocs.hubspot.com/docs/methods/cms_email/get-all-marketing-email-statistics)
- [Owners](https://developers.hubspot.com/docs/methods/owners/get_owners) \(Client-Side Incremental\)
- [Products](https://developers.hubspot.com/docs/api/crm/products) \(Incremental\)
- [Property History](https://legacydocs.hubspot.com/docs/methods/contacts/get_contacts) \(Incremental\)
- [Contacts Property History](https://legacydocs.hubspot.com/docs/methods/contacts/get_contacts) \(Client-Side Incremental\)
- [Companies Property History](https://legacydocs.hubspot.com/docs/methods/companies/get-all-companies) \(Client-Side Incremental\)
- [Deals Property History](https://legacydocs.hubspot.com/docs/methods/deals/get-all-deals) \(Client-Side Incremental\)
- [Subscription Changes](https://developers.hubspot.com/docs/methods/email/get_subscriptions_timeline) \(Incremental\)
- [Tickets](https://developers.hubspot.com/docs/api/crm/tickets) \(Incremental\)
- [Ticket Pipelines](https://developers.hubspot.com/docs/api/crm/pipelines) \(Client-Side Incremental\)
@@ -300,7 +302,8 @@ The connector is restricted by normal HubSpot [rate limitations](https://legacyd
| Version | Date | Pull Request | Subject |
|:--------|:-----------|:---------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| 1.9.0 | 2023-12-04 | [33042](https://github.com/airbytehq/airbyte/pull/33042) | Add Web Analytics streams |
| 2.0.0 | 2023-12-08 | [33266](https://github.com/airbytehq/airbyte/pull/33266) | Added ContactsPropertyHistory, CompaniesPropertyHistory, DealsPropertyHistory streams |
| 1.9.0 | 2023-12-04 | [33042](https://github.com/airbytehq/airbyte/pull/33042) | Added Web Analytics streams |
| 1.8.0 | 2023-11-23 | [32778](https://github.com/airbytehq/airbyte/pull/32778) | Extend `PropertyHistory` stream to support incremental sync |
| 1.7.0 | 2023-11-01 | [32035](https://github.com/airbytehq/airbyte/pull/32035) | Extend the `Forms` stream schema |
| 1.6.1 | 2023-10-20 | [31644](https://github.com/airbytehq/airbyte/pull/31644) | Base image migration: remove Dockerfile and use the python-connector-base image |