[Core] Added ingest url support (#1972)

### **User description**
# Description

What - Added ingest url support

Why - To be able to send data to the ingest url

How - new env variable

## Type of change

Please leave one option from the following and delete the rest:

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] New Integration (non-breaking change which adds a new integration)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Non-breaking change (fix of existing functionality that will not
change current behavior)
- [ ] Documentation (added/updated documentation)

<h4> All tests should be run against the port production
environment(using a testing org). </h4>

### Core testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Resync finishes successfully
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Scheduled resync able to abort existing resync and start a new one
- [ ] Tested with at least 2 integrations from scratch
- [ ] Tested with Kafka and Polling event listeners
- [ ] Tested deletion of entities that don't pass the selector


___

### **PR Type**
Enhancement


___

### **Description**
- Added ingest URL support for data ingestion

- Modified Port client to accept ingest URL parameter

- Updated raw data posting to use ingest endpoint

- Added ingest URL configuration with default value


___

### Diagram Walkthrough


```mermaid
flowchart LR
  A["Configuration"] -- "adds ingest_url" --> B["Port Settings"]
  B -- "passes to" --> C["Port Client"]
  C -- "initializes" --> D["Port Authentication"]
  D -- "provides URL for" --> E["Raw Data Ingestion"]
```



<details> <summary><h3> File Walkthrough</h3></summary>

<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>authentication.py</strong><dd><code>Add ingest URL
parameter to authentication</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
</dd></summary>
<hr>

port_ocean/clients/port/authentication.py

<ul><li>Added <code>ingest_url</code> parameter to constructor<br> <li>
Stored ingest URL as instance variable</ul>


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1972/files#diff-8f8323c5413065b6eb1248e6236a4cf41985610d82b71b405b3a72ab35b44fda">+2/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>client.py</strong><dd><code>Update PortClient to
support ingest URL</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; </dd></summary>
<hr>

port_ocean/clients/port/client.py

<ul><li>Added <code>ingest_url</code> parameter to PortClient
constructor<br> <li> Passed ingest URL to PortAuthentication
initialization<br> <li> Minor import reordering</ul>


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1972/files#diff-ad4a0acea99647c007146b6c19e22ecd670838144cf5c58f3b08a0996190b82c">+5/-2</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>integrations.py</strong><dd><code>Use ingest URL for
raw data posting</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

port_ocean/clients/port/mixins/integrations.py

<ul><li>Changed raw data POST endpoint to use
<code>ingest_url</code><br> <li> Updated logging from debug to info
level</ul>


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1972/files#diff-6fa1fdcc86d6cf46d57bc513204cd5ffeafede20d1a079786705644c4937a521">+3/-3</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>sync_raw.py</strong><dd><code>Add logging for lakehouse
feature flags</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; </dd></summary>
<hr>

port_ocean/core/integrations/mixins/sync_raw.py

- Added info logging for lakehouse data flags


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1972/files#diff-7087c0350a44c64c6ca1e4333b34e158b354bdd15932e51860efaddda535cddc">+1/-0</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>ocean.py</strong><dd><code>Initialize PortClient with
ingest URL</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; </dd></summary>
<hr>

port_ocean/ocean.py

<ul><li>Passed ingest URL to PortClient initialization<br> <li> Minor
import reordering and formatting</ul>


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1972/files#diff-c8b6a91b406bdc777f220ab2fcead0b5027c8817c7c2e2e7adb06c192cc95ad6">+11/-12</a>&nbsp;
</td>

</tr>
</table></td></tr><tr><td><strong>Configuration
changes</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>settings.py</strong><dd><code>Add ingest URL
configuration setting</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

port_ocean/config/settings.py

<ul><li>Added <code>ingest_url</code> field to PortSettings<br> <li> Set
default value to "https://ingest.getport.io"<br> <li> Minor import
cleanup</ul>


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1972/files#diff-c4ab144128bdbbe50dd95b726ae9c2f4cbc7306025e07a9e1db3dff7c5d4b3bb">+1/-1</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

</details>

___

