[Core] Port 14266 ocean to use new bulk inserts route (#1742)

### **User description**
# Description

What - release bulk upserts to all ocean integrations

Why - infra improvments

How - remove the is saas if

## 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


### Integration testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Resync finishes successfully
- [ ] If new resource kind is added or updated in the integration, add
example raw data, mapping and expected result to the `examples` folder
in the integration directory.
- [ ] If resource kind is updated, run the integration with the example
data and check if the expected result is achieved
- [ ] If new resource kind is added or updated, validate that
live-events for that resource are working as expected
- [ ] Docs PR link [here](#)

### Preflight checklist

- [ ] Handled rate limiting
- [ ] Handled pagination
- [ ] Implemented the code in async
- [ ] Support Multi account

## Screenshots

Include screenshots from your environment showing how the resources of
the integration will look.

## API Documentation

Provide links to the API documentation used for this integration.


___

### **PR Type**
Enhancement, Tests


___

### **Description**
- Refactored entity upsert batching to always use bulk route
  - Removed SaaS-specific conditional logic for bulk upserts
  - Unified batching logic for all environments
  - Improved error handling for batch upserts

- Removed `bulk_upserts_enabled` feature flag from configuration and
tests
  - Simplified configuration by eliminating obsolete flag
  - Updated tests to reflect new default behavior


___



### **Changes walkthrough** 📝
<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>entities.py</strong><dd><code>Always use bulk upsert
batching and unify error handling</code>&nbsp; </dd></summary>
<hr>

port_ocean/clients/port/mixins/entities.py

<li>Removed SaaS-only check for bulk upserts; now always uses bulk
<br>batching<br> <li> Unified batching and error handling logic for all
environments<br> <li> Improved fallback for 413 errors by retrying
entities individually<br> <li> Cleaned up legacy conditional branches


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1742/files#diff-1cd3dc7778947127c84baab24278808bee9c4ded96b1c320e28ece0f4d55ba7f">+50/-56</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>Remove bulk upserts
feature flag from configuration</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; </dd></summary>
<hr>

port_ocean/config/settings.py

<li>Removed <code>bulk_upserts_enabled</code> feature flag from
configuration<br> <li> Cleaned up related configuration code


</details>


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

</tr>
</table></td></tr><tr><td><strong>Tests</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>test_sync_raw.py</strong><dd><code>Remove obsolete bulk
upserts flag from tests</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

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

<li>Removed test configuration of <code>bulk_upserts_enabled</code>
flag<br> <li> Updated tests to match new default bulk upsert behavior


</details>


  </td>
<td><a
href="https://github.com/port-labs/ocean/pull/1742/files#diff-0530b15484dd5afce1e42bb43fe26cd56639f7302e997400c173a0df0a636eae">+0/-3</a>&nbsp;
&nbsp; &nbsp; </td>

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

___

> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about Qodo Merge
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>

---------

Co-authored-by: Tom Tankilevitch <59158507+Tankilevitch@users.noreply.github.com>

Author: itamar-smirra-port

CHANGELOG.md

@@ -7,6 +7,11 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+## 0.24.3 (2025-06-08)
+
+### Improvements
+- Using Port bulk upserts api in resyncs in all Ocean.
+
 ## 0.24.2 (2025-06-04)
 
 ### Improvements

port_ocean/clients/port/mixins/entities.py

@@ -353,66 +353,60 @@ async def upsert_entities_in_batches(
         entities_results: list[tuple[bool, Entity]] = []
         blueprint = entities[0].blueprint
 
-        if ocean.app.is_saas():
-            bulk_size = self.calculate_entities_batch_size(entities)
-            bulks = [
-                entities[i : i + bulk_size] for i in range(0, len(entities), bulk_size)
-            ]
-
-            bulk_results = await asyncio.gather(
-                *(
-                    self.upsert_entities_bulk(
-                        blueprint,
-                        bulk,
-                        request_options,
-                        user_agent_type,
-                        should_raise=should_raise,
-                    )
-                    for bulk in bulks
-                ),
-                return_exceptions=True,
-            )
+        bulk_size = self.calculate_entities_batch_size(entities)
+        bulks = [
+            entities[i : i + bulk_size] for i in range(0, len(entities), bulk_size)
+        ]
+
+        bulk_results = await asyncio.gather(
+            *(
+                self.upsert_entities_bulk(
+                    blueprint,
+                    bulk,
+                    request_options,
+                    user_agent_type,
+                    should_raise=should_raise,
+                )
+                for bulk in bulks
+            ),
+            return_exceptions=True,
+        )
 
-            for bulk, bulk_result in zip(bulks, bulk_results):
-                if isinstance(bulk_result, httpx.HTTPStatusError) or isinstance(
-                    bulk_result, Exception
+        for bulk, bulk_result in zip(bulks, bulk_results):
+            if isinstance(bulk_result, httpx.HTTPStatusError) or isinstance(
+                bulk_result, Exception
+            ):
+                if should_raise:
+                    raise bulk_result
+                # If should_raise is False, retry batch in sequential order as a fallback only for 413 errors
+                if (
+                    isinstance(bulk_result, httpx.HTTPStatusError)
+                    and bulk_result.response.status_code == 413
                 ):
-                    if should_raise:
-                        raise bulk_result
-                    # If should_raise is False, retry batch in sequential order as a fallback only for 413 errors
+                    individual_upsert_results = (
+                        await self._upsert_entities_batch_individually(
+                            bulk, request_options, user_agent_type, should_raise
+                        )
+                    )
+                    entities_results.extend(individual_upsert_results)
+                else:
+                    # For other errors, mark all entities in the batch as failed
+                    for entity in bulk:
+                        failed_result: tuple[bool, Entity] = (
+                            False,
+                            self._reduce_entity(entity),
+                        )
+                        entities_results.append(failed_result)
+            elif isinstance(bulk_result, list):
+                for status, entity in bulk_result:
                     if (
-                        isinstance(bulk_result, httpx.HTTPStatusError)
-                        and bulk_result.response.status_code == 413
-                    ):
-                        individual_upsert_results = (
-                            await self._upsert_entities_batch_individually(
-                                bulk, request_options, user_agent_type, should_raise
-                            )
+                        status is not None
+                    ):  # when using the search identifier we might not have an actual identifier
+                        bulk_result_tuple: tuple[bool, Entity] = (
+                            bool(status),
+                            entity,
                         )
-                        entities_results.extend(individual_upsert_results)
-                    else:
-                        # For other errors, mark all entities in the batch as failed
-                        for entity in bulk:
-                            failed_result: tuple[bool, Entity] = (
-                                False,
-                                self._reduce_entity(entity),
-                            )
-                            entities_results.append(failed_result)
-                elif isinstance(bulk_result, list):
-                    for status, entity in bulk_result:
-                        if (
-                            status is not None
-                        ):  # when using the search identifier we might not have an actual identifier
-                            bulk_result_tuple: tuple[bool, Entity] = (
-                                bool(status),
-                                entity,
-                            )
-                            entities_results.append(bulk_result_tuple)
-        else:
-            individual_upsert_results = await self._upsert_entities_batch_individually(
-                entities, request_options, user_agent_type, should_raise
-            )
-            entities_results.extend(individual_upsert_results)
+                        entities_results.append(bulk_result_tuple)
 
         return entities_results
 

port_ocean/config/settings.py

@@ -108,7 +108,6 @@ class IntegrationConfiguration(BaseOceanSettings, extra=Extra.allow):
 
     upsert_entities_batch_max_length: int = 20
     upsert_entities_batch_max_size_in_bytes: int = 1024 * 1024
-    bulk_upserts_enabled: bool = False
 
     @validator("process_execution_mode")
     def validate_process_execution_mode(

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

@@ -98,7 +98,6 @@ async def test_sync_raw_mixin_self_dependency(
 ) -> None:
     mock_ocean.config.upsert_entities_batch_max_length = 20
     mock_ocean.config.upsert_entities_batch_max_size_in_bytes = 1024 * 1024
-    mock_ocean.config.bulk_upserts_enabled = True
 
     entities_params = [
         ("entity_1", "service", {"service": "entity_1"}, True),
@@ -219,7 +218,6 @@ async def test_sync_raw_mixin_circular_dependency(
 ) -> None:
     mock_ocean.config.upsert_entities_batch_max_length = 20
     mock_ocean.config.upsert_entities_batch_max_size_in_bytes = 1024 * 1024
-    mock_ocean.config.bulk_upserts_enabled = True
 
     entities_params = [
         ("entity_1", "service", {"service": "entity_2"}, True),
@@ -359,7 +357,6 @@ async def test_sync_raw_mixin_dependency(
 ) -> None:
     mock_ocean.config.upsert_entities_batch_max_length = 20
     mock_ocean.config.upsert_entities_batch_max_size_in_bytes = 1024 * 1024
-    mock_ocean.config.bulk_upserts_enabled = True
 
     entities_params = [
         ("entity_1", "service", {"service": "entity_3"}, True),

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "port-ocean"
-version = "0.24.2"
+version = "0.24.3"
 description = "Port Ocean is a CLI tool for managing your Port projects."
 readme = "README.md"
 homepage = "https://app.getport.io"