refactor(infra): share permissive type-coercion helpers

packages/auth/src/pipefy_auth/discovery.py

@@ -6,6 +6,7 @@
 
 import httpx
 from pipefy_infra import security
+from pipefy_infra.coerce import optional_str
 
 from pipefy_auth import _http
 
@@ -108,8 +109,7 @@ def fetch_provider_metadata(
 
     authorization_endpoint = str(data["authorization_endpoint"])
     token_endpoint = str(data["token_endpoint"])
-    raw_end_session = data.get("end_session_endpoint")
-    end_session_endpoint = str(raw_end_session) if raw_end_session else None
+    end_session_endpoint = optional_str(data.get("end_session_endpoint"))
 
     endpoints: list[tuple[str, str]] = [
         ("authorization_endpoint", authorization_endpoint),

packages/auth/src/pipefy_auth/storage.py

@@ -18,6 +18,8 @@
 from typing import Literal
 from urllib.parse import urlparse
 
+from pipefy_infra.coerce import optional_int, optional_str
+
 _SERVICE = "pipefy"
 _KEYRING_FILENAME = "keyring.cfg"
 
@@ -110,10 +112,10 @@ def store_session(
         refresh_token=refresh_token,
         token_type=str(token_response.get("token_type") or "Bearer"),
         obtained_at=int(time.time()),
-        expires_in=_optional_int(token_response.get("expires_in")),
-        refresh_expires_in=_optional_int(token_response.get("refresh_expires_in")),
-        scope=_optional_str(token_response.get("scope")),
-        id_token=_optional_str(token_response.get("id_token")),
+        expires_in=optional_int(token_response.get("expires_in")),
+        refresh_expires_in=optional_int(token_response.get("refresh_expires_in")),
+        scope=optional_str(token_response.get("scope")),
+        id_token=optional_str(token_response.get("id_token")),
     )
     keyring.set_password(
         _SERVICE, keychain_key(issuer, client_id), json.dumps(asdict(session))
@@ -175,22 +177,6 @@ def keychain_backend_name() -> str:
     return backend.__class__.__name__
 
 
-def _optional_int(value: object) -> int | None:
-    if value is None:
-        return None
-    try:
-        return int(value)  # type: ignore[arg-type]
-    except (TypeError, ValueError):
-        return None
-
-
-def _optional_str(value: object) -> str | None:
-    if value is None:
-        return None
-    text = str(value).strip()
-    return text or None
-
-
 __all__ = [
     "SessionDeleteError",
     "StoredSession",

packages/infra/src/pipefy_infra/__init__.py

@@ -2,12 +2,13 @@
 
 The package root exposes only ``__version__``. Consumers import from the
 submodule that names the concern: :mod:`pipefy_infra.config` (on-disk
-configuration: path discovery + TOML source) and :mod:`pipefy_infra.security`
+configuration: path discovery + TOML source), :mod:`pipefy_infra.security`
 (SSRF defenses, imported as ``from pipefy_infra import security`` so every
-call site is greppable for audits). See each submodule docstring and
-``README.md`` for the detailed surface. The package sits at the bottom of
-the workspace dependency graph: stdlib + ``pydantic`` /
-``pydantic-settings`` only.
+call site is greppable for audits), and :mod:`pipefy_infra.coerce`
+(permissive type-coercion helpers for JSON/GraphQL response values). See
+each submodule docstring and ``README.md`` for the detailed surface. The
+package sits at the bottom of the workspace dependency graph: stdlib +
+``pydantic`` / ``pydantic-settings`` only.
 """
 
 from __future__ import annotations

packages/infra/src/pipefy_infra/coerce.py

@@ -0,0 +1,95 @@
+"""Permissive coercion helpers for values of uncertain shape.
+
+When a value crosses a boundary (deserialization, user input, third-party
+payload) it may arrive as a string, int, float, or wrong shape entirely.
+These helpers normalize the common cases without raising, and explicitly
+reject misshapen types (booleans, bytes) that Python's built-in conversions
+would silently accept:
+
+- :func:`optional_int` / :func:`optional_str` / :func:`optional_float` collapse
+  ``None`` and any unparseable or wrong-shape value to ``None``. Use when
+  absence and malformed-input are handled identically by the caller.
+- :func:`try_int` returns the input unchanged if it can't be parsed as int.
+  Use when a non-numeric input is itself a meaningful value to preserve.
+
+Booleans are rejected (or passed through, for ``try_int``) despite ``bool``
+being a subclass of ``int``: a bool in a numeric- or string-shaped slot is
+a shape signal, not the value ``1`` / ``0`` / ``"True"``. Bytes are likewise
+rejected because ``str(b"x") == "b'x'"`` (the repr) is almost never what a
+caller wants.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+def optional_int(value: object) -> int | None:
+    """Return ``int(value)``, or ``None`` for None/bool/bytes/unparseable input.
+
+    Floats truncate (``42.7 -> 42``); non-finite floats and overflowing values
+    return ``None``. Numeric strings (including signed and PEP 515 underscored
+    forms) parse via Python's built-in ``int()``.
+    """
+    if value is None or isinstance(value, (bool, bytes, bytearray)):
+        return None
+    try:
+        return int(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError, OverflowError):
+        return None
+
+
+def optional_str(value: object) -> str | None:
+    """Return ``str(value).strip()``, or ``None`` for None/bool/bytes/empty input.
+
+    Booleans return ``None`` (a bool in a string-shaped slot is a shape error,
+    not the literal ``'True'`` / ``'False'``). Bytes return ``None`` to avoid
+    producing the bytes-repr form. Numeric inputs are stringified
+    (``optional_str(42) == '42'``) for the case where a numeric id needs
+    comparing against a string token.
+    """
+    if value is None or isinstance(value, (bool, bytes, bytearray)):
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def optional_float(value: object) -> float | None:
+    """Return ``float(value)``, or ``None`` for None/bool/bytes/unparseable input.
+
+    Non-finite floats (``inf``, ``-inf``, ``nan``) return ``None`` so callers
+    can treat them like missing data rather than threading edge-case checks
+    through every comparison.
+    """
+    if value is None or isinstance(value, (bool, bytes, bytearray)):
+        return None
+    try:
+        result = float(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return None
+    if not math.isfinite(result):
+        return None
+    return result
+
+
+def try_int(value: T) -> T | int:
+    """Return ``int(value)``, or ``value`` unchanged if conversion fails.
+
+    Distinct from :func:`optional_int` in that a non-numeric input is preserved
+    rather than collapsed to ``None``. Use when the caller wants to upgrade a
+    numeric form to an int but leave non-numeric values alone. Booleans and
+    bytes pass through unchanged (the misshapen-type rejection that
+    :func:`optional_int` uses ``None`` for).
+    """
+    if isinstance(value, (bool, bytes, bytearray)):
+        return value
+    try:
+        return int(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError, OverflowError):
+        return value
+
+
+__all__ = ["optional_float", "optional_int", "optional_str", "try_int"]

packages/infra/tests/test_coerce.py

@@ -0,0 +1,108 @@
+"""Unit tests for ``pipefy_infra.coerce`` type-coercion helpers."""
+
+from __future__ import annotations
+
+import pytest
+
+from pipefy_infra.coerce import optional_float, optional_int, optional_str, try_int
+
+
+class TestOptionalInt:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [
+            (None, None),
+            (0, 0),
+            (42, 42),
+            ("42", 42),
+            ("  42  ", 42),
+            ("-5", -5),
+            ("1_000", 1000),
+            (42.7, 42),
+            (-0.9, 0),
+            ("", None),
+            ("abc", None),
+            ([], None),
+            ({}, None),
+            (float("inf"), None),
+            (float("nan"), None),
+            (True, None),
+            (False, None),
+            (b"42", None),
+            (bytearray(b"42"), None),
+        ],
+    )
+    def test_cases(self, value: object, expected: int | None) -> None:
+        assert optional_int(value) == expected
+
+
+class TestOptionalStr:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [
+            (None, None),
+            ("", None),
+            ("   ", None),
+            ("hello", "hello"),
+            ("  hello  ", "hello"),
+            (42, "42"),
+            (-5, "-5"),
+            (3.14, "3.14"),
+            (True, None),
+            (False, None),
+            (b"hello", None),
+            (bytearray(b"hello"), None),
+        ],
+    )
+    def test_cases(self, value: object, expected: str | None) -> None:
+        assert optional_str(value) == expected
+
+
+class TestOptionalFloat:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [
+            (None, None),
+            (0, 0.0),
+            (42, 42.0),
+            (42.7, 42.7),
+            ("42", 42.0),
+            ("3.14", 3.14),
+            ("  3.14  ", 3.14),
+            ("", None),
+            ("abc", None),
+            (float("inf"), None),
+            (float("-inf"), None),
+            (float("nan"), None),
+            (True, None),
+            (False, None),
+            (b"3.14", None),
+        ],
+    )
+    def test_cases(self, value: object, expected: float | None) -> None:
+        assert optional_float(value) == expected
+
+
+class TestTryInt:
+    @pytest.mark.parametrize(
+        ("value", "expected"),
+        [
+            ("42", 42),
+            ("-5", -5),
+            ("1_000", 1000),
+            ("abc", "abc"),
+            ("", ""),
+            (7, 7),
+            (3.9, 3),
+            (None, None),
+            (True, True),
+            (False, False),
+            (b"42", b"42"),
+        ],
+    )
+    def test_cases(self, value: object, expected: object) -> None:
+        assert try_int(value) == expected
+
+    def test_inf_preserved(self) -> None:
+        value = float("inf")
+        assert try_int(value) is value

