diff --git a/python/python/lancedb/job.py b/python/python/lancedb/job.py index 92b545ce7..86a1295cc 100644 --- a/python/python/lancedb/job.py +++ b/python/python/lancedb/job.py @@ -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)" diff --git a/python/python/tests/test_job.py b/python/python/tests/test_job.py index c3189cbad..07b8a8c5e 100644 --- a/python/python/tests/test_job.py +++ b/python/python/tests/test_job.py @@ -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): diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 671d0ff23..789df1726 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -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},", ")", ] )