From 7aac7ba7e1fac2716a1adc57c8413efc06856f8b Mon Sep 17 00:00:00 2001 From: Sam RL Date: Wed, 27 May 2026 22:40:07 +0100 Subject: [PATCH 1/2] Map Pipedrive V2 custom fields --- .../pipedrive/helpers/custom_fields_munger.py | 67 +++++- sources/pipedrive/rest_v2/__init__.py | 91 +++++++- tests/pipedrive/test_pipedrive_source.py | 199 +++++++++++++++++- 3 files changed, 345 insertions(+), 12 deletions(-) diff --git a/sources/pipedrive/helpers/custom_fields_munger.py b/sources/pipedrive/helpers/custom_fields_munger.py index b65a4e30e..cd9872b00 100644 --- a/sources/pipedrive/helpers/custom_fields_munger.py +++ b/sources/pipedrive/helpers/custom_fields_munger.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, TypedDict, Optional +from typing import Any, Dict, TypedDict, Optional, cast import dlt @@ -100,3 +100,68 @@ def rename_fields(data: TDataPage, fields_mapping: Dict[str, Any]) -> TDataPage: field_value = options_map.get(str(field_value), field_value) data_item[field_name] = field_value return data + + +def build_v2_fields_mapping(fields: TDataPage) -> Dict[str, TFieldMapping]: + """Build a field mapping from Pipedrive API v2 field metadata.""" + fields_mapping: Dict[str, TFieldMapping] = {} + for field in fields: + if not field.get("is_custom_field"): + continue + field_code = field.get("field_code") + if not field_code: + continue + fields_mapping[field_code] = { + "name": field["field_name"], + "normalized_name": _normalized_name(field["field_name"]), + "options": _build_v2_options_map(field.get("options")), + "field_type": field["field_type"], + } + return fields_mapping + + +def update_v2_fields_mapping( + new_fields_mapping: TDataPage, existing_fields_mapping: Dict[str, Any] +) -> Dict[str, Any]: + """Update custom fields state from Pipedrive API v2 field metadata.""" + for field in new_fields_mapping: + if not field.get("is_custom_field"): + continue + field_code = field.get("field_code") + if not field_code: + continue + data_item = { + "key": field_code, + "name": field["field_name"], + "field_type": field["field_type"], + "options": _build_v2_options(field.get("options")), + } + existing_fields_mapping = _update_field(data_item, existing_fields_mapping) + return existing_fields_mapping + + +def rename_v2_custom_fields( + data_item: Dict[str, Any], fields_mapping: Dict[str, Any] +) -> Dict[str, Any]: + """Move mapped Pipedrive API v2 custom fields to readable top-level fields.""" + custom_fields = data_item.get("custom_fields") + if not isinstance(custom_fields, dict) or not fields_mapping: + return data_item + data_item.pop("custom_fields") + rename_fields([custom_fields], fields_mapping) + data_item.update(custom_fields) + return data_item + + +def _build_v2_options_map(options: Optional[Any]) -> Dict[str, str]: + return {str(option["id"]): option["label"] for option in _build_v2_options(options)} + + +def _build_v2_options(options: Optional[Any]) -> TDataPage: + if not options: + return [] + if isinstance(options, dict): + return [ + {"id": option_id, "label": label} for option_id, label in options.items() + ] + return cast(TDataPage, options) diff --git a/sources/pipedrive/rest_v2/__init__.py b/sources/pipedrive/rest_v2/__init__.py index 37c6c4472..0ffdb1b31 100644 --- a/sources/pipedrive/rest_v2/__init__.py +++ b/sources/pipedrive/rest_v2/__init__.py @@ -1,12 +1,17 @@ """Pipedrive API v2 source. Loads data from Pipedrive API v2 endpoints with incremental support.""" -from typing import Iterable, Dict, Any, List, Optional, Union, cast +from typing import Callable, Iterable, Dict, Any, List, Optional, Union, cast import dlt from dlt.common import pendulum from dlt.common.time import ensure_pendulum_datetime from dlt.sources import DltResource +from dlt.sources.helpers import requests from dlt.sources.rest_api import rest_api_resources, RESTAPIConfig, EndpointResource +from ..helpers.custom_fields_munger import ( + rename_v2_custom_fields, + update_v2_fields_mapping, +) from ..settings import ENTITIES_V2, NESTED_ENTITIES_V2 INCREMENTAL_API_RESOURCES = { @@ -17,6 +22,14 @@ "products", } +CUSTOM_FIELD_API_RESOURCES = { + "activities": "activityFields", + "deals": "dealFields", + "persons": "personFields", + "organizations": "organizationFields", + "products": "productFields", +} + @dlt.source(name="pipedrive_v2", section="pipedrive") def pipedrive_v2_source( @@ -90,7 +103,41 @@ def rest_v2_resources( Returns: Iterable[DltResource]: Configured REST API resources. """ - resources = [] + resources: List[Union[str, EndpointResource, DltResource]] = [] + custom_fields_mappings: Dict[str, Dict[str, Any]] = {} + + def get_custom_fields_mapping(resource_name: str) -> Dict[str, Any]: + if resource_name not in custom_fields_mappings: + custom_fields_mapping_state = dlt.current.source_state().setdefault( + "custom_fields_mapping", {} + ) + fields = get_v2_fields( + pipedrive_api_key, + company_domain, + CUSTOM_FIELD_API_RESOURCES[resource_name], + ) + custom_fields_mappings[resource_name] = update_v2_fields_mapping( + fields, custom_fields_mapping_state.setdefault(resource_name, {}) + ) + custom_fields_mapping_state[resource_name] = custom_fields_mappings[ + resource_name + ] + return custom_fields_mappings[resource_name] + + def map_custom_fields( + data_item: Dict[str, Any], resource_name: str + ) -> Dict[str, Any]: + return rename_v2_custom_fields( + data_item, get_custom_fields_mapping(resource_name) + ) + + def make_custom_fields_mapper( + resource_name: str, + ) -> Callable[[Dict[str, Any]], Dict[str, Any]]: + def _map_custom_fields(data_item: Dict[str, Any]) -> Dict[str, Any]: + return map_custom_fields(data_item, resource_name) + + return _map_custom_fields for resource_name, endpoint_config in resource_configs.items(): endpoint_def = { @@ -106,15 +153,18 @@ def rest_v2_resources( "initial_value": since_timestamp, } - resources.append( - { - "name": f"{prefix}{resource_name}", - "endpoint": endpoint_def, - } - ) + resource: EndpointResource = { + "name": f"{prefix}{resource_name}", + "endpoint": cast(Any, endpoint_def), + } + if resource_name in CUSTOM_FIELD_API_RESOURCES: + resource["processing_steps"] = [ + {"map": make_custom_fields_mapper(resource_name)} + ] + resources.append(resource) for name, conf in nested_configs.items(): - nested_res = { + nested_res: EndpointResource = { "name": f"{prefix}{name}", "endpoint": { "path": conf["endpoint_path"].replace( @@ -162,7 +212,28 @@ def rest_v2_resources( }, }, }, - "resources": cast(List[Union[str, EndpointResource, DltResource]], resources), + "resources": resources, } yield from rest_api_resources(config) + + +def get_v2_fields( + pipedrive_api_key: str, company_domain: str, fields_endpoint: str +) -> List[Dict[str, Any]]: + """Fetch Pipedrive API v2 field metadata.""" + fields: List[Dict[str, Any]] = [] + params = {"api_token": pipedrive_api_key, "limit": "500"} + while True: + response = requests.get( + f"https://{company_domain}.pipedrive.com/api/v2/{fields_endpoint}", + params=params, + ) + response.raise_for_status() + page = response.json() + fields.extend(page["data"] or []) + next_cursor = page.get("additional_data", {}).get("next_cursor") + if not next_cursor: + break + params["cursor"] = next_cursor + return fields diff --git a/tests/pipedrive/test_pipedrive_source.py b/tests/pipedrive/test_pipedrive_source.py index e92646f34..2db8732aa 100644 --- a/tests/pipedrive/test_pipedrive_source.py +++ b/tests/pipedrive/test_pipedrive_source.py @@ -9,9 +9,10 @@ from dlt.common import pendulum from dlt.common.pipeline import TSourceState from dlt.common.schema import Schema +from dlt.extract.items_transform import MapItem from dlt.sources.helpers import requests -from sources.pipedrive import pipedrive_source, leads +from sources.pipedrive import pipedrive_source, leads, pipedrive_v2_source from tests.utils import ( ALL_DESTINATIONS, assert_load_info, @@ -21,6 +22,9 @@ from sources.pipedrive.helpers.custom_fields_munger import ( update_fields_mapping, rename_fields, + build_v2_fields_mapping, + rename_v2_custom_fields, + update_v2_fields_mapping, ) @@ -439,6 +443,199 @@ def test_rename_fields_with_set() -> None: assert result == [{"custom_field_1": ["b", "a", "c"], "id": 44, "name": "asdf"}] +def test_build_v2_fields_mapping() -> None: + fields = [ + { + "is_custom_field": True, + "field_code": "enum_hash", + "field_name": "Single option", + "field_type": "enum", + "options": [{"id": 12, "label": "Selected option"}], + }, + { + "is_custom_field": False, + "field_code": "regular_field", + "field_name": "Regular field", + "field_type": "varchar", + "options": [], + }, + ] + + with mock.patch.object(dlt.current, "source_schema", return_value=Schema("test")): + result = build_v2_fields_mapping(fields) + + assert result == { + "enum_hash": { + "name": "Single option", + "normalized_name": "single_option", + "field_type": "enum", + "options": {"12": "Selected option"}, + } + } + + +def test_update_v2_fields_mapping_keeps_existing_field_name() -> None: + mapping: Dict[str, Any] = { + "revenue_hash": { + "name": "Revenue", + "normalized_name": "revenue", + "field_type": "monetary", + "options": {}, + } + } + fields = [ + { + "is_custom_field": True, + "field_code": "revenue_hash", + "field_name": "Revenue GBP", + "field_type": "monetary", + "options": [], + }, + ] + + with mock.patch.object(dlt.current, "source_schema", return_value=Schema("test")): + result = update_v2_fields_mapping(fields, mapping) + + assert result["revenue_hash"]["name"] == "Revenue" + assert result["revenue_hash"]["normalized_name"] == "revenue" + + +def test_update_v2_fields_mapping_adds_new_options() -> None: + mapping: Dict[str, Any] = { + "services_hash": { + "name": "Services", + "normalized_name": "services", + "field_type": "set", + "options": {"62": "Distribution"}, + } + } + fields = [ + { + "is_custom_field": True, + "field_code": "services_hash", + "field_name": "Services", + "field_type": "set", + "options": [ + {"id": 62, "label": "Distribution renamed"}, + {"id": 63, "label": "Publishing"}, + ], + }, + ] + + with mock.patch.object(dlt.current, "source_schema", return_value=Schema("test")): + result = update_v2_fields_mapping(fields, mapping) + + assert result["services_hash"]["options"] == { + "62": "Distribution", + "63": "Publishing", + } + + +def test_rename_v2_custom_fields() -> None: + data_item = { + "id": 395, + "title": "Example deal", + "update_time": "2026-05-27T17:33:46Z", + "custom_fields": { + "enum_hash": 12, + "set_hash": [62, 63, 65], + "monetary_hash": {"value": 100, "currency": "GBP"}, + "empty_hash": None, + }, + } + mapping = { + "enum_hash": { + "name": "Single option", + "normalized_name": "single_option", + "field_type": "enum", + "options": {"12": "Selected option"}, + }, + "set_hash": { + "name": "Services", + "normalized_name": "services", + "field_type": "set", + "options": { + "62": "Service A", + "63": "Service B", + "65": "Service C", + }, + }, + "monetary_hash": { + "name": "Budget", + "normalized_name": "budget", + "field_type": "monetary", + "options": {}, + }, + "empty_hash": { + "name": "Empty field", + "normalized_name": "empty_field", + "field_type": "varchar", + "options": {}, + }, + } + + result = rename_v2_custom_fields(data_item, mapping) + + assert result == { + "id": 395, + "title": "Example deal", + "update_time": "2026-05-27T17:33:46Z", + "Single option": "Selected option", + "Services": ["Service A", "Service B", "Service C"], + "Budget": {"value": 100, "currency": "GBP"}, + "Empty field": None, + } + + +def test_rename_v2_custom_fields_keeps_custom_fields_without_mapping() -> None: + data_item = {"id": 395, "custom_fields": {"enum_hash": 12}} + + result = rename_v2_custom_fields(data_item, {}) + + assert result == {"id": 395, "custom_fields": {"enum_hash": 12}} + + +def test_pipedrive_v2_deals_maps_custom_fields() -> None: + with mock.patch("sources.pipedrive.rest_v2.get_v2_fields") as get_v2_fields: + source = pipedrive_v2_source( + pipedrive_api_key="token", + company_domain="company", + resources=["deals"], + ) + + deals = source.resources["v2_deals"] + pipes = list(deals._pipe) + + get_v2_fields.assert_not_called() + assert any(isinstance(item, MapItem) for item in pipes) + + +def test_pipedrive_v2_persons_maps_custom_fields() -> None: + source = pipedrive_v2_source( + pipedrive_api_key="token", + company_domain="company", + resources=["persons"], + ) + + persons = source.resources["v2_persons"] + pipes = list(persons._pipe) + + assert any(isinstance(item, MapItem) for item in pipes) + + +def test_pipedrive_v2_pipelines_do_not_map_custom_fields() -> None: + source = pipedrive_v2_source( + pipedrive_api_key="token", + company_domain="company", + resources=["pipelines"], + ) + + pipelines = source.resources["v2_pipelines"] + pipes = list(pipelines._pipe) + + assert all(not isinstance(item, MapItem) for item in pipes) + + def test_recents_none_data_items_from_recents() -> None: pytest.skip("Unskip after setting up credentials.") """Pages from /recents sometimes contain `None` data items which cause errors. From 546d552f953678d9c7da7cfdbed139ec0253f362 Mon Sep 17 00:00:00 2001 From: Sam RL Date: Thu, 28 May 2026 10:45:59 +0100 Subject: [PATCH 2/2] Make Pipedrive V2 custom field mapping opt-in --- sources/pipedrive/rest_v2/__init__.py | 11 ++++++++--- tests/pipedrive/test_pipedrive_source.py | 16 ++++++++++++++++ 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/sources/pipedrive/rest_v2/__init__.py b/sources/pipedrive/rest_v2/__init__.py index 0ffdb1b31..294aa617f 100644 --- a/sources/pipedrive/rest_v2/__init__.py +++ b/sources/pipedrive/rest_v2/__init__.py @@ -38,6 +38,7 @@ def pipedrive_v2_source( resources: Optional[List[str]] = None, prefix: str = "v2_", since_timestamp: Optional[Union[pendulum.DateTime, str]] = "1970-01-01 00:00:00", + map_custom_fields: bool = False, ) -> Iterable[DltResource]: """Load Pipedrive API v2 data with incremental filtering. @@ -47,6 +48,7 @@ def pipedrive_v2_source( resources: List of resource names to load. If None, all available resources are loaded. prefix: Prefix to add to resource names. Defaults to "v2_". since_timestamp: Start timestamp for incremental loading. Defaults to "1970-01-01 00:00:00". + map_custom_fields: Map Pipedrive custom field hashes to readable names. Defaults to False. Returns: Iterable[DltResource]: Resources for Pipedrive API v2 endpoints. @@ -77,6 +79,7 @@ def pipedrive_v2_source( nested_configs_to_create, prefix, since_timestamp_rfc3339, + map_custom_fields, ) for resource in v2_resources: yield resource @@ -89,6 +92,7 @@ def rest_v2_resources( nested_configs: Dict[str, Dict[str, Any]], prefix: str, since_timestamp: str, + map_custom_fields: bool, ) -> Iterable[DltResource]: """Build REST v2 resources with nested endpoints. @@ -99,6 +103,7 @@ def rest_v2_resources( nested_configs: Configuration for nested/dependent resources. prefix: Prefix to add to resource names. since_timestamp: Timestamp for incremental filtering. + map_custom_fields: Whether to map custom field hashes to readable names. Returns: Iterable[DltResource]: Configured REST API resources. @@ -124,7 +129,7 @@ def get_custom_fields_mapping(resource_name: str) -> Dict[str, Any]: ] return custom_fields_mappings[resource_name] - def map_custom_fields( + def map_resource_custom_fields( data_item: Dict[str, Any], resource_name: str ) -> Dict[str, Any]: return rename_v2_custom_fields( @@ -135,7 +140,7 @@ def make_custom_fields_mapper( resource_name: str, ) -> Callable[[Dict[str, Any]], Dict[str, Any]]: def _map_custom_fields(data_item: Dict[str, Any]) -> Dict[str, Any]: - return map_custom_fields(data_item, resource_name) + return map_resource_custom_fields(data_item, resource_name) return _map_custom_fields @@ -157,7 +162,7 @@ def _map_custom_fields(data_item: Dict[str, Any]) -> Dict[str, Any]: "name": f"{prefix}{resource_name}", "endpoint": cast(Any, endpoint_def), } - if resource_name in CUSTOM_FIELD_API_RESOURCES: + if map_custom_fields and resource_name in CUSTOM_FIELD_API_RESOURCES: resource["processing_steps"] = [ {"map": make_custom_fields_mapper(resource_name)} ] diff --git a/tests/pipedrive/test_pipedrive_source.py b/tests/pipedrive/test_pipedrive_source.py index 2db8732aa..38bdf219b 100644 --- a/tests/pipedrive/test_pipedrive_source.py +++ b/tests/pipedrive/test_pipedrive_source.py @@ -601,6 +601,7 @@ def test_pipedrive_v2_deals_maps_custom_fields() -> None: pipedrive_api_key="token", company_domain="company", resources=["deals"], + map_custom_fields=True, ) deals = source.resources["v2_deals"] @@ -615,6 +616,7 @@ def test_pipedrive_v2_persons_maps_custom_fields() -> None: pipedrive_api_key="token", company_domain="company", resources=["persons"], + map_custom_fields=True, ) persons = source.resources["v2_persons"] @@ -623,11 +625,25 @@ def test_pipedrive_v2_persons_maps_custom_fields() -> None: assert any(isinstance(item, MapItem) for item in pipes) +def test_pipedrive_v2_does_not_map_custom_fields_by_default() -> None: + source = pipedrive_v2_source( + pipedrive_api_key="token", + company_domain="company", + resources=["deals"], + ) + + deals = source.resources["v2_deals"] + pipes = list(deals._pipe) + + assert all(not isinstance(item, MapItem) for item in pipes) + + def test_pipedrive_v2_pipelines_do_not_map_custom_fields() -> None: source = pipedrive_v2_source( pipedrive_api_key="token", company_domain="company", resources=["pipelines"], + map_custom_fields=True, ) pipelines = source.resources["v2_pipelines"]