Skip to content

feat(fal_client): add a tags request option #1158

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dazip wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(fal_client): add a tags request option #1158

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 90 additions & 1 deletion projects/fal_client/src/fal_client/_headers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

from typing import Literal, Union, get_args, Optional, Any, Callable
import re
from typing import Literal, Mapping, Union, get_args, Optional, Any, Callable

from httpx import Headers

Expand Down Expand Up @@ -33,10 +34,24 @@ def _current_fal_app_request() -> Optional[Any]:
REQUEST_TIMEOUT_TYPE_HEADER = "X-Fal-Request-Timeout-Type"
RUNNER_HINT_HEADER = "X-Fal-Runner-Hint"
QUEUE_PRIORITY_HEADER = "X-Fal-Queue-Priority"
TAGS_HEADER = "X-Fal-Tags"

# Valid priority values
Priority = Literal["normal", "low"]

# Tag limits, mirroring the gateway's validator. The gateway fails open and
# drops pairs it rejects, so the client validates up front instead of silently
# sending tags that never make it into the account's usage breakdown.
MAX_TAG_PAIRS = 10
MAX_TAG_KEY_LENGTH = 64
MAX_TAG_VALUE_LENGTH = 256
MAX_TAGS_TOTAL_BYTES = 1024

# System-only tag namespace, rejected in caller input.
RESERVED_TAG_KEY_PREFIX = "fal."

_TAG_KEY_PATTERN = re.compile(r"^[a-z0-9._-]+$")


def add_timeout_header(timeout: Union[int, float], headers: dict[str, str]) -> None:
"""
Expand Down Expand Up @@ -78,6 +93,80 @@ def add_priority_header(priority: Priority, headers: dict[str, str]) -> None:
headers[QUEUE_PRIORITY_HEADER] = priority


def _is_valid_tag_value(value: str) -> bool:
# Printable ASCII (incl. space), excluding control chars and the "," separator.
return all(char.isascii() and char.isprintable() and char != "," for char in value)


def add_tags_header(tags: Mapping[str, str], headers: dict[str, str]) -> None:
"""
Validates the tags and adds the packed tags header to the headers dictionary.

Keys and values are trimmed, keys are lowercased, and the pairs are packed
into a single `key=value,key=value` header. An empty mapping adds no header.

Args:
tags: Tags to attach to the request, as a key to value mapping.
headers: Headers dictionary to add the tags header to.