packages/sdk/src/pipefy_sdk/services/observability_export_csv.py

@@ -11,6 +11,7 @@
 import httpx
 from openpyxl import load_workbook
 from pipefy_infra import security
+from pipefy_infra.coerce import optional_int
 
 _ALLOWED_HOST_SUFFIX: Final[str] = ".pipefy.com"
 
@@ -125,17 +126,11 @@ async def stream_bytes(url: str, *, max_bytes: int) -> AsyncIterator[bytes]:
                     continue
 
                 response.raise_for_status()
-                cl = response.headers.get("content-length")
-                if cl is not None:
-                    try:
-                        declared = int(cl)
-                    except ValueError:
-                        pass
-                    else:
-                        if declared > max_bytes:
-                            raise ValueError(
-                                f"Export file exceeds max_download_bytes ({max_bytes} bytes)."
-                            )
+                declared = optional_int(response.headers.get("content-length"))
+                if declared is not None and declared > max_bytes:
+                    raise ValueError(
+                        f"Export file exceeds max_download_bytes ({max_bytes} bytes)."
+                    )
                 total = 0
                 async for chunk in response.aiter_bytes():
                     total += len(chunk)

packages/sdk/src/pipefy_sdk/services/pipe_config_service.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from httpx import Auth
+from pipefy_infra.coerce import optional_str
 
 from pipefy_sdk.base_client import BasePipefyClient
 from pipefy_sdk.queries.pipe_config_queries import (
@@ -208,10 +209,9 @@ def _field_rows_matching_token(
         for field in fields:
             if not isinstance(field, dict):
                 continue
-            fid = str(field.get("id", "")).strip()
-            iid = field.get("internal_id")
-            iid_str = str(iid).strip() if iid is not None else ""
-            uuid_str = str(field.get("uuid") or "").strip()
+            fid = optional_str(field.get("id")) or ""
+            iid_str = optional_str(field.get("internal_id")) or ""
+            uuid_str = optional_str(field.get("uuid")) or ""
             if (
                 tok == fid
                 or (iid_str and tok == iid_str)

packages/sdk/src/pipefy_sdk/services/pipe_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from httpx import Auth
+from pipefy_infra.coerce import optional_int
 from rapidfuzz import fuzz
 
 from pipefy_sdk.base_client import BasePipefyClient
@@ -25,16 +26,6 @@ def _clamp_max_pipes_per_org(value: int) -> int:
     return max(SEARCH_PIPES_MAX_PER_ORG_MIN, min(SEARCH_PIPES_MAX_PER_ORG_CAP, value))
 
 
-def _coerce_org_pipes_count(raw: object) -> int | None:
-    """Parse GraphQL ``Organization.pipesCount`` for truncation checks."""
-    if raw is None:
-        return None
-    try:
-        return int(raw)
-    except (TypeError, ValueError):
-        return None
-
-
 class PipeService(BasePipefyClient):
     """Service for Pipe-related operations."""
 
@@ -156,7 +147,7 @@ async def search_pipes(
             organizations: list[dict] = []
             for org in raw_orgs:
                 pipes = list(org.get("pipes") or [])
-                pipes_count = _coerce_org_pipes_count(org.get("pipesCount"))
+                pipes_count = optional_int(org.get("pipesCount"))
                 len_p = len(pipes)
                 over_cap = len_p > per_org_cap
                 fewer_than_reported_total = (