Revert "stop-gap concurrency control solution: separate on CDF project"

This reverts commit 56de84d.

Author: haakonvt

cognite/client/_api/data_modeling/instances.py

@@ -170,6 +170,7 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
         super().__init__(config, api_version, cognite_client)
         self._AGGREGATE_LIMIT = 1000
         self._SEARCH_LIMIT = 1000
+        self.__dm_semaphore = get_global_data_modeling_semaphore()
 
         self._warn_on_alpha_debug_settings = FeaturePreviewWarning(
             api_maturity="alpha",
@@ -297,7 +298,7 @@ async def __call__(
                 filter=filter.dump(camel_case_property=False) if isinstance(filter, Filter) else filter,
                 other_params=other_params,
                 headers=headers,
-                semaphore=get_global_data_modeling_semaphore(self._config.project),
+                semaphore=self.__dm_semaphore,
             ):
                 yield item
             return
@@ -310,7 +311,7 @@ async def __call__(
             filter=filter.dump(camel_case_property=False) if isinstance(filter, Filter) else filter,
             other_params=other_params,
             headers=headers,
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
+            semaphore=self.__dm_semaphore,
         ):
             yield list_cls._load_raw_api_response([raw])
 
@@ -634,7 +635,7 @@ def _load_raw_api_response(cls, responses: list[dict[str, Any]]) -> _NodeOrEdgeL
             identifiers=identifiers,
             other_params=other_params,
             settings_forcing_raw_response_loading=[f"{include_typing=}"] if include_typing else None,
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
+            semaphore=self.__dm_semaphore,
         )
 
         return InstancesResult[T_Node, T_Edge](
@@ -722,7 +723,7 @@ async def delete(
                 identifiers,
                 wrap_ids=True,
                 returns_items=True,
-                semaphore=get_global_data_modeling_semaphore(self._config.project),
+                semaphore=self.__dm_semaphore,
             ),
         )
         node_ids = [NodeId.load(item) for item in deleted_instances if item["instanceType"] == "node"]
