# This file was auto-generated by Fern from our API Definition. import os import time import typing import httpx from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions 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, ) from .raw_client import ( AsyncRawDocumentIntelligenceClient, RawDocumentIntelligenceClient, ) # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class DocumentIntelligenceJob: """ A convenience wrapper for managing document intelligence jobs. This class provides high-level methods for the complete document processing workflow: create job → upload file → start → wait → download output. """ def __init__( self, *, client: "DocumentIntelligenceClient", job_id: str, language: typing.Optional[str] = None, output_format: typing.Optional[str] = None, ): self._client = client self._job_id = job_id self._language = language self._output_format = output_format self._status: typing.Optional[DocDigitizationJobStatusResponse] = None @property def job_id(self) -> str: """The unique identifier for this job.""" return self._job_id @property def language(self) -> typing.Optional[str]: """The language configured for this job.""" return self._language @property def output_format(self) -> typing.Optional[str]: """The output format configured for this job.""" return self._output_format def upload_file(self, file_path: str) -> None: """ Upload a file for processing. Parameters ---------- file_path : str Path to the file to upload (PDF, PNG, JPG, or ZIP) """ filename = os.path.basename(file_path) # Get presigned upload URL upload_response = self._client.get_upload_links( job_id=self._job_id, files=[filename] ) if not upload_response.upload_urls: raise ValueError("No upload URL returned") # Get the upload URL for the filename file_details = upload_response.upload_urls.get(filename) if not file_details: raise ValueError(f"No upload URL for file: {filename}") upload_url = file_details.file_url # Determine content type ext = os.path.splitext(filename)[1].lower() content_types = { ".pdf": "application/pdf", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".zip": "application/zip", } content_type = content_types.get(ext, "application/octet-stream") # Upload file with open(file_path, "rb") as f: file_data = f.read() response = httpx.put( upload_url, content=file_data, headers={ "Content-Type": content_type, "x-ms-blob-type": "BlockBlob", # Required for Azure Blob Storage }, timeout=300.0, ) response.raise_for_status() def start(self) -> DocDigitizationJobStatusResponse: """ Start processing the uploaded file. Returns ------- DocDigitizationJobStatusResponse The job status after starting """ self._status = self._client.start(self._job_id) return self._status def get_status(self) -> DocDigitizationJobStatusResponse: """ Get the current status of the job. Returns ------- DocDigitizationJobStatusResponse The current job status """ self._status = self._client.get_status(self._job_id) return self._status def wait_until_complete( self, poll_interval: float = 2.0, timeout: typing.Optional[float] = None, ) -> DocDigitizationJobStatusResponse: """ Poll the job status until it completes or fails. Parameters ---------- poll_interval : float Seconds between status checks (default: 2.0) timeout : float, optional Maximum seconds to wait (default: None = wait forever) Returns ------- DocDigitizationJobStatusResponse The final job status Raises ------ TimeoutError If timeout is reached before job completes """ start_time = time.time() terminal_states = {"Completed", "PartiallyCompleted", "Failed"} while True: status = self.get_status() if status.job_state in terminal_states: return status if timeout is not None and (time.time() - start_time) >= timeout: raise TimeoutError( f"Job {self._job_id} did not complete within {timeout} seconds" ) time.sleep(poll_interval) def get_page_metrics(self) -> typing.Optional[typing.Dict[str, typing.Any]]: """ Get page-level metrics from the last status check. Returns ------- dict or None Dictionary with total_pages, pages_processed, pages_succeeded, pages_failed """ if self._status is None: self.get_status() if ( self._status and self._status.job_details and len(self._status.job_details) > 0 ): detail = self._status.job_details[0] return { "total_pages": detail.total_pages, "pages_processed": detail.pages_processed, "pages_succeeded": detail.pages_succeeded, "pages_failed": detail.pages_failed, } return None def download_output(self, output_path: str) -> str: """ Download the processed output to a file. Parameters ---------- output_path : str Path where the output file will be saved Returns ------- str The path to the downloaded file """ download_response = self._client.get_download_links(self._job_id) if not download_response.download_urls: raise ValueError("No download URL available") # Get the first available download URL first_filename = next(iter(download_response.download_urls.keys())) file_details = download_response.download_urls[first_filename] download_url = file_details.file_url # Download file response = httpx.get(download_url, timeout=300.0) response.raise_for_status() # Ensure output directory exists output_dir = os.path.dirname(output_path) if output_dir: os.makedirs(output_dir, exist_ok=True) with open(output_path, "wb") as f: f.write(response.content) return output_path class DocumentIntelligenceClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawDocumentIntelligenceClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawDocumentIntelligenceClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawDocumentIntelligenceClient """ return self._raw_client def create_job( self, *, language: str = "hi-IN", output_format: str = "html", callback_url: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> DocumentIntelligenceJob: """ Create a new document intelligence job with convenience methods. This is a high-level method that returns a DocumentIntelligenceJob object with methods for uploading, starting, waiting, and downloading. Parameters ---------- language : str Language code in BCP-47 format (default: "hi-IN") Supported: hi-IN, en-IN, bn-IN, gu-IN, kn-IN, ml-IN, mr-IN, or-IN, pa-IN, ta-IN, te-IN, ur-IN, as-IN, bodo-IN, doi-IN, ks-IN, kok-IN, mai-IN, mni-IN, ne-IN, sa-IN, sat-IN, sd-IN output_format : str Output format: "html" or "md" (default: "html") callback_url : str, optional Webhook URL for completion notification request_options : RequestOptions, optional Request-specific configuration Returns ------- DocumentIntelligenceJob A job object with convenience methods for the workflow Examples -------- from sarvamai import SarvamAI client = SarvamAI(api_subscription_key="YOUR_API_KEY") # Create job job = client.document_intelligence.create_job( language="hi-IN", output_format="html" ) # Upload, start, wait, download job.upload_file("document.pdf") job.start() job.wait_until_complete() job.download_output("./output.html") """ # Build job parameters job_params: DocDigitizationJobParametersParams = { "language": language, "output_format": output_format, } # Build callback if provided callback: typing.Optional[DocDigitizationWebhookCallbackParams] = None if callback_url is not None: callback = {"url": callback_url} # Create the job via the API response = self.initialise( job_parameters=job_params, callback=callback, request_options=request_options, ) # Return a job object with convenience methods return DocumentIntelligenceJob( client=self, job_id=response.job_id, language=language, output_format=output_format, ) def initialise( self, *, job_parameters: typing.Optional[DocDigitizationJobParametersParams] = OMIT, callback: typing.Optional[DocDigitizationWebhookCallbackParams] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> 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:** - `html`: Structured HTML with layout preservation (default) - `md`: Markdown format **Prompt Types:** Customize how specific content types are processed: - `default_ocr`: Standard text extraction (default for all text blocks) - `table_to_html`: Convert tables to HTML format - `table_to_markdown`: Convert tables to Markdown format - `chart_to_markdown`: Extract chart data as Markdown table - `chart_to_json`: Extract chart data as JSON - `describe_image`: Generate image caption - `caption_en`: Same as describe_image (English) - `caption_in`: Caption in document language **Webhook Callback:** Optionally provide a callback URL to receive notification when processing completes. Parameters ---------- job_parameters : typing.Optional[DocDigitizationJobParametersParams] Job configuration parameters. Omit the request body to use defaults. callback : typing.Optional[DocDigitizationWebhookCallbackParams] Optional webhook for completion notification request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DocDigitizationCreateJobResponse Successful Response Examples -------- from sarvamai import SarvamAI client = SarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) client.document_intelligence.initialise() """ _response = self._raw_client.initialise( job_parameters=job_parameters, callback=callback, request_options=request_options, ) return _response.data def get_upload_links( self, *, job_id: str, files: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None, ) -> 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 ------- DocDigitizationUploadFilesResponse Successful Response Examples -------- from sarvamai import SarvamAI client = SarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) client.document_intelligence.get_upload_links( job_id="job_id", files=["files"], ) """ _response = self._raw_client.get_upload_links( job_id=job_id, files=files, request_options=request_options ) return _response.data def start( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> 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 ------- DocDigitizationJobStatusResponse Successful Response Examples -------- from sarvamai import SarvamAI client = SarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) client.document_intelligence.start( job_id="job_id", ) """ _response = self._raw_client.start(job_id, request_options=request_options) return _response.data def get_status( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> 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 ------- DocDigitizationJobStatusResponse Successful Response Examples -------- from sarvamai import SarvamAI client = SarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) client.document_intelligence.get_status( job_id="job_id", ) """ _response = self._raw_client.get_status(job_id, request_options=request_options) return _response.data def get_download_links( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> 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 ------- DocDigitizationDownloadFilesResponse Successful Response Examples -------- from sarvamai import SarvamAI client = SarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) client.document_intelligence.get_download_links( job_id="job_id", ) """ _response = self._raw_client.get_download_links( job_id, request_options=request_options ) return _response.data class AsyncDocumentIntelligenceJob: """ An async convenience wrapper for managing document intelligence jobs. This class provides high-level async methods for the complete document processing workflow: create job → upload file → start → wait → download output. """ def __init__( self, *, client: "AsyncDocumentIntelligenceClient", job_id: str, language: typing.Optional[str] = None, output_format: typing.Optional[str] = None, ): self._client = client self._job_id = job_id self._language = language self._output_format = output_format self._status: typing.Optional[DocDigitizationJobStatusResponse] = None @property def job_id(self) -> str: """The unique identifier for this job.""" return self._job_id @property def language(self) -> typing.Optional[str]: """The language configured for this job.""" return self._language @property def output_format(self) -> typing.Optional[str]: """The output format configured for this job.""" return self._output_format async def upload_file(self, file_path: str) -> None: """ Upload a file for processing. Parameters ---------- file_path : str Path to the file to upload (PDF, PNG, JPG, or ZIP) """ filename = os.path.basename(file_path) # Get presigned upload URL upload_response = await self._client.get_upload_links( job_id=self._job_id, files=[filename] ) if not upload_response.upload_urls: raise ValueError("No upload URL returned") # Get the upload URL for the filename file_details = upload_response.upload_urls.get(filename) if not file_details: raise ValueError(f"No upload URL for file: {filename}") upload_url = file_details.file_url # Determine content type ext = os.path.splitext(filename)[1].lower() content_types = { ".pdf": "application/pdf", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".zip": "application/zip", } content_type = content_types.get(ext, "application/octet-stream") # Upload file (sync read is fine before async HTTP call) with open(file_path, "rb") as f: file_data = f.read() async with httpx.AsyncClient() as http_client: response = await http_client.put( upload_url, content=file_data, headers={ "Content-Type": content_type, "x-ms-blob-type": "BlockBlob", # Required for Azure Blob Storage }, timeout=300.0, ) response.raise_for_status() async def start(self) -> DocDigitizationJobStatusResponse: """ Start processing the uploaded file. Returns ------- DocDigitizationJobStatusResponse The job status after starting """ self._status = await self._client.start(self._job_id) return self._status async def get_status(self) -> DocDigitizationJobStatusResponse: """ Get the current status of the job. Returns ------- DocDigitizationJobStatusResponse The current job status """ self._status = await self._client.get_status(self._job_id) return self._status async def wait_until_complete( self, poll_interval: float = 2.0, timeout: typing.Optional[float] = None, ) -> DocDigitizationJobStatusResponse: """ Poll the job status until it completes or fails. Parameters ---------- poll_interval : float Seconds between status checks (default: 2.0) timeout : float, optional Maximum seconds to wait (default: None = wait forever) Returns ------- DocDigitizationJobStatusResponse The final job status Raises ------ TimeoutError If timeout is reached before job completes """ import asyncio start_time = time.time() terminal_states = {"Completed", "PartiallyCompleted", "Failed"} while True: status = await self.get_status() if status.job_state in terminal_states: return status if timeout is not None and (time.time() - start_time) >= timeout: raise TimeoutError( f"Job {self._job_id} did not complete within {timeout} seconds" ) await asyncio.sleep(poll_interval) def get_page_metrics(self) -> typing.Optional[typing.Dict[str, typing.Any]]: """ Get page-level metrics from the last status check. Returns ------- dict or None Dictionary with total_pages, pages_processed, pages_succeeded, pages_failed """ if ( self._status and self._status.job_details and len(self._status.job_details) > 0 ): detail = self._status.job_details[0] return { "total_pages": detail.total_pages, "pages_processed": detail.pages_processed, "pages_succeeded": detail.pages_succeeded, "pages_failed": detail.pages_failed, } return None async def download_output(self, output_path: str) -> str: """ Download the processed output to a file. Parameters ---------- output_path : str Path where the output file will be saved Returns ------- str The path to the downloaded file """ download_response = await self._client.get_download_links(self._job_id) if not download_response.download_urls: raise ValueError("No download URL available") # Get the first available download URL first_filename = next(iter(download_response.download_urls.keys())) file_details = download_response.download_urls[first_filename] download_url = file_details.file_url # Download file async with httpx.AsyncClient() as http_client: response = await http_client.get(download_url, timeout=300.0) response.raise_for_status() # Ensure output directory exists output_dir = os.path.dirname(output_path) if output_dir: os.makedirs(output_dir, exist_ok=True) # Sync write is fine after async HTTP call with open(output_path, "wb") as f: f.write(response.content) return output_path class AsyncDocumentIntelligenceClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawDocumentIntelligenceClient( client_wrapper=client_wrapper ) @property def with_raw_response(self) -> AsyncRawDocumentIntelligenceClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawDocumentIntelligenceClient """ return self._raw_client async def create_job( self, *, language: str = "hi-IN", output_format: str = "html", callback_url: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncDocumentIntelligenceJob: """ Create a new document intelligence job with convenience methods. This is a high-level method that returns an AsyncDocumentIntelligenceJob object with async methods for uploading, starting, waiting, and downloading. Parameters ---------- language : str Language code in BCP-47 format (default: "hi-IN") Supported: hi-IN, en-IN, bn-IN, gu-IN, kn-IN, ml-IN, mr-IN, or-IN, pa-IN, ta-IN, te-IN, ur-IN, as-IN, bodo-IN, doi-IN, ks-IN, kok-IN, mai-IN, mni-IN, ne-IN, sa-IN, sat-IN, sd-IN output_format : str Output format: "html" or "md" (default: "html") callback_url : str, optional Webhook URL for completion notification request_options : RequestOptions, optional Request-specific configuration Returns ------- AsyncDocumentIntelligenceJob A job object with async convenience methods for the workflow Examples -------- import asyncio from sarvamai import AsyncSarvamAI async def main(): client = AsyncSarvamAI(api_subscription_key="YOUR_API_KEY") # Create job job = await client.document_intelligence.create_job( language="hi-IN", output_format="html" ) # Upload, start, wait, download await job.upload_file("document.pdf") await job.start() await job.wait_until_complete() await job.download_output("./output.html") asyncio.run(main()) """ # Build job parameters job_params: DocDigitizationJobParametersParams = { "language": language, "output_format": output_format, } # Build callback if provided callback: typing.Optional[DocDigitizationWebhookCallbackParams] = None if callback_url is not None: callback = {"url": callback_url} # Create the job via the API response = await self.initialise( job_parameters=job_params, callback=callback, request_options=request_options, ) # Return a job object with convenience methods return AsyncDocumentIntelligenceJob( client=self, job_id=response.job_id, language=language, output_format=output_format, ) async def initialise( self, *, job_parameters: typing.Optional[DocDigitizationJobParametersParams] = OMIT, callback: typing.Optional[DocDigitizationWebhookCallbackParams] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> 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:** - `html`: Structured HTML with layout preservation (default) - `md`: Markdown format **Prompt Types:** Customize how specific content types are processed: - `default_ocr`: Standard text extraction (default for all text blocks) - `table_to_html`: Convert tables to HTML format - `table_to_markdown`: Convert tables to Markdown format - `chart_to_markdown`: Extract chart data as Markdown table - `chart_to_json`: Extract chart data as JSON - `describe_image`: Generate image caption - `caption_en`: Same as describe_image (English) - `caption_in`: Caption in document language **Webhook Callback:** Optionally provide a callback URL to receive notification when processing completes. Parameters ---------- job_parameters : typing.Optional[DocDigitizationJobParametersParams] Job configuration parameters. Omit the request body to use defaults. callback : typing.Optional[DocDigitizationWebhookCallbackParams] Optional webhook for completion notification request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DocDigitizationCreateJobResponse Successful Response Examples -------- import asyncio from sarvamai import AsyncSarvamAI client = AsyncSarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) async def main() -> None: await client.document_intelligence.initialise() asyncio.run(main()) """ _response = await self._raw_client.initialise( job_parameters=job_parameters, callback=callback, request_options=request_options, ) return _response.data async def get_upload_links( self, *, job_id: str, files: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None, ) -> 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 ------- DocDigitizationUploadFilesResponse Successful Response Examples -------- import asyncio from sarvamai import AsyncSarvamAI client = AsyncSarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) async def main() -> None: await client.document_intelligence.get_upload_links( job_id="job_id", files=["files"], ) asyncio.run(main()) """ _response = await self._raw_client.get_upload_links( job_id=job_id, files=files, request_options=request_options ) return _response.data async def start( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> 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 ------- DocDigitizationJobStatusResponse Successful Response Examples -------- import asyncio from sarvamai import AsyncSarvamAI client = AsyncSarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) async def main() -> None: await client.document_intelligence.start( job_id="job_id", ) asyncio.run(main()) """ _response = await self._raw_client.start( job_id, request_options=request_options ) return _response.data async def get_status( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> 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 ------- DocDigitizationJobStatusResponse Successful Response Examples -------- import asyncio from sarvamai import AsyncSarvamAI client = AsyncSarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) async def main() -> None: await client.document_intelligence.get_status( job_id="job_id", ) asyncio.run(main()) """ _response = await self._raw_client.get_status( job_id, request_options=request_options ) return _response.data async def get_download_links( self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> 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 ------- DocDigitizationDownloadFilesResponse Successful Response Examples -------- import asyncio from sarvamai import AsyncSarvamAI client = AsyncSarvamAI( api_subscription_key="YOUR_API_SUBSCRIPTION_KEY", ) async def main() -> None: await client.document_intelligence.get_download_links( job_id="job_id", ) asyncio.run(main()) """ _response = await self._raw_client.get_download_links( job_id, request_options=request_options ) return _response.data