Skip to content

Commit 403ecdc

Browse files
authored
Support per-row path in named_credential_request_col (#158)
* Support per-row url in named_credential_request_col Add an optional url Column to named_credential_request_col / SparkNamedCredential.request_col so each row can call a different endpoint (e.g. per-entity REST GET lookups). A null row value falls back to request.url; omitting url keeps the existing expression unchanged. * Lint fixes * Take a per-row path suffix instead of a full url Rows may only select an endpoint under the template's Named Credential: the path is appended to request.url, with '/' inserted unless it starts with '/' or '?', so a row can never change the Named Credential name. A null or empty path uses request.url.
1 parent 5f34e3e commit 403ecdc

7 files changed

Lines changed: 220 additions & 16 deletions

File tree

‎pyproject.toml‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -87,10 +87,10 @@ warn_unused_configs = true
8787

8888
[tool.poetry]
8989
include = [
90-
{path = "src/datacustomcode/templates/**/*"},
91-
{path = "src/datacustomcode/config.yaml"}
90+
{ path = "src/datacustomcode/templates/**/*" },
91+
{ path = "src/datacustomcode/config.yaml" }
9292
]
93-
packages = [{include = "datacustomcode", from = "src"}]
93+
packages = [{ include = "datacustomcode", from = "src" }]
9494
version = "0.0.0"
9595

9696
[tool.poetry.build]
@@ -115,7 +115,7 @@ coverage = ">=7.0.0,<8.0.0"
115115
ipykernel = "^6.29.5"
116116
mypy = "*"
117117
pipreqs = "*"
118-
poetry-dynamic-versioning = {extras = ["plugin"], version = "^1.8.2"}
118+
poetry-dynamic-versioning = { extras = ["plugin"], version = "^1.8.2" }
119119
pre-commit = "*"
120120
pytest = "*"
121121
pytest-cov = "*"
@@ -127,7 +127,7 @@ types-requests = "*"
127127
[tool.poetry.plugins."poetry.plugin"]
128128

129129
[tool.poetry.requires-plugins]
130-
poetry-dynamic-versioning = {version = ">=1.0.0,<2.0.0", extras = ["plugin"]}
130+
poetry-dynamic-versioning = { version = ">=1.0.0,<2.0.0", extras = ["plugin"] }
131131

132132
[tool.poetry.scripts]
133133
datacustomcode = "datacustomcode.cli:cli"
@@ -152,7 +152,6 @@ testpaths = ["tests"]
152152
[tool.ruff]
153153
fix = true
154154
line-length = 88
155-
target-version = 'py310'
156155
lint.ignore = [
157156
# do not assign a lambda expression, use a def
158157
'E731',
@@ -211,6 +210,7 @@ lint.select = [
211210
'RUF',
212211
'S102'
213212
]
213+
target-version = 'py310'
214214

215215
[tool.setuptools.package_data]
216216
datacustomcode = ["templates/**/*", "config.yaml"]

‎src/datacustomcode/client.py‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -231,11 +231,12 @@ def einstein_predict_col(
231231
def named_credential_request_col(
232232
request: "HTTPRequest",
233233
body: Optional["Column"] = None,
234+
path: Optional["Column"] = None,
234235
) -> "Column":
235236
"""Build a Spark Column that makes one Named Credential callout per row.
236237
237-
The endpoint, method, and headers are fixed for the call (taken from
238-
``request``); only ``body`` varies per row. Use this instead of
238+
The method and headers are fixed for the call (taken from ``request``);
239+
``body`` and, optionally, ``path`` vary per row. Use this instead of
239240
:meth:`Client.named_credential_request` when the callout runs across a
240241
DataFrame so each row is dispatched independently rather than one-shot on
241242
the driver.
@@ -253,13 +254,20 @@ def named_credential_request_col(
253254
headers are applied to every row.
254255
body: Optional per-row ``Column`` holding the request body as a
255256
string (or null for no body).
257+
path: Optional per-row ``Column`` appended to ``request.url``, e.g.
258+
``concat(lit("?address="), col("address"))`` or ``lit("/v1/models")``.
259+
The callout always stays on ``request.url``'s Named Credential: a
260+
path not starting with ``/`` or ``?`` is joined with ``/``. A null
261+
or empty row value uses ``request.url`` as is.
256262
257263
Returns:
258264
A Spark ``Column`` of ``StructType`` with fields ``status``,
259265
``response``, ``error_code``, and ``error_message``.
260266
"""
261267
named_credential = Client()._get_spark_named_credential()
262-
return named_credential.request_col(request, body=body)
268+
if path is None:
269+
return named_credential.request_col(request, body=body)
270+
return named_credential.request_col(request, body=body, path=path)
263271

264272

265273
class DataCloudObjectType(Enum):

‎src/datacustomcode/named_credential/spark_base.py‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,18 +66,24 @@ def request_col(
6666
self,
6767
request: HTTPRequest,
6868
body: Optional["Column"] = None,
69+
path: Optional["Column"] = None,
6970
) -> "Column":
7071
"""Build a Spark ``Column`` that makes one external callout per row.
7172
72-
The endpoint, method, and headers are fixed for the call (taken from
73-
``request``); only ``body`` varies per row. Use this instead of
73+
The method and headers are fixed for the call (taken from ``request``);
74+
``body`` and, optionally, ``path`` vary per row. Use this instead of
7475
:meth:`request` when the callout runs across a DataFrame so each row is
7576
dispatched independently rather than one-shot on the driver.
7677
7778
Args:
7879
request: The callout template
7980
body: Optional per-row ``Column`` holding the request body as a
8081
string, sent verbatim (or null for no body).
82+
path: Optional per-row ``Column`` appended to ``request.url``
83+
(e.g. ``/v1/models`` or ``?id=42``) so each row can target a
84+
different endpoint of the same Named Credential. A path not
85+
starting with ``/`` or ``?`` is joined with ``/``; a null or
86+
empty row value uses ``request.url`` as is.
8187
8288
Returns:
8389
A ``Column`` yielding a struct

‎src/datacustomcode/named_credential/spark_default.py‎

Lines changed: 40 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ def request_col(
7979
self,
8080
request: "HTTPRequest",
8181
body: Optional["Column"] = None,
82+
path: Optional["Column"] = None,
8283
) -> "Column":
8384
"""Per-row callout via a client-side Spark UDF.
8485
@@ -114,11 +115,46 @@ def request_col(
114115
]
115116
)
116117

117-
def _callout(body_str: Optional[str]) -> Dict[str, Any]:
118-
return _invoke_callout_as_struct(self._named_credential, request, body_str)
119-
120118
body_col = body if body is not None else lit(None).cast(StringType())
121-
return udf(_callout, result_schema)(body_col)
119+
120+
if path is None:
121+
122+
def _callout(body_str: Optional[str]) -> Dict[str, Any]:
123+
return _invoke_callout_as_struct(
124+
self._named_credential, request, body_str
125+
)
126+
127+
return udf(_callout, result_schema)(body_col)
128+
129+
def _callout_with_path(
130+
body_str: Optional[str], path_str: Optional[str]
131+
) -> Dict[str, Any]:
132+
row_request = (
133+
request.model_copy(
134+
update={"url": join_callout_path(request.url, path_str)}
135+
)
136+
if path_str
137+
else request
138+
)
139+
return _invoke_callout_as_struct(
140+
self._named_credential, row_request, body_str
141+
)
142+
143+
return udf(_callout_with_path, result_schema)(body_col, path)
144+
145+
146+
def join_callout_path(base_url: str, path: Optional[str]) -> str:
147+
"""Append a per-row ``path`` to the template ``base_url``.
148+
149+
A path not starting with ``/`` or ``?`` is joined with ``/`` so the row can
150+
never extend the Named Credential name (``callout:NC`` + ``2/x`` must not
151+
become ``callout:NC2/x``). A null or empty path returns ``base_url``.
152+
"""
153+
if not path:
154+
return base_url
155+
if path.startswith(("/", "?")):
156+
return base_url + path
157+
return f"{base_url}/{path}"
122158

123159

124160
def _invoke_callout_as_struct(

‎tests/spark/test_column_hints_deferred.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,8 @@ def _run(script: str) -> None:
3838
[sys.executable, "-c", textwrap.dedent(script)],
3939
capture_output=True,
4040
text=True,
41-
timeout=60, check=False,
41+
timeout=60,
42+
check=False,
4243
)
4344
if result.returncode != 0:
4445
pytest.fail(

‎tests/test_client.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -733,6 +733,27 @@ def test_delegates_to_spark_named_credential(self, mock_build, reset_client):
733733
assert result is sentinel_col
734734
mock_nc.request_col.assert_called_once_with(request, body=body_col)
735735

736+
@patch("datacustomcode.client._build_spark_named_credential")
737+
def test_forwards_per_row_path_column(self, mock_build, reset_client):
738+
mock_nc = MagicMock()
739+
sentinel_col = MagicMock(name="col")
740+
mock_nc.request_col.return_value = sentinel_col
741+
mock_build.return_value = mock_nc
742+
743+
reader = MagicMock(spec=BaseDataCloudReader)
744+
writer = MagicMock(spec=BaseDataCloudWriter)
745+
Client(reader=reader, writer=writer)
746+
747+
request = MagicMock(name="request")
748+
body_col = MagicMock(name="body_col")
749+
path_col = MagicMock(name="path_col")
750+
result = named_credential_request_col(request, body_col, path=path_col)
751+
752+
assert result is sentinel_col
753+
mock_nc.request_col.assert_called_once_with(
754+
request, body=body_col, path=path_col
755+
)
756+
736757

737758
class TestClientNamedCredentialRequest:
738759

‎tests/test_named_credential.py‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -403,6 +403,138 @@ def test_defaults_body_to_typed_null_column(self, mock_lit, mock_udf):
403403
# With no body column, a typed null string column is applied instead.
404404
sentinel_udf.assert_called_once_with(null_col)
405405

406+
@patch("pyspark.sql.functions.udf")
407+
@patch("pyspark.sql.functions.lit")
408+
def test_per_row_path_is_appended_to_request_url(self, mock_lit, mock_udf):
409+
from datacustomcode.named_credential.spark_default import (
410+
DefaultSparkNamedCredential,
411+
)
412+
413+
sentinel_udf = MagicMock(name="udf")
414+
mock_udf.return_value = sentinel_udf
415+
416+
underlying = MagicMock()
417+
underlying.request.return_value = HTTPResponse(
418+
status_code=200, headers={}, body="{}"
419+
)
420+
spark_nc = DefaultSparkNamedCredential(named_credential=underlying)
421+
422+
request = (
423+
HTTPRequestBuilder()
424+
.set_url("callout:gemini")
425+
.set_method("POST")
426+
.set_headers({"Accept": "application/json"})
427+
.build()
428+
)
429+
body_col = MagicMock(name="body_col")
430+
path_col = MagicMock(name="path_col")
431+
spark_nc.request_col(request, body_col, path=path_col)
432+
433+
# The UDF is applied over both the body and the per-row path columns.
434+
sentinel_udf.assert_called_once_with(body_col, path_col)
435+
436+
udf_fn = mock_udf.call_args.args[0]
437+
out = udf_fn('{"q": 1}', "/gemini-2.5-pro")
438+
439+
assert out["status"] == "SUCCESS"
440+
sent_request, sent_body = underlying.request.call_args.args
441+
# The row's path is joined onto the template url; method/headers kept.
442+
assert sent_request.url == "callout:gemini/gemini-2.5-pro"
443+
assert sent_request.method == "POST"
444+
assert sent_request.headers == {"Accept": "application/json"}
445+
assert sent_body == '{"q": 1}'
446+
# The caller's template request is left untouched.
447+
assert request.url == "callout:gemini"
448+
449+
@pytest.mark.parametrize(
450+
"base, path, expected",
451+
[
452+
("callout:gemini", "/v1/models", "callout:gemini/v1/models"),
453+
("callout:gemini", "?key=a", "callout:gemini?key=a"),
454+
(
455+
"callout:Geo/geocode",
456+
"?address=1+Main",
457+
"callout:Geo/geocode?address=1+Main",
458+
),
459+
("callout:Geo/v1", "/places/42", "callout:Geo/v1/places/42"),
460+
# Without a leading separator a "/" is inserted, so the row can never
461+
# extend the Named Credential name (callout:gemini2/...).
462+
("callout:gemini", "2/v1", "callout:gemini/2/v1"),
463+
("callout:gemini", "@evil.example.com", "callout:gemini/@evil.example.com"),
464+
],
465+
)
466+
@patch("pyspark.sql.functions.udf")
467+
@patch("pyspark.sql.functions.lit")
468+
def test_path_join_keeps_named_credential_fixed(
469+
self, mock_lit, mock_udf, base, path, expected
470+
):
471+
from datacustomcode.named_credential.spark_default import (
472+
DefaultSparkNamedCredential,
473+
)
474+
475+
mock_udf.return_value = MagicMock(name="udf")
476+
underlying = MagicMock()
477+
underlying.request.return_value = HTTPResponse(
478+
status_code=200, headers={}, body=""
479+
)
480+
spark_nc = DefaultSparkNamedCredential(named_credential=underlying)
481+
482+
request = HTTPRequestBuilder().set_url(base).build()
483+
spark_nc.request_col(request, path=MagicMock(name="path_col"))
484+
485+
udf_fn = mock_udf.call_args.args[0]
486+
udf_fn(None, path)
487+
488+
assert underlying.request.call_args.args[0].url == expected
489+
490+
@pytest.mark.parametrize("path", [None, ""])
491+
@patch("pyspark.sql.functions.udf")
492+
@patch("pyspark.sql.functions.lit")
493+
def test_null_or_empty_row_path_uses_request_url(self, mock_lit, mock_udf, path):
494+
from datacustomcode.named_credential.spark_default import (
495+
DefaultSparkNamedCredential,
496+
)
497+
498+
mock_udf.return_value = MagicMock(name="udf")
499+
500+
underlying = MagicMock()
501+
underlying.request.return_value = HTTPResponse(
502+
status_code=200, headers={}, body=""
503+
)
504+
spark_nc = DefaultSparkNamedCredential(named_credential=underlying)
505+
506+
request = HTTPRequestBuilder().set_url("callout:NC/default").build()
507+
spark_nc.request_col(request, path=MagicMock(name="path_col"))
508+
509+
udf_fn = mock_udf.call_args.args[0]
510+
udf_fn('{"a": 1}', path)
511+
512+
sent_request, sent_body = underlying.request.call_args.args
513+
assert sent_request is request
514+
assert sent_body == '{"a": 1}'
515+
516+
@patch("pyspark.sql.functions.udf")
517+
@patch("pyspark.sql.functions.lit")
518+
def test_path_without_body_applies_typed_null_body(self, mock_lit, mock_udf):
519+
from datacustomcode.named_credential.spark_default import (
520+
DefaultSparkNamedCredential,
521+
)
522+
523+
null_col = MagicMock(name="null_col")
524+
lit_none = MagicMock(name="lit_none")
525+
lit_none.cast.return_value = null_col
526+
mock_lit.return_value = lit_none
527+
528+
sentinel_udf = MagicMock(name="udf")
529+
mock_udf.return_value = sentinel_udf
530+
531+
spark_nc = DefaultSparkNamedCredential(named_credential=MagicMock())
532+
request = HTTPRequestBuilder().set_url("callout:NC/status").build()
533+
path_col = MagicMock(name="path_col")
534+
spark_nc.request_col(request, path=path_col)
535+
536+
sentinel_udf.assert_called_once_with(null_col, path_col)
537+
406538

407539
class TestInvokeCalloutAsStruct:
408540
"""The callout-to-struct shaping shared by every row."""

0 commit comments

Comments
 (0)