Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions changes/vercel/deadline.feature.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Expose `get_deadline()` for reading the current Function invocation deadline.
6 changes: 6 additions & 0 deletions src/vercel/functions/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ Functions.
from vercel.functions import (
AsyncRuntimeCache,
geolocation,
get_deadline,
get_env,
ip_address,
set_headers,
Expand All @@ -19,20 +20,25 @@ async def handler(request):
wait_until(record_request_analytics(request))

env = get_env()
deadline = get_deadline()
cache = AsyncRuntimeCache(namespace="api")
await cache.set("last_region", env.VERCEL_REGION, {"ttl": 60})

return {
"ip": ip_address(request),
"geo": geolocation(request),
"region": env.VERCEL_REGION,
"deadline": deadline.isoformat() if deadline else None,
}
```

Exports include environment helpers from `vercel.env`, header and geolocation
helpers from `vercel.headers`, cache clients from `vercel.cache`, and
`wait_until()` for work that should finish after the response is sent.

`get_deadline()` returns the current invocation deadline as a timezone-aware UTC
`datetime`, or `None` when the runtime does not provide one.

`wait_until()` is not a durable task queue. Its work must finish within the
Function's configured maximum duration, and it is not retried if the
invocation terminates. It accepts awaitables only; run synchronous work with
Expand Down
2 changes: 2 additions & 0 deletions src/vercel/functions/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from ..cache import AsyncRuntimeCache, RuntimeCache, get_cache
from ..env import Env, get_env
from ..headers import Geo, geolocation, get_headers, ip_address, set_headers
from .deadline import get_deadline
from .wait_until import wait_until

__all__ = [
Expand All @@ -15,4 +16,5 @@
"RuntimeCache",
"AsyncRuntimeCache",
"wait_until",
"get_deadline",
]
24 changes: 24 additions & 0 deletions src/vercel/functions/deadline.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
from __future__ import annotations

from datetime import datetime
from importlib import import_module


def get_deadline() -> datetime | None:
"""Return the current Function invocation deadline, if available."""
try:
runtime = import_module("vercel_runtime")
except ModuleNotFoundError as exc:
# Only the runtime itself being absent means "not on Vercel". A
# missing transitive dependency is a broken install and must surface.
if exc.name != "vercel_runtime":
raise
return None

accessor = getattr(runtime, "get_deadline", None)
if not callable(accessor):
# Older runtime without deadline support.
return None

value = accessor()
return value if isinstance(value, datetime) else None
40 changes: 40 additions & 0 deletions src/vercel/tests/unit/test_functions_deadline.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
from __future__ import annotations

import sys
from datetime import datetime, timezone
from types import SimpleNamespace

import pytest

from vercel.functions import get_deadline


def test_get_deadline_reads_runtime_context(monkeypatch) -> None:
expected = datetime(2026, 8, 18, 12, 30, tzinfo=timezone.utc)
runtime = SimpleNamespace(get_deadline=lambda: expected)
monkeypatch.setitem(sys.modules, "vercel_runtime", runtime)

assert get_deadline() == expected


def test_get_deadline_returns_none_outside_vercel(monkeypatch) -> None:
monkeypatch.delitem(sys.modules, "vercel_runtime", raising=False)

assert get_deadline() is None


def test_get_deadline_returns_none_without_runtime_support(monkeypatch) -> None:
monkeypatch.setitem(sys.modules, "vercel_runtime", SimpleNamespace())

assert get_deadline() is None


def test_get_deadline_propagates_runtime_bugs(monkeypatch) -> None:
def raise_error() -> None:
raise ValueError("invalid deadline")

runtime = SimpleNamespace(get_deadline=raise_error)
monkeypatch.setitem(sys.modules, "vercel_runtime", runtime)

with pytest.raises(ValueError, match="invalid deadline"):
get_deadline()