@@ -788,7 +789,7 @@ async def inspect(
             response = await self._post(
                 self._RESOURCE_PATH + "/inspect",
                 json={"items": chunk.as_dicts(), "inspectionOperations": inspect_operations},
-                semaphore=get_global_data_modeling_semaphore(self._config.project),
+                semaphore=self.__dm_semaphore,
             )
             items.extend(unpack_items(response))
 
@@ -1067,7 +1068,7 @@ async def apply(
             resource_cls=_NodeOrEdgeApplyResultAdapter,  # type: ignore[type-var]
             extra_body_fields=other_parameters,
             input_resource_cls=_NodeOrEdgeApplyAdapter,  # type: ignore[arg-type]
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
+            semaphore=self.__dm_semaphore,
         )
         return InstancesApplyResult(
             nodes=NodeApplyResultList([item for item in res if isinstance(item, NodeApplyResult)]),
@@ -1257,11 +1258,7 @@ async def search(
                     raise ValueError("nulls_first argument is not supported when sorting on instance search")
             body["sort"] = [self._dump_instance_sort(s) for s in sorts]
 
-        res = await self._post(
-            url_path=self._RESOURCE_PATH + "/search",
-            json=body,
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
-        )
+        res = await self._post(url_path=self._RESOURCE_PATH + "/search", json=body, semaphore=self.__dm_semaphore)
         result = res.json()
         return list_cls(
             [resource_cls._load(item) for item in result["items"]],  # type: ignore [misc]
@@ -1386,11 +1383,7 @@ async def aggregate(
         if target_units:
             body["targetUnits"] = [unit.dump(camel_case=True) for unit in target_units]
 
-        res = await self._post(
-            url_path=self._RESOURCE_PATH + "/aggregate",
-            json=body,
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
-        )
+        res = await self._post(url_path=self._RESOURCE_PATH + "/aggregate", json=body, semaphore=self.__dm_semaphore)
         result_list = InstanceAggregationResultList._load(res.json()["items"])
         if group_by is not None:
             return result_list
@@ -1499,11 +1492,7 @@ async def histogram(
         if target_units:
             body["targetUnits"] = [unit.dump(camel_case=True) for unit in target_units]
 
-        res = await self._post(
-            url_path=self._RESOURCE_PATH + "/aggregate",
-            json=body,
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
-        )
+        res = await self._post(url_path=self._RESOURCE_PATH + "/aggregate", json=body, semaphore=self.__dm_semaphore)
         if is_singleton:
             return HistogramValue.load(res.json()["items"][0]["aggregates"][0])
         else:
@@ -1662,10 +1651,7 @@ async def _query_or_sync(
                 headers = {"cdf-version": f"{self._config.api_subversion}-alpha"}
 
         response = await self._post(
-            url_path=self._RESOURCE_PATH + f"/{endpoint}",
-            json=body,
-            headers=headers,
-            semaphore=get_global_data_modeling_semaphore(self._config.project),
+            url_path=self._RESOURCE_PATH + f"/{endpoint}", json=body, headers=headers, semaphore=self.__dm_semaphore
         )
         json_payload = response.json()
         default_by_reference = query.instance_type_by_result_expression()
@@ -1865,7 +1851,7 @@ async def list(
                 other_params=other_params,
                 settings_forcing_raw_response_loading=settings_forcing_raw_response_loading,
                 headers=headers,
-                semaphore=get_global_data_modeling_semaphore(self._config.project),
+                semaphore=self.__dm_semaphore,
             ),
         )
 

cognite/client/_api/datapoints.py

@@ -112,7 +112,7 @@ async def _request_datapoints(self, payload: dict[str, Any]) -> Sequence[DataPoi
                     f"{self.dps_client._RESOURCE_PATH}/list",
                     json=payload,
                     headers={"accept": "application/protobuf"},
-                    semaphore=get_global_datapoints_semaphore(self.dps_client._config.project),
+                    semaphore=get_global_datapoints_semaphore(),
                 )
             ).content
         )
@@ -388,7 +388,7 @@ async def _queue_new_subtasks(
         self,
         futures_dct: dict[asyncio.Task, list[BaseDpsFetchSubtask]],
     ) -> None:
-        sem = get_global_datapoints_semaphore(self.dps_client._config.project)
+        sem = get_global_datapoints_semaphore()
 
         # This may seem silly (to ask the semaphore for unused capacity), but the logic is sound:
         # we want to combine subtasks into requests *as late as possible* for optimal chunking.
@@ -2222,7 +2222,7 @@ async def insert_dataframe(self, df: pd.DataFrame, dropna: bool = True) -> None:
         await self.insert_multiple(dps)  # type: ignore[arg-type]
 
     def _select_dps_fetch_strategy(self, queries: list[DatapointsQuery]) -> type[DpsFetchStrategy]:
-        semaphore = get_global_datapoints_semaphore(self._config.project)
+        semaphore = get_global_datapoints_semaphore()
 
         # We decide the fetching strategy based on how many time series the user has requested VS the
         # max concurrency we allow for datapoints requests. When the number of time series is small enough
@@ -2362,7 +2362,7 @@ async def _insert_datapoints(self, payload: list[dict[str, Any]]) -> None:
             url_path=self.dps_client._RESOURCE_PATH,
             json={"items": payload},
             headers=headers,
-            semaphore=get_global_datapoints_semaphore(self.dps_client._config.project),
+            semaphore=get_global_datapoints_semaphore(),
         )
         for dct in payload:
             dct["datapoints"].clear()

cognite/client/_http_client.py

@@ -29,7 +29,6 @@
 )
 from cognite.client.response import CogniteHTTPResponse
 from cognite.client.utils._concurrency import get_global_semaphore
-from cognite.client.utils._url import extract_project_from_url
 
 logger = logging.getLogger(__name__)
 
@@ -266,7 +265,7 @@ async def _with_retry(
         if semaphore is None:
             # By default, we run with a semaphore decided by user settings of 'max_workers' in 'global_config'.
             # Since the user can run any number of SDK tasks concurrently, this needs to be global:
-            semaphore = get_global_semaphore(extract_project_from_url(url))
+            semaphore = get_global_semaphore()
 
         is_auto_retryable = False
         retry_tracker = RetryTracker(url, self.config)

cognite/client/_sync_api/data_modeling/instances.py

@@ -1,6 +1,6 @@
 """
 ===============================================================================
-9da25b05ee67f4f54dca5c516c7b71ec
+f96b1d5b543aefd8aff5e1e8a645a614
 This file is auto-generated from the Async API modules, - do not edit manually!
 ===============================================================================
 """

cognite/client/_sync_api/datapoints.py

@@ -1,6 +1,6 @@
 """
 ===============================================================================
-8876f83ce494d0a44fc1e354625554ca
+8b1cd40ee0ab9406c9edf75c089f78fa
 This file is auto-generated from the Async API modules, - do not edit manually!
 ===============================================================================
 """

cognite/client/utils/_concurrency.py

@@ -18,23 +18,22 @@
 from cognite.client.utils._auxiliary import no_op
 
 
-# We add the 'project' argument to make sure concurrency limits are applied per project:
 @cache
-def get_global_semaphore(project: str) -> asyncio.BoundedSemaphore:
+def get_global_semaphore() -> asyncio.BoundedSemaphore:
     from cognite.client import global_config
 
     return asyncio.BoundedSemaphore(global_config.max_workers)
 
 
 @cache
-def get_global_datapoints_semaphore(project: str) -> asyncio.BoundedSemaphore:
+def get_global_datapoints_semaphore() -> asyncio.BoundedSemaphore:
     from cognite.client import global_config
 
     return asyncio.BoundedSemaphore(global_config.max_workers)
 
 
 @cache
-def get_global_data_modeling_semaphore(project: str) -> asyncio.BoundedSemaphore:
+def get_global_data_modeling_semaphore() -> asyncio.BoundedSemaphore:
     return asyncio.BoundedSemaphore(2)
 
 

cognite/client/utils/_url.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import re
-import warnings
 from typing import TYPE_CHECKING, Any
 from urllib.parse import quote
 
@@ -65,17 +64,6 @@
 VALID_URL_PATTERN = re.compile(r"^https?://[a-z\d.:\-]+(?:/api/v1/projects/[^/]+)?((/[^\?]+)?(\?.+)?)")
 VALID_METHODS = {"GET", "POST", "PUT", "DELETE", "PATCH"}
 
-EXTRACT_PROJECT = re.compile(r"/api/v1/projects/([^/]+)")
-
-
-def extract_project_from_url(url: str, default: str = "") -> str:
-    # TODO: Stop-gap solution while we await the final concurrency limit implementation
-    try:
-        return EXTRACT_PROJECT.search(url).group(1)  # type: ignore[union-attr]
-    except AttributeError:
-        warnings.warn("No project found in URL", UserWarning)
-        return default
-
 
 def resolve_url(api_client: BasicAsyncAPIClient, method: str, url_path: str) -> tuple[bool, str]:
     if not url_path.startswith("/"):

tests/utils.py

@@ -271,7 +271,7 @@ def override_semaphore(new: int, target: Literal["basic", "datapoints", "data_mo
 
     # The new semaphore should now pick up the changed max_workers value:
     semaphore_get_fn.cache_clear()
-    sem = semaphore_get_fn("test_project")  # TODO: Stop-gap solution awaiting final concurrency limit implementation
+    sem = semaphore_get_fn()
     assert new == sem._value == sem._bound_value, "Semaphore didn't update according to overridden max_workers"  # type: ignore[attr-defined]
 
     try: