diff --git a/tableauserverclient/server/endpoint/datasources_endpoint.py b/tableauserverclient/server/endpoint/datasources_endpoint.py index f5212485a..9d1e190b1 100644 --- a/tableauserverclient/server/endpoint/datasources_endpoint.py +++ b/tableauserverclient/server/endpoint/datasources_endpoint.py @@ -497,7 +497,22 @@ def create_extract(self, datasource_item: DatasourceItem, encrypt: bool = False) Returns ------- JobItem - The job item. + The job item. The id it carries cannot be round-tripped through + `server.jobs.get_by_id(job.id)` -- that call raises + `ServerResponseError` with error code `400031` (see Notes). + `server.jobs.wait_for_job(job)` handles this transparently. + + Notes + ----- + Unlike `datasources.refresh(...)`, the JobItem returned here is not + addressable by `server.jobs.get_by_id(job.id)`: that call raises + `ServerResponseError` with error code `400031` and the message + "There was a problem querying job '{id}'.". See + tableau/server-client-python#1093. `server.jobs.wait_for_job(job)` + works normally -- it detects that error and falls back to polling + the paginated `/jobs` listing (see its own docstring for detail). + Callers who need a single JobItem outside the polling flow can + match on id themselves via `TSC.Pager(server.jobs)`. """ id_ = getattr(datasource_item, "id", datasource_item) url = f"{self.baseurl}/{id_}/createExtract?encrypt={encrypt}" diff --git a/tableauserverclient/server/endpoint/jobs_endpoint.py b/tableauserverclient/server/endpoint/jobs_endpoint.py index 655b7cc6f..406ac91c8 100644 --- a/tableauserverclient/server/endpoint/jobs_endpoint.py +++ b/tableauserverclient/server/endpoint/jobs_endpoint.py @@ -1,16 +1,32 @@ import logging +from typing import cast from typing_extensions import Self, overload from tableauserverclient.models import JobItem, BackgroundJobItem, PaginationItem from tableauserverclient.server.endpoint.endpoint import QuerysetEndpoint, api -from tableauserverclient.server.endpoint.exceptions import JobCancelledException, JobFailedException +from tableauserverclient.server.endpoint.exceptions import ( + JobCancelledException, + JobFailedException, + ServerResponseError, +) +from tableauserverclient.server.pager import Endpoint, Pager from tableauserverclient.server.query import QuerySet -from tableauserverclient.server.request_options import RequestOptionsBase +from tableauserverclient.server.request_options import RequestOptions, RequestOptionsBase from tableauserverclient.exponential_backoff import ExponentialBackoffTimer from tableauserverclient.helpers.logging import logger +# Server error code raised by GET /jobs/{id} for jobs whose ids are not +# addressable via the single-job endpoint (notably the JobItems returned by +# `create_extract`). Tracked at tableau/server-client-python#1093. +_UNQUERYABLE_JOB_ERROR_CODE = "400031" + +# Server max for the pageSize query parameter (Tableau raises 403014 above +# this). Used when the fallback poll walks /jobs looking for a specific id; +# a large page keeps sequential GETs on a busy site down to O(1) per poll. +_JOBS_LISTING_FALLBACK_PAGE_SIZE = 1000 + class Jobs(QuerysetEndpoint[BackgroundJobItem]): @property @@ -171,6 +187,26 @@ def wait_for_job(self, job_id: str | JobItem, *, timeout: float | None = None) - JobCancelledException If the job was cancelled. + + Notes + ----- + Some job ids returned elsewhere in the client are not addressable + via `GET /jobs/{id}` and produce a `ServerResponseError` with code + `400031`. When that happens, this method transparently falls back + to polling the paginated `/jobs` listing and matches on `id`. See + tableau/server-client-python#1093. + + On the fallback path the returned `JobItem` is translated from a + `BackgroundJobItem` (the shape the listing endpoint returns). + `BackgroundJobItem` doesn't carry every field a `JobItem` normally + has, so these come back as defaults: + * `progress` = `""` + * `notes` = `[]` + * `status_notes` = `[]` + * `mode`, `workbook_id`, `datasource_id`, `updated_at`, + `workbook_name`, `datasource_name`, `flow_run` = `None` + Check the values, not the type -- e.g. use `if not job.notes:`, not + `if job.notes is None:`. """ if isinstance(job_id, JobItem): job_id = job_id.id @@ -178,10 +214,10 @@ def wait_for_job(self, job_id: str | JobItem, *, timeout: float | None = None) - logger.debug(f"Waiting for job {job_id}") backoffTimer = ExponentialBackoffTimer(timeout=timeout) - job = self.get_by_id(job_id) + job = self._poll_job(job_id) while job.completed_at is None: backoffTimer.sleep() - job = self.get_by_id(job_id) + job = self._poll_job(job_id) logger.debug(f"\tJob {job_id} progress={job.progress}") logger.info(f"Job {job_id} Completed: Finish Code: {job.finish_code} - Notes:{job.notes}") @@ -195,6 +231,58 @@ def wait_for_job(self, job_id: str | JobItem, *, timeout: float | None = None) - else: raise AssertionError("Unexpected finish_code in job", job) + def _poll_job(self, job_id: str) -> JobItem: + try: + return self.get_by_id(job_id) + except ServerResponseError as err: + if err.code != _UNQUERYABLE_JOB_ERROR_CODE: + raise + logger.debug( + f"Job {job_id} not queryable via /jobs/{{id}} (400031); falling back to /jobs listing (see #1093)." + ) + + # `Jobs.get`'s overloads don't line up with the single-signature + # `Endpoint`/`CallableEndpoint` protocols, so a bare `Pager(self)` + # fails mypy even though it works at runtime. Cast at the call + # site: expresses intent locally, and if `Jobs.get`'s overloads + # ever change to match the protocol the cast surfaces the + # mismatch instead of silently hiding it. + options = RequestOptions(pagesize=_JOBS_LISTING_FALLBACK_PAGE_SIZE) + pager: Pager[BackgroundJobItem] = Pager(cast(Endpoint[BackgroundJobItem], self), request_opts=options) + for bg_job in pager: + if bg_job.id == job_id: + return self._background_to_job(bg_job) + raise + + @staticmethod + def _background_to_job(bg_job: BackgroundJobItem) -> JobItem: + status_map = { + BackgroundJobItem.Status.Success: JobItem.FinishCode.Success, + BackgroundJobItem.Status.Failed: JobItem.FinishCode.Failed, + BackgroundJobItem.Status.Cancelled: JobItem.FinishCode.Cancelled, + } + # Success, Failed, and Cancelled map to their FinishCode. Anything else -- + # Pending, InProgress, a None status, or a future addition -- keeps + # completed_at=None so wait_for_job's loop re-polls, and uses finish_code=-1 + # (never a real FinishCode) so a reader inspecting finish_code alone can't + # mistake an unfinished job for Success. + if bg_job.status in status_map: + finish_code = status_map[bg_job.status] + completed_at = bg_job.ended_at + else: + finish_code = -1 + completed_at = None + return JobItem( + id_=bg_job.id, + job_type=bg_job.type, + progress="", + created_at=bg_job.created_at, + started_at=bg_job.started_at, + completed_at=completed_at, + finish_code=finish_code, + notes=None, + ) + def filter(self, *invalid, page_size: int | None = None, **kwargs) -> QuerySet[BackgroundJobItem]: """ Queries the Tableau Server for items using the specified filters. Page diff --git a/tableauserverclient/server/endpoint/workbooks_endpoint.py b/tableauserverclient/server/endpoint/workbooks_endpoint.py index 8310d58b3..06fd1537c 100644 --- a/tableauserverclient/server/endpoint/workbooks_endpoint.py +++ b/tableauserverclient/server/endpoint/workbooks_endpoint.py @@ -187,7 +187,23 @@ def create_extract( Returns ------- JobItem - The job item for the extract creation. + The job item for the extract creation. The id it carries cannot + be round-tripped through `server.jobs.get_by_id(job.id)` -- that + call raises `ServerResponseError` with error code `400031` (see + Notes). `server.jobs.wait_for_job(job)` handles this + transparently. + + Notes + ----- + Unlike `workbooks.refresh(...)`, the JobItem returned here is not + addressable by `server.jobs.get_by_id(job.id)`: that call raises + `ServerResponseError` with error code `400031` and the message + "There was a problem querying job '{id}'.". See + tableau/server-client-python#1093. `server.jobs.wait_for_job(job)` + works normally -- it detects that error and falls back to polling + the paginated `/jobs` listing (see its own docstring for detail). + Callers who need a single JobItem outside the polling flow can + match on id themselves via `TSC.Pager(server.jobs)`. """ id_ = getattr(workbook_item, "id", workbook_item) url = f"{self.baseurl}/{id_}/createExtract?encrypt={encrypt}" @@ -222,6 +238,7 @@ def delete_extract(self, workbook_item: WorkbookItem, includeAll: bool = True, d Returns ------- JobItem + The job item. """ id_ = getattr(workbook_item, "id", workbook_item) url = f"{self.baseurl}/{id_}/deleteExtract" diff --git a/test/test_job.py b/test/test_job.py index 7bfdd1840..328eafa0d 100644 --- a/test/test_job.py +++ b/test/test_job.py @@ -6,7 +6,11 @@ import tableauserverclient as TSC from tableauserverclient.datetime_helpers import utc -from tableauserverclient.server.endpoint.exceptions import JobFailedException +from tableauserverclient.server.endpoint.exceptions import ( + JobCancelledException, + JobFailedException, + ServerResponseError, +) from ._utils import mocked_time TEST_ASSET_DIR = Path(__file__).parent / "assets" @@ -218,3 +222,249 @@ def test_background_job_str() -> None: assert not str(job).startswith("< str: + ended_attr = ' endedAt="2024-01-02T00:00:45Z"' if ended else "" + status_attr = f' status="{status}"' if status is not None else "" + return ( + '' + '' + f'' + "" + f'{extra_jobs}' + "" + "" + ) + + +def _listing_xml_page1_only( + filler_job_id: str, + page_size: int, + total_available: int, +) -> str: + # A page that does NOT contain the target job -- used to force Pager to + # request page 2. + return ( + '' + '' + f'' + "" + f'' + "" + "" + ) + + +def _empty_listing_xml() -> str: + return ( + '' + '' + '' + "" + "" + ) + + +def _error_xml(code: str, summary: str = "Bad Request", detail: str = "") -> str: + # `ServerResponseError.from_response` parses `t:error` using the tableau + # namespace, so a bare `` (as produced by + # `_utils.server_response_error_factory`) returns .code == "". Always + # include the xmlns declaration here so the parsed error carries the code. + return ( + '' + '' + f'' + f"{summary}" + f"{detail}" + "" + "" + ) + + +def _400031_error_xml() -> str: + return _error_xml( + "400031", + detail=f"There was a problem querying job '{UNQUERYABLE_JOB_ID}'.", + ) + + +def test_wait_for_job_no_listing_call_on_happy_path(server: TSC.Server) -> None: + response_xml = GET_BY_ID_XML.read_text() + job_id = "2eef4225-aa0c-41c4-8662-a76d89ed7336" + with mocked_time(), requests_mock.mock() as m: + m.get(f"{server.jobs.baseurl}/{job_id}", text=response_xml) + listing_mock = m.get(server.jobs.baseurl, text=_empty_listing_xml()) + + job = server.jobs.wait_for_job(job_id) + + assert job.id == job_id + assert listing_mock.call_count == 0 + + +def test_wait_for_job_fallback_success(server: TSC.Server) -> None: + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get( + server.jobs.baseurl, + text=_listing_xml(UNQUERYABLE_JOB_ID, status="Success", ended=True), + ) + + job = server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + + assert isinstance(job, TSC.JobItem) + assert job.id == UNQUERYABLE_JOB_ID + assert job.finish_code == TSC.JobItem.FinishCode.Success + assert job.completed_at == datetime(2024, 1, 2, 0, 0, 45, tzinfo=utc) + + +def test_wait_for_job_fallback_failed(server: TSC.Server) -> None: + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get( + server.jobs.baseurl, + text=_listing_xml(UNQUERYABLE_JOB_ID, status="Failed", ended=True), + ) + + with pytest.raises(JobFailedException): + server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + + +def test_wait_for_job_fallback_cancelled(server: TSC.Server) -> None: + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get( + server.jobs.baseurl, + text=_listing_xml(UNQUERYABLE_JOB_ID, status="Cancelled", ended=True), + ) + + with pytest.raises(JobCancelledException): + server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + + +def test_wait_for_job_fallback_not_in_listing(server: TSC.Server) -> None: + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get(server.jobs.baseurl, text=_empty_listing_xml()) + + with pytest.raises(ServerResponseError) as exc_info: + server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + assert exc_info.value.code == "400031" + + +def test_wait_for_job_non_400031_not_swallowed(server: TSC.Server) -> None: + other_error = _error_xml("400000", detail="Something else") + with mocked_time(), requests_mock.mock() as m: + m.get(f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", text=other_error, status_code=400) + listing_mock = m.get(server.jobs.baseurl, text=_empty_listing_xml()) + + with pytest.raises(ServerResponseError) as exc_info: + server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + assert exc_info.value.code == "400000" + assert listing_mock.call_count == 0 + + +def test_wait_for_job_fallback_inprogress_then_success(server: TSC.Server) -> None: + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get( + server.jobs.baseurl, + [ + {"text": _listing_xml(UNQUERYABLE_JOB_ID, status="InProgress", ended=False)}, + {"text": _listing_xml(UNQUERYABLE_JOB_ID, status="Success", ended=True)}, + ], + ) + + job = server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + + assert job.finish_code == TSC.JobItem.FinishCode.Success + assert job.completed_at == datetime(2024, 1, 2, 0, 0, 45, tzinfo=utc) + + +def test_wait_for_job_fallback_walks_multiple_pages(server: TSC.Server) -> None: + other_job_id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get( + server.jobs.baseurl, + [ + {"text": _listing_xml_page1_only(other_job_id, page_size=1, total_available=2)}, + { + "text": _listing_xml( + UNQUERYABLE_JOB_ID, + status="Success", + ended=True, + page_number=2, + page_size=1, + total_available=2, + ) + }, + ], + ) + + job = server.jobs.wait_for_job(UNQUERYABLE_JOB_ID) + + assert job.id == UNQUERYABLE_JOB_ID + assert job.finish_code == TSC.JobItem.FinishCode.Success + + +def test_wait_for_job_fallback_timeout(server: TSC.Server) -> None: + with mocked_time(), requests_mock.mock() as m: + m.get( + f"{server.jobs.baseurl}/{UNQUERYABLE_JOB_ID}", + text=_400031_error_xml(), + status_code=400, + ) + m.get( + server.jobs.baseurl, + text=_listing_xml(UNQUERYABLE_JOB_ID, status="InProgress", ended=False), + ) + + with pytest.raises(TimeoutError): + server.jobs.wait_for_job(UNQUERYABLE_JOB_ID, timeout=30)