Raises:
ValueError: If a key or value is invalid, or a tag limit is exceeded.
"""
if not tags:
return

if len(tags) > MAX_TAG_PAIRS:
raise ValueError(f"At most {MAX_TAG_PAIRS} tags are allowed, got {len(tags)}")

packed: dict[str, str] = {}
total_bytes = 0

for raw_key, raw_value in tags.items():
if not isinstance(raw_key, str) or not isinstance(raw_value, str):
raise ValueError(
f"Tag keys and values must be strings, got {raw_key!r}: {raw_value!r}"
)

key = raw_key.strip().lower()
value = raw_value.strip()

if not _TAG_KEY_PATTERN.match(key):
raise ValueError(
f"Tag key must be non-empty and match [a-z0-9._-], got '{raw_key}'"
)
if not _is_valid_tag_value(value):
raise ValueError(
f"Tag value must be printable ASCII without ',', got '{raw_value}'"
)
if len(key) > MAX_TAG_KEY_LENGTH:
raise ValueError(
f"Tag key must be at most {MAX_TAG_KEY_LENGTH} characters, "
f"got '{raw_key}'"
)
if len(value) > MAX_TAG_VALUE_LENGTH:
raise ValueError(
f"Tag value must be at most {MAX_TAG_VALUE_LENGTH} characters, "
f"got '{raw_value}'"
)
if key.startswith(RESERVED_TAG_KEY_PREFIX):
raise ValueError(
f"Tag keys starting with '{RESERVED_TAG_KEY_PREFIX}' are reserved, "
f"got '{raw_key}'"
)

# Both the pair count and this byte budget (key and value) are measured
# before duplicate keys collapse -- the same way the gateway counts them.
total_bytes += len(key) + len(value)
if total_bytes > MAX_TAGS_TOTAL_BYTES:
raise ValueError(
f"Tags must be at most {MAX_TAGS_TOTAL_BYTES} bytes of keys and values"
)

packed[key] = value # last-wins on duplicate keys

headers[TAGS_HEADER] = ",".join(f"{key}={value}" for key, value in packed.items())
Comment thread
dazip marked this conversation as resolved.


def add_fal_app_context_headers(headers: dict[str, str]) -> None:
if request := _current_fal_app_request():
if cdn_token := request.headers.get("x-fal-cdn-token"):
Expand Down
40 changes: 40 additions & 0 deletions projects/fal_client/src/fal_client/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
Awaitable,
Dict,
Iterator,
Mapping,
TYPE_CHECKING,
Optional,
Literal,
Expand Down Expand Up @@ -52,6 +53,7 @@
add_priority_header,
add_timeout_header,
add_hint_header,
add_tags_header,
add_fal_app_context_headers,
handle_response_headers,
REQUEST_TIMEOUT_TYPE_HEADER,
Expand Down Expand Up @@ -1646,6 +1648,7 @@ async def run(
timeout: Optional[Union[int, float]] = None,
start_timeout: Optional[Union[int, float]] = None,
hint: str | None = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
) -> AnyJSON:
"""Run an application with the given arguments (which will be JSON serialized). The path parameter can be used to
Expand All @@ -1657,6 +1660,8 @@ async def run(
client waits for a response. Defaults to the client's default_timeout.
start_timeout: Server-side request timeout in seconds. Limits total time spent
waiting before processing starts. Does not apply once the application begins processing.
tags: Tags to attach to the request, as a key to value mapping. Sent
as one packed X-Fal-Tags header; invalid or over-limit tags raise.
"""

client = await self._client
Expand All @@ -1673,6 +1678,9 @@ async def run(
if start_timeout is not None:
add_timeout_header(start_timeout, _headers)

if tags is not None:
add_tags_header(tags, _headers)

add_fal_app_context_headers(_headers)

response = await _async_maybe_retry_request(
Expand All @@ -1697,6 +1705,7 @@ async def submit(
hint: str | None = None,
webhook_url: str | None = None,
priority: Optional[Priority] = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
start_timeout: Optional[Union[int, float]] = None,
) -> AsyncRequestHandle:
Expand All @@ -1708,6 +1717,8 @@ async def submit(
start_timeout: Server-side request timeout in seconds. Limits total time spent
waiting before processing starts (includes queue wait, retries, and
routing). Does not apply once the application begins processing.
tags: Tags to attach to the request, as a key to value mapping. Sent
as one packed X-Fal-Tags header; invalid or over-limit tags raise.
"""

client = await self._client
Expand All @@ -1730,6 +1741,9 @@ async def submit(
if start_timeout is not None:
add_timeout_header(start_timeout, _headers)

if tags is not None:
add_tags_header(tags, _headers)

add_fal_app_context_headers(_headers)

response = await _async_maybe_retry_request(
Expand Down Expand Up @@ -1764,6 +1778,7 @@ async def subscribe(
on_enqueue: Optional[Callable[[str], None | Awaitable[None]]] = None,
on_queue_update: Optional[Callable[[Status], None | Awaitable[None]]] = None,
priority: Optional[Priority] = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
start_timeout: Optional[Union[int, float]] = None,
client_timeout: Optional[Union[int, float]] = None,
Expand Down Expand Up @@ -1799,6 +1814,7 @@ async def _do_subscribe() -> AnyJSON:
path=path,
hint=hint,
priority=priority,
tags=tags,
headers=headers,
start_timeout=start_timeout,
)
Expand Down Expand Up @@ -1862,6 +1878,7 @@ async def stream(
*,
path: str = "/stream",
timeout: float | None = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
) -> AsyncIterator[dict[str, Any]]:
"""Stream the output of an application with the given arguments (which will be JSON serialized). This is only supported
Expand All @@ -1877,6 +1894,10 @@ async def stream(
url += "/" + path.lstrip("/")

_headers: dict[str, str] = {**headers}

if tags is not None:
add_tags_header(tags, _headers)

add_fal_app_context_headers(_headers)

async with aconnect_sse(
Expand Down Expand Up @@ -2178,6 +2199,7 @@ def run(
timeout: Optional[Union[int, float]] = None,
start_timeout: Optional[Union[int, float]] = None,
hint: str | None = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
) -> AnyJSON:
"""Run an application with the given arguments (which will be JSON serialized).
Expand All @@ -2188,6 +2210,8 @@ def run(
client waits for a response. Defaults to the client's default_timeout.
start_timeout: Server-side request timeout in seconds. Limits total time spent
waiting before processing starts. Does not apply once the application begins processing.
tags: Tags to attach to the request, as a key to value mapping. Sent
as one packed X-Fal-Tags header; invalid or over-limit tags raise.
"""

url = RUN_URL_FORMAT + application
Expand All @@ -2201,6 +2225,9 @@ def run(
if start_timeout is not None:
add_timeout_header(start_timeout, _headers)

if tags is not None:
add_tags_header(tags, _headers)

add_fal_app_context_headers(_headers)

response = _maybe_retry_request(
Expand All @@ -2225,6 +2252,7 @@ def submit(
hint: str | None = None,
webhook_url: str | None = None,
priority: Optional[Priority] = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
start_timeout: Optional[Union[int, float]] = None,
) -> SyncRequestHandle:
Expand All @@ -2234,6 +2262,8 @@ def submit(
start_timeout: Server-side request timeout in seconds. Limits total time spent
waiting before processing starts (includes queue wait, retries, and
routing). Does not apply once the application begins processing.
tags: Tags to attach to the request, as a key to value mapping. Sent
as one packed X-Fal-Tags header; invalid or over-limit tags raise.
"""

url = QUEUE_URL_FORMAT + application
Expand All @@ -2254,6 +2284,9 @@ def submit(
if start_timeout is not None:
add_timeout_header(start_timeout, _headers)

if tags is not None:
add_tags_header(tags, _headers)

add_fal_app_context_headers(_headers)

response = _maybe_retry_request(
Expand Down Expand Up @@ -2288,6 +2321,7 @@ def subscribe(
on_enqueue: Optional[Callable[[str], None]] = None,
on_queue_update: Optional[Callable[[Status], None]] = None,
priority: Optional[Priority] = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
start_timeout: Optional[Union[int, float]] = None,
client_timeout: Optional[Union[int, float]] = None,
Expand Down Expand Up @@ -2323,6 +2357,7 @@ def _do_subscribe() -> AnyJSON:
path=path,
hint=hint,
priority=priority,
tags=tags,
headers=headers,
start_timeout=start_timeout,
)
Expand Down Expand Up @@ -2379,6 +2414,7 @@ def stream(
*,
path: str = "/stream",
timeout: float | None = None,
tags: Optional[Mapping[str, str]] = None,
headers: dict[str, str] = {},
) -> Iterator[dict[str, Any]]:
"""Stream the output of an application with the given arguments (which will be JSON serialized). This is only supported
Expand All @@ -2393,6 +2429,10 @@ def stream(
url += "/" + path.lstrip("/")

_headers: dict[str, str] = {**headers}

if tags is not None:
add_tags_header(tags, _headers)

add_fal_app_context_headers(_headers)

with connect_sse(
Expand Down
Loading
Loading