fix(client): resolve project id once per resource manager, never raise from get_trace_url

langfuse/_client/client.py

@@ -350,7 +350,6 @@ def __init__(
             or os.environ.get(LANGFUSE_RELEASE, None)
             or get_common_release_envs()
         )
-        self._project_id: Optional[str] = None
         if sample_rate is None:
             sample_rate = float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0))
         if not 0.0 <= sample_rate <= 1.0:
@@ -2408,15 +2407,22 @@ def get_current_observation_id(self) -> Optional[str]:
         return self._get_otel_span_id(current_otel_span) if current_otel_span else None
 
     def _get_project_id(self) -> Optional[str]:
-        """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys."""
-        if not self._project_id:
-            proj = self.api.projects.get()
-            if not proj.data or not proj.data[0].id:
-                return None
-
-            self._project_id = proj.data[0].id
+        """Fetch and return the current project id. Returns None if no project id is found for api keys.
+
+        Delegates to the resource manager, which caches the outcome. The cache
+        cannot live here: `get_client()` constructs a new `Langfuse` on every
+        call, so anything memoized on this object dies with it, and a caller
+        generating one URL per row would issue one blocking lookup per row. The
+        resource manager is a singleton keyed by public key, which is the
+        granularity a project id has.
+        """
+        if self._resources is None:
+            langfuse_logger.debug(
+                "Operation skipped: _get_project_id - Client is not initialized."
+            )
+            return None
 
