mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
fix(python): bound Job and AsyncJob payload previews (#4939)
Printing a Job or a bound status method could dump hundreds of lines of serialized Function initialization. Bound nested payload previews in both sync and async handles while preserving cached identity, state, failure and result information. Full spec/result values remain available through their properties, with no additional backend calls. Adds regression coverage for large payloads, bound method display, unobserved handles and completed local jobs. The broader method/property API design remains separate. Formatting applied; no local builds or tests were run. CI will execute the regression coverage.
This commit is contained in:
@@ -4,7 +4,7 @@
|
||||
"""Handles to asynchronous registry jobs, SQL queries, and local operations."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import reprlib
|
||||
import time
|
||||
from datetime import timedelta
|
||||
from typing import Any, Callable, Generic, Optional, TypeVar, cast
|
||||
@@ -461,43 +461,49 @@ class AsyncJob(Generic[T]):
|
||||
|
||||
|
||||
_REPR_INDENT = " " * 4
|
||||
_REPR_MAX_FIELD_LENGTH = 240
|
||||
_PAYLOAD_REPR = reprlib.Repr()
|
||||
_PAYLOAD_REPR.maxlevel = 2
|
||||
_PAYLOAD_REPR.maxdict = 4
|
||||
_PAYLOAD_REPR.maxlist = 4
|
||||
_PAYLOAD_REPR.maxtuple = 4
|
||||
_PAYLOAD_REPR.maxstring = 120
|
||||
_PAYLOAD_REPR.maxother = 120
|
||||
|
||||
|
||||
def _repr_payload(value: Any) -> str:
|
||||
"""Render a job payload as indented JSON, aligned under its field."""
|
||||
try:
|
||||
rendered = json.dumps(value, indent=4)
|
||||
except TypeError:
|
||||
return repr(value)
|
||||
return rendered.replace("\n", "\n" + _REPR_INDENT)
|
||||
"""Bound nested payload previews without changing the stored values."""
|
||||
rendered = _PAYLOAD_REPR.repr(value).replace("\n", "\\n")
|
||||
if len(rendered) > _REPR_MAX_FIELD_LENGTH:
|
||||
return rendered[: _REPR_MAX_FIELD_LENGTH - 3] + "..."
|
||||
return rendered
|
||||
|
||||
|
||||
def _job_repr(kind: str, job: Any) -> str:
|
||||
"""Render every field the handle currently knows, omitting the rest.
|
||||
"""Show cached job identity and bounded previews without backend calls.
|
||||
|
||||
One field per line, with the JSON payloads indented, because a refresh
|
||||
job's spec and result are the point of printing it.
|
||||
The complete payloads remain available through ``spec`` and ``result``.
|
||||
"""
|
||||
state = job.state
|
||||
if state is None:
|
||||
# Nothing has been fetched yet, so there is nothing to lay out.
|
||||
known = f"id={job.id!r}, " if job.id is not None else ""
|
||||
known = f"id={_repr_payload(job.id)}, " if job.id is not None else ""
|
||||
return f"{kind}({known}not refreshed)"
|
||||
|
||||
fields = []
|
||||
if job.id is not None:
|
||||
fields.append(f"id={job.id!r}")
|
||||
fields.append(f"state={state!r}")
|
||||
fields.append(f"id={_repr_payload(job.id)}")
|
||||
fields.append(f"state={_repr_payload(state)}")
|
||||
for name in ("job_type", "creation_ms"):
|
||||
value = getattr(job, name)
|
||||
if value is not None:
|
||||
fields.append(f"{name}={value!r}")
|
||||
fields.append(f"{name}={_repr_payload(value)}")
|
||||
for name in ("spec", "result"):
|
||||
value = getattr(job, name)
|
||||
if value is not None:
|
||||
fields.append(f"{name}={_repr_payload(value)}")
|
||||
if job.failure is not None:
|
||||
fields.append(f"failure={job.failure!r}")
|
||||
fields.append(f"failure={_repr_payload(job.failure)}")
|
||||
body = "".join(f"\n{_REPR_INDENT}{field}," for field in fields)
|
||||
return f"{kind}({body}\n)"
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
import asyncio
|
||||
import time
|
||||
from datetime import timedelta
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
@@ -13,6 +14,50 @@ from lancedb.job import AsyncJob, Job, _progress_eta
|
||||
from lancedb.remote.errors import HttpError, RetryError
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sync", [False, True])
|
||||
def test_job_repr_bounds_large_payloads_without_fetching(sync):
|
||||
spec = {"initialization": list(range(100000)), "source": "x" * 100000}
|
||||
result = {"rows_computed": 7, "details": [{"data": b"x" * 10000}] * 1000}
|
||||
inner = SimpleNamespace(
|
||||
id="job-repr",
|
||||
_state="failed",
|
||||
job_type="function_refresh",
|
||||
_description=SimpleNamespace(
|
||||
creation_ms=2000,
|
||||
spec=spec,
|
||||
result=result,
|
||||
failure="worker failed " + "x" * 10000,
|
||||
),
|
||||
)
|
||||
job = AsyncJob(inner)
|
||||
if sync:
|
||||
job = Job(job)
|
||||
rendered = repr(job)
|
||||
assert len(rendered) < 1600
|
||||
assert len(rendered.splitlines()) <= 10
|
||||
assert "id='job-repr'" in rendered
|
||||
assert "state='failed'" in rendered
|
||||
assert "job_type='function_refresh'" in rendered
|
||||
assert "rows_computed" in rendered
|
||||
assert "worker failed" in rendered
|
||||
assert "..." in rendered
|
||||
assert len(repr(job.status)) < 1700
|
||||
assert job.spec is spec
|
||||
assert job.result is result
|
||||
assert len(job.spec["initialization"]) == 100000
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sync", [False, True])
|
||||
def test_job_repr_handles_unobserved_and_local_jobs(sync):
|
||||
unobserved = AsyncJob(SimpleNamespace(id="job-pending", _state=None))
|
||||
local = AsyncJob(None)
|
||||
if sync:
|
||||
unobserved, local = Job(unobserved), Job(local)
|
||||
assert "id='job-pending', not refreshed" in repr(unobserved)
|
||||
assert "state='finished'" in repr(local)
|
||||
assert "failure=" not in repr(local)
|
||||
|
||||
|
||||
@pytest.fixture(params=["async", "sync"])
|
||||
def wait_job(request):
|
||||
async def wait(inner, timeout=None, result_decoder=None):
|
||||
|
||||
@@ -5539,9 +5539,6 @@ def test_remote_job_handle_reports_its_own_detail():
|
||||
assert json.loads(job._spec_json) == {"column": "vec"}
|
||||
assert json.loads(job._result_json) == {"rows_assigned": 1000000}
|
||||
|
||||
# print() shows everything the handle knows and nothing it does not.
|
||||
# print() lays every known field out on its own line, with the JSON
|
||||
# payloads indented rather than crammed onto one line.
|
||||
assert repr(job) == "\n".join(
|
||||
[
|
||||
"Job(",
|
||||
@@ -5549,12 +5546,8 @@ def test_remote_job_handle_reports_its_own_detail():
|
||||
" state='finished',",
|
||||
" job_type='refresh_column',",
|
||||
" creation_ms=2000,",
|
||||
" spec={",
|
||||
' "column": "vec"',
|
||||
" },",
|
||||
" result={",
|
||||
' "rows_assigned": 1000000',
|
||||
" },",
|
||||
" spec={'column': 'vec'},",
|
||||
" result={'rows_assigned': 1000000},",
|
||||
")",
|
||||
]
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user