fix(client): preserve URL components when appending endpoint paths

src/workos/_base_client.py

@@ -12,7 +12,7 @@
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
 from typing import Any, Dict, List, Optional, Sequence, Type, Union, cast, overload
-from urllib.parse import quote, urlencode
+from urllib.parse import quote, urlencode, urlsplit, urlunsplit
 
 from ._errors import (
     APIError,
@@ -101,7 +101,7 @@ def __init__(
             "WORKOS_BASE_URL", "https://api.workos.com"
         )
         # Ensure base_url has a trailing slash for backward compatibility
-        self._base_url = resolved_base_url.rstrip("/") + "/"
+        self._base_url = self._join_url(resolved_base_url, "")
         self._request_timeout = (
             request_timeout
             if request_timeout is not None
@@ -126,11 +126,24 @@ def build_url(
         self, path: Sequence[str], params: Optional[Dict[str, Any]] = None
     ) -> str:
         """Build a full URL with query parameters for redirect/authorization endpoints."""
-        base = self._base_url.rstrip("/")
-        url = f"{base}/{self._encode_path(path)}"
-        if params:
-            url = f"{url}?{urlencode(params)}"
-        return url
+        return self._join_url(
+            self._base_url, self._encode_path(path), urlencode(params) if params else ""
+        )
+
+    @staticmethod
+    def _join_url(base_url: str, path: str, query: str = "") -> str:
+        """Append an encoded endpoint to the URL path, preserving URL components."""
+        base = urlsplit(base_url)
+        combined_query = "&".join(part for part in (base.query, query) if part)
+        return urlunsplit(
+            (
+                base.scheme,
+                base.netloc,
+                f"{base.path.rstrip('/')}/{path}",
+                combined_query,
+                base.fragment,
+            )
+        )
 
     @staticmethod
     def _parse_retry_after(retry_after: Optional[str]) -> Optional[float]:
@@ -180,8 +193,8 @@ def _resolve_base_url(self, request_options: Optional[RequestOptions]) -> str:
         if request_options:
             base_url = request_options.get("base_url")
             if base_url:
-                return str(base_url).rstrip("/")
-        return self._base_url.rstrip("/")
+                return str(base_url)
+        return self._base_url
 
     @staticmethod
     def _encode_path(path: Sequence[str]) -> str:
@@ -282,9 +295,11 @@ def _build_request_url(
         params: Optional[Dict[str, Any]],
         request_options: Optional[RequestOptions],
     ) -> str:
-        url = f"{self._resolve_base_url(request_options)}/{self._encode_path(path)}"
-        query = self._encode_query(params)
-        return f"{url}?{query}" if query else url
+        return self._join_url(
+            self._resolve_base_url(request_options),
+            self._encode_path(path),
+            self._encode_query(params),
+        )
 
     def _require_api_key(self) -> str:
         if not self._api_key:

tests/test_base_url_components.py

@@ -0,0 +1,90 @@
+# @oagen-ignore-file
+"""Public-client URL routing through a proxy base with URL components."""
+
+from urllib.parse import parse_qs, urlsplit
+
+import httpx2
+import pytest
+
+from workos import AsyncWorkOSClient, WorkOSClient, RequestOptions
+
+
+BASES = [
+    "https://proxy.example.test/prefix",
+    "https://proxy.example.test/prefix/",
+    "https://proxy.example.test/prefix?route=a%2Fb&label=x+y",
+    "https://proxy.example.test/prefix/#section",
+    "https://proxy.example.test/prefix?route=a%2Fb&label=x+y#section",
+]
+
+
+@pytest.mark.parametrize("base", BASES)
+@pytest.mark.parametrize("override", [False, True])
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.asyncio
+async def test_unit_resource_requests_preserve_base_url_components(
+    base, override, mode
+):
+    seen = []
+
+    def respond(request):
+        seen.append(request.url)
+        return httpx2.Response(200, json={"data": [], "list_metadata": {}})
+
+    options: RequestOptions | None = {"base_url": base} if override else None
+    transport = httpx2.MockTransport(respond)
+    if mode == "sync":
+        with httpx2.Client(transport=transport) as http:
+            client = WorkOSClient(
+                api_key="sk_test_dummy",
+                client_id="client_test",
+                base_url=None if override else base,
+                http_client=http,
+            )
+            with client:
+                page = client.organizations.list_organizations(
+                    limit=2, request_options=options
+                )
+                assert page.data == []
+    else:
+        async with httpx2.AsyncClient(transport=transport) as http:
+            async_client = AsyncWorkOSClient(
+                api_key="sk_test_dummy",
+                client_id="client_test",
+                base_url=None if override else base,
+                http_client=http,
+            )
+            async with async_client:
+                async_page = await async_client.organizations.list_organizations(
+                    limit=2, request_options=options
+                )
+                assert async_page.data == []
+
+    assert len(seen) == 1
+    assert seen[0].path == "/prefix/organizations"
+    expected = parse_qs(urlsplit(base).query)
+    expected["limit"] = ["2"]
+    expected["order"] = ["desc"]
+    assert parse_qs(seen[0].query.decode()) == expected
+
+
+@pytest.mark.parametrize("base", BASES)
+@pytest.mark.parametrize("resource", ["sso", "user_management"])
+def test_unit_authorization_urls_preserve_base_url_components(base, resource):
+    with WorkOSClient(
+        api_key="sk_test_dummy", client_id="client_test", base_url=base
+    ) as client:
+        url = getattr(client, resource).get_authorization_url(
+            redirect_uri="https://app.example.test/callback", state="a/b?c"
+        )
+    parts = urlsplit(url)
+    assert (
+        parts.path
+        == f"/prefix/{'sso' if resource == 'sso' else 'user_management'}/authorize"
+    )
+    query = parse_qs(parts.query)
+    for key, value in parse_qs(urlsplit(base).query).items():
+        assert query[key] == value
+    assert query["redirect_uri"] == ["https://app.example.test/callback"]
+    assert query["state"] == ["a/b?c"]
+    assert parts.fragment == urlsplit(base).fragment