-        return self._project_id
+        return self._resources.get_project_id()
 
     def get_trace_url(self, *, trace_id: Optional[str] = None) -> Optional[str]:
         """Get the URL to view a trace in the Langfuse UI.

langfuse/_client/resource_manager.py

@@ -275,6 +275,12 @@ def _initialize_instance(
         # Prompt cache
         self.prompt_cache = PromptCache()
 
+        # Project id cache. Resolved lazily on first use, at most once per
+        # instance -- see get_project_id.
+        self._project_id: Optional[str] = None
+        self._project_id_resolved = False
+        self._project_id_lock = threading.Lock()
+
         # Register shutdown handler
         atexit.register(self.shutdown)
 
@@ -399,6 +405,57 @@ def _init_consumer_threads(self) -> None:
         ingestion_consumer.start()
         self._ingestion_consumers.append(ingestion_consumer)
 
+    def get_project_id(self) -> Optional[str]:
+        """Return the project id for this instance's credentials, or None if unavailable.
+
+        The lookup is a blocking network request, so its outcome is resolved at
+        most once per instance -- including a negative outcome. A project id does
+        not change for a given key pair, so caching only the success meant a
+        failing lookup was repeated on every call: `Langfuse.get_trace_url` is
+        commonly called once per row when rendering links, which turned an auth
+        failure or an empty project list into one blocking request per row.
+
+        This lives on the resource manager rather than on `Langfuse` because
+        `get_client()` constructs a new `Langfuse` on every call; state cached on
+        that object does not outlive the call that created it. The singleton here
+        is keyed by public key, which is the granularity a project id has.
+
+        Failures are logged and swallowed rather than raised: the only caller is
+        URL generation, which is a convenience and must not surface an exception
+        into the caller's code path.
+        """
+        if self._project_id_resolved:
+            return self._project_id
+
+        with self._project_id_lock:
+            # Another thread may have resolved it while this one waited.
+            if not self._project_id_resolved:
+                self._project_id = self._fetch_project_id()
+                self._project_id_resolved = True
+
+            return self._project_id
+
+    def _fetch_project_id(self) -> Optional[str]:
+        """Fetch the project id for this instance's credentials over the API."""
+        try:
+            projects = self.api.projects.get()
+        except Exception as e:
+            langfuse_logger.warning(
+                f"Could not resolve project id: {e}. URL generation is disabled until the client is recreated."
+            )
+            return None
+
+        if not projects.data or not projects.data[0].id:
+            langfuse_logger.warning(
+                "Could not resolve project id: no project found for the configured API keys. "
+                "URL generation is disabled until the client is recreated."
+            )
+            return None
+
+        project_id: Optional[str] = projects.data[0].id
+
+        return project_id
+
     def _at_fork_reinit(self) -> None:
         """Reinitialize consumer threads after fork in child process.
 
@@ -421,6 +478,13 @@ def _at_fork_reinit(self) -> None:
         # even if this particular instance was already shut down.
         LangfuseResourceManager._lock = threading.RLock()
 
+        # Same hazard for the project id lock: a thread in the parent may have
+        # been holding it across the lookup at fork time, which would deadlock
+        # the first caller here. The cached value survives (a project id is not
+        # per-process); only the mutex is replaced, so a lookup interrupted by
+        # the fork is simply retried in this child.
+        self._project_id_lock = threading.Lock()
+
         if self._shutdown:
             return
 

tests/unit/test_trace_url.py

@@ -0,0 +1,187 @@
+"""Tests for project id resolution behind `get_trace_url`.
+
+The project id lookup is a blocking network request. `get_trace_url` is commonly
+called once per row when rendering links, so the lookup must be attempted at most
+once -- on failure as well as on success -- and it must never raise into the
+caller. "At most once" is per resource manager rather than per `Langfuse`, since
+`get_client()` returns a new `Langfuse` on every call.
+"""
+
+import threading
+from types import SimpleNamespace
+from typing import List, Optional
+from unittest.mock import Mock
+
+import pytest
+
+from langfuse import Langfuse
+from langfuse._client.get_client import get_client
+from langfuse._client.resource_manager import LangfuseResourceManager
+
+TRACE_ID = "1234567890abcdef1234567890abcdef"
+
+
+@pytest.fixture
+def client(monkeypatch):
+    """A client with fake credentials; `api` is replaced per test."""
+    with LangfuseResourceManager._lock:
+        LangfuseResourceManager._instances.clear()
+
+    monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-trace-url")
+    monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-trace-url")
+    monkeypatch.setenv("LANGFUSE_BASE_URL", "http://localhost:3000")
+
+    langfuse = Langfuse()
+
+    yield langfuse
+
+    with LangfuseResourceManager._lock:
+        LangfuseResourceManager._instances.clear()
+
+
+def _api_returning(project_id):
+    api = Mock()
+    api.projects.get.return_value = SimpleNamespace(
+        data=[SimpleNamespace(id=project_id)] if project_id else []
+    )
+    return api
+
+
+def test_project_id_is_fetched_once_on_success(client):
+    client.api = _api_returning("project-id")
+
+    first = client.get_trace_url(trace_id=TRACE_ID)
+    second = client.get_trace_url(trace_id=TRACE_ID)
+
+    assert first == f"http://localhost:3000/project/project-id/traces/{TRACE_ID}"
+    assert second == first
+    assert client.api.projects.get.call_count == 1
+
+
+def test_project_id_is_not_refetched_when_no_project_is_found(client):
+    client.api = _api_returning(None)
+
+    assert client.get_trace_url(trace_id=TRACE_ID) is None
+    assert client.get_trace_url(trace_id=TRACE_ID) is None
+    assert client.api.projects.get.call_count == 1
+
+
+def test_project_id_lookup_failure_does_not_raise_and_is_not_retried(client):
+    api = Mock()
+    api.projects.get.side_effect = RuntimeError("401 Unauthorized")
+    client.api = api
+
+    assert client.get_trace_url(trace_id=TRACE_ID) is None
+    assert client.get_trace_url(trace_id=TRACE_ID) is None
+    assert api.projects.get.call_count == 1
+
+
+def test_concurrent_callers_all_see_the_resolved_project_id(client):
+    """A caller must not observe the pre-resolution state of an in-flight lookup.
+
+    The lookup is slow enough here that every other thread arrives while the
+    first one is still waiting on the network. Each must block and then see the
+    resolved id -- not a `None` that a request in flight is about to fill in.
+    """
+    started = threading.Event()
+    release = threading.Event()
+
+    def slow_get():
+        started.set()
+        release.wait(timeout=5)
+        return SimpleNamespace(data=[SimpleNamespace(id="project-id")])
+
+    api = Mock()
+    api.projects.get.side_effect = slow_get
+    client.api = api
+
+    results: List[Optional[str]] = []
+    results_lock = threading.Lock()
+
+    def call():
+        url = client.get_trace_url(trace_id=TRACE_ID)
+        with results_lock:
+            results.append(url)
+
+    threads = [threading.Thread(target=call) for _ in range(8)]
+    for thread in threads:
+        thread.start()
+
+    assert started.wait(timeout=5), "the first lookup never started"
+    release.set()
+
+    for thread in threads:
+        thread.join(timeout=5)
+        assert not thread.is_alive()
+
+    expected = f"http://localhost:3000/project/project-id/traces/{TRACE_ID}"
+    assert results == [expected] * 8
+    assert api.projects.get.call_count == 1
+
+
+def test_no_project_id_lookup_without_a_trace_id(client):
+    client.api = _api_returning("project-id")
+
+    assert client.get_trace_url() is None
+    assert client.api.projects.get.call_count == 0
+
+
+def test_project_id_is_shared_across_client_instances(client):
+    """`get_client()` returns a new `Langfuse` per call; the cache must outlive it.
+
+    This is why the resolution lives on the resource manager. Were it on the
+    `Langfuse` object, rendering one link per row via `get_client()` would issue
+    one blocking lookup per row.
+    """
+    api = _api_returning("project-id")
+    client.api = api
+    expected = f"http://localhost:3000/project/project-id/traces/{TRACE_ID}"
+
+    assert client.get_trace_url(trace_id=TRACE_ID) == expected
+    assert api.projects.get.call_count == 1
+
+    for _ in range(3):
+        other = get_client()
+        assert other is not client
+        assert other.get_trace_url(trace_id=TRACE_ID) == expected
+
+    assert api.projects.get.call_count == 1
+
+
+def test_a_failed_lookup_is_not_retried_by_a_later_client_instance(client):
+    """The negative outcome has to outlive the instance too.
+
+    An auth failure is the case that produced a blocking request per rendered
+    row, so caching it only for the lifetime of one `Langfuse` object would
+    leave the original problem in place.
+    """
+    api = Mock()
+    api.projects.get.side_effect = RuntimeError("401 Unauthorized")
+    client.api = api
+
+    assert client.get_trace_url(trace_id=TRACE_ID) is None
+
+    for _ in range(3):
+        assert get_client().get_trace_url(trace_id=TRACE_ID) is None
+
+    assert api.projects.get.call_count == 1
+
+
+def test_fork_replaces_the_project_id_lock(client):
+    """A lock held at fork time would deadlock the child, so it is replaced.
+
+    `_at_fork_reinit` resets it before the `_shutdown` early return, for the same
+    reason the class lock is reset there: the child needs a usable lock whether
+    or not this instance was torn down. Setting `_shutdown` here stops the method
+    before it recreates HTTP clients and consumer threads, which this test does
+    not exercise.
+    """
+    resources = client._resources
+    held = resources._project_id_lock
+    held.acquire()  # a thread in the parent, mid-lookup when the fork happened
+    resources._shutdown = True
+
+    resources._at_fork_reinit()
+
+    assert resources._project_id_lock is not held
+    assert not resources._project_id_lock.locked()