# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.http_response import AsyncHttpResponse, HttpResponse from ..core.jsonable_encoder import jsonable_encoder from ..core.pydantic_utilities import parse_obj_as from ..core.request_options import RequestOptions from ..core.serialization import convert_and_respect_annotation_metadata from ..errors.bad_request_error import BadRequestError from ..errors.forbidden_error import ForbiddenError from ..errors.internal_server_error import InternalServerError from ..errors.service_unavailable_error import ServiceUnavailableError from ..errors.too_many_requests_error import TooManyRequestsError from ..requests.doc_digitization_job_parameters import DocDigitizationJobParametersParams from ..requests.doc_digitization_webhook_callback import DocDigitizationWebhookCallbackParams from ..types.doc_digitization_create_job_response import DocDigitizationCreateJobResponse from ..types.doc_digitization_download_files_response import DocDigitizationDownloadFilesResponse from ..types.doc_digitization_job_status_response import DocDigitizationJobStatusResponse from ..types.doc_digitization_upload_files_response import DocDigitizationUploadFilesResponse # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawDocumentIntelligenceClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def initialise( self, *, job_parameters: typing.Optional[DocDigitizationJobParametersParams] = OMIT, callback: typing.Optional[DocDigitizationWebhookCallbackParams] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[DocDigitizationCreateJobResponse]: """ Creates a new Document Intelligence job. **Supported Languages (BCP-47 format):** - `hi-IN`: Hindi (default) - `en-IN`: English - `bn-IN`: Bengali - `gu-IN`: Gujarati - `kn-IN`: Kannada - `ml-IN`: Malayalam - `mr-IN`: Marathi - `or-IN`: Odia - `pa-IN`: Punjabi - `ta-IN`: Tamil - `te-IN`: Telugu - `ur-IN`: Urdu - `as-IN`: Assamese - `bodo-IN`: Bodo - `doi-IN`: Dogri - `ks-IN`: Kashmiri - `kok-IN`: Konkani - `mai-IN`: Maithili - `mni-IN`: Manipuri - `ne-IN`: Nepali - `sa-IN`: Sanskrit - `sat-IN`: Santali - `sd-IN`: Sindhi **Output Formats (delivered as ZIP file):** - `html`: Structured HTML files with layout preservation - `md`: Markdown files (default) - `json`: Structured JSON files for programmatic processing Parameters ---------- job_parameters : typing.Optional[DocDigitizationJobParametersParams] Configuration parameters for the Document Intelligence job including language and output format. Defaults to Hindi (hi-IN) and Markdown output if omitted. callback : typing.Optional[DocDigitizationWebhookCallbackParams] Optional webhook for completion notification request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DocDigitizationCreateJobResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( "doc-digitization/job/v1", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "job_parameters": convert_and_respect_annotation_metadata( object_=job_parameters, annotation=DocDigitizationJobParametersParams, direction="write" ), "callback": convert_and_respect_annotation_metadata( object_=callback, annotation=DocDigitizationWebhookCallbackParams, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationCreateJobResponse, parse_obj_as( type_=DocDigitizationCreateJobResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_upload_links( self, *, job_id: str, files: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DocDigitizationUploadFilesResponse]: """ Returns presigned URLs for uploading input files. **File Constraints:** - Exactly one file required (PDF or ZIP) - PDF files: `.pdf` extension - ZIP files: `.zip` extension Parameters ---------- job_id : str Job identifier returned from Create Job files : typing.Sequence[str] List of filenames to upload (exactly 1 file: PDF or ZIP) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DocDigitizationUploadFilesResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( "doc-digitization/job/v1/upload-files", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "job_id": job_id, "files": files, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationUploadFilesResponse, parse_obj_as( type_=DocDigitizationUploadFilesResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def start( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DocDigitizationJobStatusResponse]: """ Validates the uploaded file and starts processing. **Validation Checks:** - File must be uploaded before starting - File size must not exceed 200 MB - PDF must be parseable by the PDF parser - ZIP must contain only JPEG/PNG images - ZIP must be flat (no nested folders beyond one level) - ZIP must contain at least one valid image - Page/image count must not exceed 500 - User must have sufficient credits **Processing:** Job runs asynchronously. Poll the status endpoint or use webhook callback for completion notification. Parameters ---------- job_id : str The unique identifier of the job request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DocDigitizationJobStatusResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"doc-digitization/job/v1/{jsonable_encoder(job_id)}/start", base_url=self._client_wrapper.get_environment().base, method="POST", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationJobStatusResponse, parse_obj_as( type_=DocDigitizationJobStatusResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_status( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DocDigitizationJobStatusResponse]: """ Returns the current status of a job with page-level metrics. **Job States:** - `Accepted`: Job created, awaiting file upload - `Pending`: File uploaded, waiting to start - `Running`: Processing in progress - `Completed`: All pages processed successfully - `PartiallyCompleted`: Some pages succeeded, some failed - `Failed`: All pages failed or job-level error **Page Metrics:** Response includes detailed progress: total pages, pages processed, succeeded, failed, and per-page errors. Parameters ---------- job_id : str The unique identifier of the job request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DocDigitizationJobStatusResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"doc-digitization/job/v1/{jsonable_encoder(job_id)}/status", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationJobStatusResponse, parse_obj_as( type_=DocDigitizationJobStatusResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def get_download_links( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DocDigitizationDownloadFilesResponse]: """ Returns presigned URLs for downloading output files. **Prerequisites:** - Job must be in `Completed` or `PartiallyCompleted` state - Failed jobs have no output available Parameters ---------- job_id : str The unique identifier of the job request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DocDigitizationDownloadFilesResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"doc-digitization/job/v1/{jsonable_encoder(job_id)}/download-files", base_url=self._client_wrapper.get_environment().base, method="POST", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationDownloadFilesResponse, parse_obj_as( type_=DocDigitizationDownloadFilesResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawDocumentIntelligenceClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def initialise( self, *, job_parameters: typing.Optional[DocDigitizationJobParametersParams] = OMIT, callback: typing.Optional[DocDigitizationWebhookCallbackParams] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[DocDigitizationCreateJobResponse]: """ Creates a new Document Intelligence job. **Supported Languages (BCP-47 format):** - `hi-IN`: Hindi (default) - `en-IN`: English - `bn-IN`: Bengali - `gu-IN`: Gujarati - `kn-IN`: Kannada - `ml-IN`: Malayalam - `mr-IN`: Marathi - `or-IN`: Odia - `pa-IN`: Punjabi - `ta-IN`: Tamil - `te-IN`: Telugu - `ur-IN`: Urdu - `as-IN`: Assamese - `bodo-IN`: Bodo - `doi-IN`: Dogri - `ks-IN`: Kashmiri - `kok-IN`: Konkani - `mai-IN`: Maithili - `mni-IN`: Manipuri - `ne-IN`: Nepali - `sa-IN`: Sanskrit - `sat-IN`: Santali - `sd-IN`: Sindhi **Output Formats (delivered as ZIP file):** - `html`: Structured HTML files with layout preservation - `md`: Markdown files (default) - `json`: Structured JSON files for programmatic processing Parameters ---------- job_parameters : typing.Optional[DocDigitizationJobParametersParams] Configuration parameters for the Document Intelligence job including language and output format. Defaults to Hindi (hi-IN) and Markdown output if omitted. callback : typing.Optional[DocDigitizationWebhookCallbackParams] Optional webhook for completion notification request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DocDigitizationCreateJobResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "doc-digitization/job/v1", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "job_parameters": convert_and_respect_annotation_metadata( object_=job_parameters, annotation=DocDigitizationJobParametersParams, direction="write" ), "callback": convert_and_respect_annotation_metadata( object_=callback, annotation=DocDigitizationWebhookCallbackParams, direction="write" ), }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationCreateJobResponse, parse_obj_as( type_=DocDigitizationCreateJobResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_upload_links( self, *, job_id: str, files: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DocDigitizationUploadFilesResponse]: """ Returns presigned URLs for uploading input files. **File Constraints:** - Exactly one file required (PDF or ZIP) - PDF files: `.pdf` extension - ZIP files: `.zip` extension Parameters ---------- job_id : str Job identifier returned from Create Job files : typing.Sequence[str] List of filenames to upload (exactly 1 file: PDF or ZIP) request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DocDigitizationUploadFilesResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( "doc-digitization/job/v1/upload-files", base_url=self._client_wrapper.get_environment().base, method="POST", json={ "job_id": job_id, "files": files, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationUploadFilesResponse, parse_obj_as( type_=DocDigitizationUploadFilesResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def start( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DocDigitizationJobStatusResponse]: """ Validates the uploaded file and starts processing. **Validation Checks:** - File must be uploaded before starting - File size must not exceed 200 MB - PDF must be parseable by the PDF parser - ZIP must contain only JPEG/PNG images - ZIP must be flat (no nested folders beyond one level) - ZIP must contain at least one valid image - Page/image count must not exceed 500 - User must have sufficient credits **Processing:** Job runs asynchronously. Poll the status endpoint or use webhook callback for completion notification. Parameters ---------- job_id : str The unique identifier of the job request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DocDigitizationJobStatusResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"doc-digitization/job/v1/{jsonable_encoder(job_id)}/start", base_url=self._client_wrapper.get_environment().base, method="POST", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationJobStatusResponse, parse_obj_as( type_=DocDigitizationJobStatusResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_status( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DocDigitizationJobStatusResponse]: """ Returns the current status of a job with page-level metrics. **Job States:** - `Accepted`: Job created, awaiting file upload - `Pending`: File uploaded, waiting to start - `Running`: Processing in progress - `Completed`: All pages processed successfully - `PartiallyCompleted`: Some pages succeeded, some failed - `Failed`: All pages failed or job-level error **Page Metrics:** Response includes detailed progress: total pages, pages processed, succeeded, failed, and per-page errors. Parameters ---------- job_id : str The unique identifier of the job request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DocDigitizationJobStatusResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"doc-digitization/job/v1/{jsonable_encoder(job_id)}/status", base_url=self._client_wrapper.get_environment().base, method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationJobStatusResponse, parse_obj_as( type_=DocDigitizationJobStatusResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def get_download_links( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DocDigitizationDownloadFilesResponse]: """ Returns presigned URLs for downloading output files. **Prerequisites:** - Job must be in `Completed` or `PartiallyCompleted` state - Failed jobs have no output available Parameters ---------- job_id : str The unique identifier of the job request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DocDigitizationDownloadFilesResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"doc-digitization/job/v1/{jsonable_encoder(job_id)}/download-files", base_url=self._client_wrapper.get_environment().base, method="POST", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DocDigitizationDownloadFilesResponse, parse_obj_as( type_=DocDigitizationDownloadFilesResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise BadRequestError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise ForbiddenError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 429: raise TooManyRequestsError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 500: raise InternalServerError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 503: raise ServiceUnavailableError( headers=dict(_response.headers), body=typing.cast( typing.Optional[typing.Any], parse_obj_as( type_=typing.Optional[typing.Any], # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)