fix: runtime issue in notebooks when using python 3.14 (DM-3678) (#2560)

Author: haakonvt

CHANGELOG.md

@@ -99,7 +99,6 @@ As of 2025-08-29, changes are grouped as follows
 - **global_config**: New setting `follow_redirects` that controls whether or not to follow redirects. Defaults to `False`.
 - **global_config**: New setting `file_download_chunk_size` that allows you to override the chunk size for streaming file downloads. Defaults to `None` (auto).
 - **global_config**: New setting `file_upload_chunk_size` that allows you to override the chunk size for streaming file uploads.
-- **global_config**: New setting `event_loop`, allowing you to override the default event loop used by the SDK.
 - **global_config**: `proxies` have been replaced by `proxy` and follow httpx directly. See: [Proxies - HTTPX](https://www.python-httpx.org/advanced/proxies/)
 - **global_config**: `max_retry_backoff` default has been increased from 30 sec to 60 sec.
 - **global_config**: `max_connection_pool_size` default has been reduced from 50 to 20.

MIGRATION_GUIDE.md

@@ -63,13 +63,12 @@ Changes are grouped as follows:
 - For `class Transformation`, which used to have an async `run` method, this is now named `run_async` to unify the overall interface. The same applies to the `cancel` and `jobs` methods for the same class, and `update` and `wait` on `TransformationJob`.
 - **ClientConfig**:
   - `max_workers` has functionally been removed (just throws a warning). Concurrency is now controlled via `global_config.concurrency_settings`.
-    See the [Settings documentation](https://cognite-sdk-python.readthedocs-hosted.com/en/v8/settings.html#concurrency-settings) for details.
+    See the [Settings documentation](https://cognite-sdk-python.readthedocs-hosted.com/en/latest/settings.html#concurrency-settings) for details.
   - `timeout`: default has been increased from 30 sec to 60 sec
 - **global_config**:
   - New setting `follow_redirects` that controls whether or not to follow redirects. Defaults to `False`.
   - New setting `file_download_chunk_size` that allows you to override the chunk size for streaming file downloads. Defaults to `None` (auto).
   - New setting `file_upload_chunk_size` that allows you to override the chunk size for streaming file uploads.
-  - New setting `event_loop`, allowing you to override the default event loop used by the SDK
   - `proxies` have been replaced by `proxy` and follow httpx directly. See: [Proxies - HTTPX](https://www.python-httpx.org/advanced/proxies/)
   - `max_retry_backoff`: default has been increased from 30 sec to 60 sec
   - `max_connection_pool_size`: default has been reduced from 50 to 20

cognite/client/_http_client.py

@@ -40,8 +40,19 @@ def set_cookie(self, cookie: Cookie) -> None:
         pass
 
 
-@functools.cache
+# One httpx.AsyncClient per event loop to avoid sharing connections (and their
+# loop-bound asyncio primitives) across different loops. This matters when a sync
+# CogniteClient (background loop) and an AsyncCogniteClient (e.g. Jupyter's loop)
+# coexist in the same process:
+_global_async_httpx_clients: dict[asyncio.AbstractEventLoop, httpx.AsyncClient] = {}
+
+
 def get_global_async_httpx_client() -> httpx.AsyncClient:
+    loop = asyncio.get_running_loop()
+    try:
+        return _global_async_httpx_clients[loop]
+    except KeyError:
+        pass
     async_transport = httpx.AsyncHTTPTransport(
         proxy=global_config.proxy,
         retries=0,  # 'retries': The maximum number of retries when trying to establish a connection.
@@ -59,14 +70,15 @@ def get_global_async_httpx_client() -> httpx.AsyncClient:
             keepalive_expiry=5,  # copy httpx default
         ),
     )
-    return httpx.AsyncClient(
+    client = _global_async_httpx_clients[loop] = httpx.AsyncClient(
         transport=async_transport,
         follow_redirects=global_config.follow_redirects,
         cookies=NoCookiesPlease(),
         # Below should not be needed when we pass transport, but... :)
         proxy=global_config.proxy,
         verify=not global_config.disable_ssl,
     )
+    return client
 
 
 class AsyncHTTPClientWithRetryConfig:
@@ -196,7 +208,11 @@ def __init__(
     ) -> None:
         self.config = config
         self.refresh_auth_header = refresh_auth_header
-        self.httpx_async_client = httpx_async_client or get_global_async_httpx_client()
+        self._httpx_async_client = httpx_async_client
+
+    @property
+    def httpx_async_client(self) -> httpx.AsyncClient:
+        return self._httpx_async_client or get_global_async_httpx_client()
 
     async def request(
         self,

cognite/client/config.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import getpass
 import pprint
 import re
@@ -37,7 +36,7 @@ class GlobalConfig:
         max_workers (int): DEPRECATED: Use 'concurrency_settings' instead. Maximum number of concurrent API calls. Defaults to 5.
         concurrency_settings (ConcurrencySettings): Settings controlling the maximum number of concurrent API requests
             for different API categories (general, raw, data_modeling etc.). These settings are frozen after the
-            first API request is made. See https://cognite-sdk-python.readthedocs-hosted.com/en/v8/settings.html#concurrency-settings
+            first API request is made. See https://cognite-sdk-python.readthedocs-hosted.com/en/latest/settings.html#concurrency-settings
         follow_redirects (bool): Whether or not to follow redirects. Defaults to False.
         file_download_chunk_size (int | None): Specify the file chunk size for streaming file downloads. When not specified
             (default is None), the actual chunk size is determined by the underlying transport, which in turn is based on the
@@ -47,7 +46,6 @@ class GlobalConfig:
             translates to 65536 (64KiB chunks).
         silence_feature_preview_warnings (bool): Whether or not to silence warnings triggered by using alpha or beta
             features. Defaults to False.
-        event_loop (asyncio.AbstractEventLoop | None): Override the default event loop used by the SDK.
     """
 
     def __new__(cls) -> GlobalConfig:
@@ -77,7 +75,6 @@ def __init__(self) -> None:
         self.file_download_chunk_size: int | None = None
         self.file_upload_chunk_size: int | None = None
         self.silence_feature_preview_warnings: bool = False
-        self.event_loop: asyncio.AbstractEventLoop | None = None
 
     @property
     def max_workers(self) -> int:
@@ -88,7 +85,7 @@ def max_workers(self, value: int) -> None:
         warnings.warn(
             "'max_workers' is no longer in use in the SDK as of v8, and will be removed in the next major version. "
             "Use 'global_config.concurrency_settings' instead for fine-grained control. For more info: "
-            "https://cognite-sdk-python.readthedocs-hosted.com/en/v8/settings.html#concurrency-settings",
+            "https://cognite-sdk-python.readthedocs-hosted.com/en/latest/settings.html#concurrency-settings",
             FutureWarning,
             stacklevel=2,
         )

cognite/client/utils/_concurrency.py

@@ -7,7 +7,6 @@
 from abc import ABC, abstractmethod
 from collections import UserList
 from collections.abc import Callable, Coroutine
-from functools import cache
 from typing import (
     Any,
     Literal,
@@ -19,7 +18,7 @@
 from typing_extensions import assert_never
 
 from cognite.client._constants import _RUNNING_IN_BROWSER
-from cognite.client.exceptions import CogniteAPIError, CogniteDuplicatedError, CogniteImportError, CogniteNotFoundError
+from cognite.client.exceptions import CogniteAPIError, CogniteDuplicatedError, CogniteNotFoundError
 from cognite.client.utils._auxiliary import no_op
 
 
@@ -50,6 +49,7 @@ def __init__(
         self._read = read
         self._write = write
         self._delete = delete
+        self._semaphore_cache: dict[tuple[str, str, asyncio.AbstractEventLoop], asyncio.BoundedSemaphore] = {}
 
     @property
     def read(self) -> int:
@@ -79,7 +79,7 @@ def delete(self, value: int) -> None:
         self._delete = value
 
     @abstractmethod
-    def _semaphore_factory(self, operation: str, project: str) -> asyncio.BoundedSemaphore: ...
+    def _semaphore_factory(self, operation: Any, project: str) -> asyncio.BoundedSemaphore: ...
 
     @abstractmethod
     def __repr__(self) -> str: ...
@@ -97,24 +97,30 @@ class CRUDConcurrency(ConcurrencyConfig):
         delete (int): Maximum number of concurrent delete requests.
     """
 
-    @cache
     def _semaphore_factory(
         self, operation: Literal["read", "write", "delete"], project: str
     ) -> asyncio.BoundedSemaphore:
-        # We include 'project' in the cache, since concurrency limits should apply per-project
+        # We include 'project' in the cache key, since concurrency limits should apply per-project.
+        # We include the event loop because semaphores are bound to the loop they're first used on,
+        # so the sync client (background loop) and async client (e.g. Jupyter's loop) need separate instances.
+        key = (operation, project, asyncio.get_running_loop())
+        if key in self._semaphore_cache:
+            return self._semaphore_cache[key]
         from cognite.client import global_config
 
         global_config.concurrency_settings._freeze()  # Disallow any further changes
 
         match operation:
             case "read":
-                return asyncio.BoundedSemaphore(self.read)
+                sem = asyncio.BoundedSemaphore(self.read)
             case "write":
-                return asyncio.BoundedSemaphore(self.write)
+                sem = asyncio.BoundedSemaphore(self.write)
             case "delete":
-                return asyncio.BoundedSemaphore(self.delete)
+                sem = asyncio.BoundedSemaphore(self.delete)
             case _:
                 assert_never(operation)
+        self._semaphore_cache[key] = sem
+        return sem
 
     def __repr__(self) -> str:
         return f"Concurrency[{self.api_name}](read={self._read}, write={self._write}, delete={self._delete})"
@@ -178,29 +184,32 @@ def write_schema(self, value: int) -> None:
         self._check_frozen("write_schema")
         self._write_schema = value
 
-    @cache
     def _semaphore_factory(
         self, operation: Literal["read", "write", "delete", "search", "read_schema", "write_schema"], project: str
     ) -> asyncio.BoundedSemaphore:
-        # We include 'project' in the cache, since concurrency limits should apply per-project
+        key = (operation, project, asyncio.get_running_loop())
+        if key in self._semaphore_cache:
+            return self._semaphore_cache[key]
         from cognite.client import global_config
 
         global_config.concurrency_settings._freeze()  # Disallow any further changes
         match operation:
             case "read":
-                return asyncio.BoundedSemaphore(self.read)
+                sem = asyncio.BoundedSemaphore(self.read)
             case "write":
-                return asyncio.BoundedSemaphore(self.write)
+                sem = asyncio.BoundedSemaphore(self.write)
             case "delete":
-                return asyncio.BoundedSemaphore(self.delete)
+                sem = asyncio.BoundedSemaphore(self.delete)
             case "search":
-                return asyncio.BoundedSemaphore(self.search)
+                sem = asyncio.BoundedSemaphore(self.search)
             case "read_schema":
-                return asyncio.BoundedSemaphore(self.read_schema)
+                sem = asyncio.BoundedSemaphore(self.read_schema)
             case "write_schema":
-                return asyncio.BoundedSemaphore(self.write_schema)
+                sem = asyncio.BoundedSemaphore(self.write_schema)
             case _:
                 assert_never(operation)
+        self._semaphore_cache[key] = sem
+        return sem
 
     def __repr__(self) -> str:
         return (
@@ -216,7 +225,7 @@ class ConcurrencySettings:
     The total concurrency budget, i.e. the maximum number of concurrent requests in flight,
     is the sum of all categories (e.g. general) and operation types (e.g. read or write).
 
-    See: https://cognite-sdk-python.readthedocs-hosted.com/en/v8/settings.html#concurrency-settings
+    See: https://cognite-sdk-python.readthedocs-hosted.com/en/latest/settings.html#concurrency-settings
 
     Note:
         The settings apply on a per-project level, thus if you have multiple clients
@@ -258,7 +267,7 @@ def _check_frozen(self, name: str, api_name: str) -> None:
             raise RuntimeError(
                 f"Cannot modify '{api_name}.{name}' after concurrency settings have been used to create semaphores. "
                 "Concurrency settings must be configured before sending any API requests. "
-                "See: https://cognite-sdk-python.readthedocs-hosted.com/en/v8/settings.html#concurrency-settings"
+                "See: https://cognite-sdk-python.readthedocs-hosted.com/en/latest/settings.html#concurrency-settings"
             )
 
     def _freeze(self) -> None:
@@ -508,73 +517,20 @@ def _raise_specific_error(
 
 
 class EventLoopThreadExecutor(threading.Thread):
-    def __init__(self, loop: asyncio.AbstractEventLoop | None = None, daemon: bool = True) -> None:
+    def __init__(self, daemon: bool = True) -> None:
         super().__init__(name=type(self).__name__, daemon=daemon)
-        self._inside_jupyter = self._detect_jupyter()
-        self._event_loop = self._patch_loop_for_jupyter(loop) or asyncio.new_event_loop()
-
-    @staticmethod
-    def _detect_jupyter() -> bool:
-        try:
-            from IPython import get_ipython  # type: ignore [attr-defined]
-
-            return "IPKernelApp" in get_ipython().config
-        except Exception:
-            return False
-
-    def _patch_loop_for_jupyter(self, loop: asyncio.AbstractEventLoop | None) -> asyncio.AbstractEventLoop | None:
-        """
-        From the 'nest_asyncio' package: By design asyncio does not allow its event loop to be nested. This presents a
-        practical problem: When in an environment where the event loop is already running it's impossible to run
-        tasks and wait for the result.
-        """
-        if not self._inside_jupyter:
-            return loop
-
-        if loop is None:
-            try:
-                import nest_asyncio  # type: ignore [import-not-found]
-            except ImportError:
-                raise CogniteImportError(
-                    module="nest_asyncio",
-                    message="Inside Jupyter notebooks, the 'nest_asyncio' package is required if you want to use the "
-                    '"non-async" CogniteClient. This is because Jupyter already runs an event loop that we need '
-                    "to patch (`pip install nest_asyncio`). Alternatively, you can use the AsyncCogniteClient "
-                    "which does not require any extra packages, but requires the use of 'await', e.g.: "
-                    "`dps = await async_client.time_series.data.retrieve(...)`",
-                ) from None
-            try:
-                # Jupyter: reuse the already running loop but patch it:
-                loop = asyncio.get_running_loop()
-                nest_asyncio.apply(loop)
-                return loop
-            except RuntimeError:
-                return None  # this would be very unexpected
-        else:
-            warnings.warn(
-                RuntimeWarning(
-                    "Overriding the event loop is not recommended inside Jupyter notebooks "
-                    "since Jupyter already runs an event loop. Proceeding with the provided loop anyway, "
-                    "beware of potential issues."
-                )
-            )
-            return loop
+        self._event_loop = asyncio.new_event_loop()
 
     def run(self) -> None:
-        if not self._inside_jupyter:
-            asyncio.set_event_loop(self._event_loop)
-            self._event_loop.run_forever()
+        asyncio.set_event_loop(self._event_loop)
+        self._event_loop.run_forever()
 
     def stop(self) -> None:
-        if not self._inside_jupyter:
-            self._event_loop.call_soon_threadsafe(self._event_loop.stop)
-            self.join()
+        self._event_loop.call_soon_threadsafe(self._event_loop.stop)
+        self.join()
 
-    def run_coro(self, coro: Coroutine[_T, Any, _T], timeout: float | None = None) -> _T:
-        if self._inside_jupyter:
-            return asyncio.get_event_loop().run_until_complete(coro)
-        else:
-            return asyncio.run_coroutine_threadsafe(coro, self._event_loop).result(timeout)
+    def run_coro(self, coro: Coroutine[Any, Any, _T], timeout: float | None = None) -> _T:
+        return asyncio.run_coroutine_threadsafe(coro, self._event_loop).result(timeout)
 
 
 class _PyodideEventLoopExecutor:
@@ -609,13 +565,11 @@ def _get_event_loop_executor() -> EventLoopThreadExecutor:
         return _INTERNAL_EVENT_LOOP_THREAD_EXECUTOR_SINGLETON
     except NameError:
         # First time we need to initialize:
-        from cognite.client import global_config
-
         ex_cls = EventLoopThreadExecutor
         if _RUNNING_IN_BROWSER:
             ex_cls = cast(type[EventLoopThreadExecutor], _PyodideEventLoopExecutor)
 
-        executor = _INTERNAL_EVENT_LOOP_THREAD_EXECUTOR_SINGLETON = ex_cls(global_config.event_loop)
+        executor = _INTERNAL_EVENT_LOOP_THREAD_EXECUTOR_SINGLETON = ex_cls()
         executor.start()
         return executor
 

docs/source/settings.rst

@@ -168,6 +168,23 @@ Alternatively, you can toggle debug logging on or off dynamically by setting the
 
 Note: Large outgoing or incoming payloads will be truncated to 1000 characters in the logs to avoid overwhelming the log output.
 
+Custom event loop (e.g. uvloop)
+-------------------------------
+The SDK automatically creates and manages a background ``asyncio`` event loop for the synchronous ``CogniteClient``.
+By default this uses Python's built-in ``asyncio`` event loop, but you can substitute any compatible implementation
+(such as `uvloop <https://github.com/MagicStack/uvloop>`_) by installing a custom loop *policy* before the first API
+call is made. The background loop is created via ``asyncio.new_event_loop()``, which respects the active policy.
+
+.. code:: python
+
+    import uvloop
+
+    uvloop.install()  # equivalent to asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+
+    from cognite.client import CogniteClient
+
+    client = CogniteClient(...)  # background loop will now be a uvloop
+
 HTTP Request logging
 --------------------
 Internally this library uses the ``httpx`` library to perform network calls to the Cognite API service endpoints. For authentication and

mypy.ini

@@ -25,3 +25,6 @@ ignore_missing_imports = true
 
 [mypy-sympy.*]
 ignore_missing_imports = true
+
+[mypy-testbook.*]
+ignore_missing_imports = true