Author: yaelibarg

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 <!-- towncrier release notes start -->
+## 0.27.10 (2025-08-24)
+
+### Improvements
+
+- Added ingest url support
+
 ## 0.27.9 (2025-08-20)
 
 ### Improvements

port_ocean/clients/port/authentication.py

@@ -35,6 +35,7 @@ def __init__(
         integration_identifier: str,
         integration_type: str,
         integration_version: str,
+        ingest_url: str,
     ):
         self.client = client
         self.api_url = api_url
@@ -43,6 +44,7 @@ def __init__(
         self.integration_identifier = integration_identifier
         self.integration_type = integration_type
         self.integration_version = integration_version
+        self.ingest_url = ingest_url
         self.last_token_object: TokenResponse | None = None
 
     async def _get_token(self, client_id: str, client_secret: str) -> TokenResponse:

port_ocean/clients/port/client.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from loguru import logger
 
 from port_ocean.clients.port.authentication import PortAuthentication
@@ -10,11 +12,10 @@
     KafkaCreds,
 )
 from port_ocean.clients.port.utils import (
-    handle_port_status_code,
     get_internal_http_client,
+    handle_port_status_code,
 )
 from port_ocean.exceptions.clients import KafkaCredentialsNotFound
-from typing import Any
 
 
 class PortClient(
@@ -32,6 +33,7 @@ def __init__(
         integration_identifier: str,
         integration_type: str,
         integration_version: str,
+        ingest_url: str,
     ):
         self.api_url = f"{base_url}/v1"
         self.client = get_internal_http_client(self)
@@ -43,6 +45,7 @@ def __init__(
             integration_identifier,
             integration_type,
             integration_version,
+            ingest_url,
         )
         EntityClientMixin.__init__(self, self.auth, self.client)
         IntegrationClientMixin.__init__(

port_ocean/clients/port/mixins/integrations.py

@@ -296,7 +296,7 @@ async def post_integration_raw_data(
         logger.debug("starting POST raw data request", raw_data=raw_data)
         headers = await self.auth.headers()
         response = await self.client.post(
-            f"{self.auth.api_url}/lakehouse/integration/{self.integration_identifier}/sync/{sync_id}/kind/{kind}/items",
+            f"{self.auth.ingest_url}/lakehouse/integration-type/{self.auth.integration_type}/integration/{self.integration_identifier}/sync/{sync_id}/kind/{kind}/items",
             headers=headers,
             json={
                 "items": raw_data,

port_ocean/config/settings.py

@@ -9,7 +9,6 @@
 
 from port_ocean.config.base import BaseOceanModel, BaseOceanSettings
 from port_ocean.core.event_listener import EventListenerSettingsType
-
 from port_ocean.core.models import (
     CachingStorageMode,
     CreatePortResourcesOrigin,
@@ -47,6 +46,7 @@ class PortSettings(BaseOceanModel, extra=Extra.allow):
     client_secret: str = Field(..., sensitive=True)
     base_url: AnyHttpUrl = parse_obj_as(AnyHttpUrl, "https://api.getport.io")
     port_app_config_cache_ttl: int = 60
+    ingest_url: AnyHttpUrl = parse_obj_as(AnyHttpUrl, "https://ingest.getport.io")
 
 
 class IntegrationSettings(BaseOceanModel, extra=Extra.allow):

port_ocean/ocean.py

@@ -1,21 +1,18 @@
 import asyncio
 import sys
-from contextlib import asynccontextmanager
 import threading
+from contextlib import asynccontextmanager
 from typing import Any, AsyncIterator, Callable, Dict, Type
 
-from port_ocean.cache.base import CacheProvider
-from port_ocean.cache.disk import DiskCacheProvider
-from port_ocean.cache.memory import InMemoryCacheProvider
-from port_ocean.core.models import ProcessExecutionMode
-import port_ocean.helpers.metric.metric
-
-from fastapi import FastAPI, APIRouter
-
+from fastapi import APIRouter, FastAPI
 from loguru import logger
 from pydantic import BaseModel
 from starlette.types import Receive, Scope, Send
 
+import port_ocean.helpers.metric.metric
+from port_ocean.cache.base import CacheProvider
+from port_ocean.cache.disk import DiskCacheProvider
+from port_ocean.cache.memory import InMemoryCacheProvider
 from port_ocean.clients.port.client import PortClient
 from port_ocean.config.settings import (
     IntegrationConfiguration,
@@ -26,16 +23,17 @@
     ocean,
 )
 from port_ocean.core.handlers.resync_state_updater import ResyncStateUpdater
+from port_ocean.core.handlers.webhook.processor_manager import (
+    LiveEventsProcessorManager,
+)
 from port_ocean.core.integrations.base import BaseIntegration
+from port_ocean.core.models import ProcessExecutionMode
 from port_ocean.log.sensetive import sensitive_log_filter
 from port_ocean.middlewares import request_handler
 from port_ocean.utils.misc import IntegrationStateStatus
 from port_ocean.utils.repeat import repeat_every
 from port_ocean.utils.signal import signal_handler
 from port_ocean.version import __integration_version__
-from port_ocean.core.handlers.webhook.processor_manager import (
-    LiveEventsProcessorManager,
-)
 
 
 class Ocean:
@@ -69,6 +67,7 @@ def __init__(
             integration_identifier=self.config.integration.identifier,
             integration_type=self.config.integration.type,
             integration_version=__integration_version__,
+            ingest_url=self.config.port.ingest_url,
         )
         self.cache_provider: CacheProvider = self._get_caching_provider()
         self.process_execution_mode: ProcessExecutionMode = (

port_ocean/tests/core/conftest.py

@@ -96,7 +96,13 @@ async def post(url: str, *args: Any, **kwargs: Any) -> Response:
 @pytest.fixture
 def mock_port_client(mock_http_client: MagicMock) -> PortClient:
     mock_port_client = PortClient(
-        MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
     )
     mock_port_client.auth = AsyncMock()
     mock_port_client.auth.headers = AsyncMock(

port_ocean/tests/core/handlers/mixins/test_live_events.py

@@ -1,7 +1,9 @@
 from typing import Any
-from httpx import Response
-import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from httpx import Response
+
 from port_ocean.clients.port.client import PortClient
 from port_ocean.clients.port.types import UserAgentType
 from port_ocean.context.ocean import PortOceanContext
@@ -11,8 +13,6 @@
 from port_ocean.core.handlers.entity_processor.jq_entity_processor import (
     JQEntityProcessor,
 )
-from port_ocean.core.handlers.webhook.webhook_event import WebhookEventRawResults
-from port_ocean.core.integrations.mixins.live_events import LiveEventsMixin
 from port_ocean.core.handlers.port_app_config.models import (
     EntityMapping,
     MappingsConfig,
@@ -21,6 +21,8 @@
     ResourceConfig,
     Selector,
 )
+from port_ocean.core.handlers.webhook.webhook_event import WebhookEventRawResults
+from port_ocean.core.integrations.mixins.live_events import LiveEventsMixin
 from port_ocean.core.models import Entity
 from port_ocean.core.ocean_types import CalculationResult, EntitySelectorDiff
 from port_ocean.ocean import Ocean
@@ -253,7 +255,13 @@ def mock_port_app_config_handler(
 @pytest.fixture
 def mock_port_client(mock_http_client: MagicMock) -> PortClient:
     mock_port_client = PortClient(
-        MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
     )
     mock_port_client.auth = AsyncMock()
     mock_port_client.auth.headers = AsyncMock(
@@ -340,10 +348,11 @@ async def test_parse_raw_event_results_to_entities_creation(
         calculation_result
     )
 
-    entities_to_create, entities_to_delete = (
-        await mock_live_events_mixin._parse_raw_event_results_to_entities(
-            [one_webhook_event_raw_results_for_creation]
-        )
+    (
+        entities_to_create,
+        entities_to_delete,
+    ) = await mock_live_events_mixin._parse_raw_event_results_to_entities(
+        [one_webhook_event_raw_results_for_creation]
     )
 
     assert entities_to_create == [entity]
@@ -367,10 +376,11 @@ async def test_parse_raw_event_results_to_entities_deletion(
         calculation_result
     )
 
-    entities_to_create, entities_to_delete = (
-        await mock_live_events_mixin._parse_raw_event_results_to_entities(
-            [one_webhook_event_raw_results_for_deletion]
-        )
+    (
+        entities_to_create,
+        entities_to_delete,
+    ) = await mock_live_events_mixin._parse_raw_event_results_to_entities(
+        [one_webhook_event_raw_results_for_deletion]
     )
 
     assert entities_to_create == []

port_ocean/tests/core/handlers/webhook/test_processor_manager.py

@@ -1,32 +1,16 @@
-import pytest
-from port_ocean.core.handlers.webhook.processor_manager import (
-    LiveEventsProcessorManager,
-)
-from port_ocean.core.handlers.webhook.abstract_webhook_processor import (
-    AbstractWebhookProcessor,
-)
-from port_ocean.core.handlers.webhook.webhook_event import (
-    EventHeaders,
-    WebhookEvent,
-    WebhookEventRawResults,
-    EventPayload,
-)
-from fastapi import APIRouter
-from port_ocean.core.integrations.mixins.handler import HandlerMixin
-from port_ocean.utils.signal import SignalHandler
-from typing import Dict, Any
 import asyncio
+from typing import Any, Dict
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import APIRouter, FastAPI
 from fastapi.testclient import TestClient
-from fastapi import FastAPI
-from port_ocean.context.ocean import PortOceanContext
-from unittest.mock import AsyncMock
-from port_ocean.context.event import EventContext, event_context, EventType
-from port_ocean.context.ocean import ocean
-from unittest.mock import MagicMock, patch
 from httpx import Response
-from port_ocean.clients.port.client import PortClient
+
 from port_ocean import Ocean
-from port_ocean.core.integrations.base import BaseIntegration
+from port_ocean.clients.port.client import PortClient
+from port_ocean.context.event import EventContext, EventType, event_context
+from port_ocean.context.ocean import PortOceanContext, ocean
 from port_ocean.core.handlers.port_app_config.models import (
     EntityMapping,
     MappingsConfig,
@@ -35,13 +19,28 @@
     ResourceConfig,
     Selector,
 )
+from port_ocean.core.handlers.queue import LocalQueue
+from port_ocean.core.handlers.webhook.abstract_webhook_processor import (
+    AbstractWebhookProcessor,
+)
+from port_ocean.core.handlers.webhook.processor_manager import (
+    LiveEventsProcessorManager,
+)
+from port_ocean.core.handlers.webhook.webhook_event import (
+    EventHeaders,
+    EventPayload,
+    WebhookEvent,
+    WebhookEventRawResults,
+)
+from port_ocean.core.integrations.base import BaseIntegration
+from port_ocean.core.integrations.mixins.handler import HandlerMixin
 from port_ocean.core.integrations.mixins.live_events import LiveEventsMixin
 from port_ocean.core.models import Entity
 from port_ocean.exceptions.webhook_processor import (
     RetryableError,
     WebhookEventNotSupportedError,
 )
-from port_ocean.core.handlers.queue import LocalQueue
+from port_ocean.utils.signal import SignalHandler
 
 
 class MockProcessor(AbstractWebhookProcessor):
@@ -275,7 +274,13 @@ async def post(url: str, *args: Any, **kwargs: Any) -> Response:
 @pytest.fixture
 def mock_port_client(mock_http_client: MagicMock) -> PortClient:
     mock_port_client = PortClient(
-        MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock(), MagicMock()
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
     )
     mock_port_client.auth = AsyncMock()
     mock_port_client.auth.headers = AsyncMock(

port_ocean/tests/helpers/port_client.py

@@ -18,4 +18,5 @@ def get_port_client_for_integration(
         integration_identifier=integration_identifier,
         integration_type=integration_type,
         integration_version=integration_version,
+        ingest_url="https://ingest.getport.io",
     )