# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing import urllib.parse from json.decoder import JSONDecodeError from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.jsonable_encoder import jsonable_encoder from ...core.remove_none_from_dict import remove_none_from_dict from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.file import File from ...types.file_id_presigned_url import FileIdPresignedUrl from ...types.http_validation_error import HttpValidationError from ...types.page_figure_metadata import PageFigureMetadata from ...types.page_screenshot_metadata import PageScreenshotMetadata from ...types.presigned_url import PresignedUrl from .types.file_create_from_url_resource_info_value import FileCreateFromUrlResourceInfoValue from .types.file_create_permission_info_value import FileCreatePermissionInfoValue from .types.file_create_resource_info_value import FileCreateResourceInfoValue try: import pydantic if pydantic.__version__.startswith("1."): raise ImportError import pydantic.v1 as pydantic # type: ignore except ImportError: import pydantic # type: ignore # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class FilesClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get_file( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> File: """ Read File metadata objects. Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.get_file( id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(File, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def delete_file( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> None: """ Delete the file from S3. Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.delete_file( id="string", ) """ _response = self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def list_files( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[File]: """ Read File metadata objects. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.list_files() """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[File], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def upload_file( self, *, external_file_id: typing.Optional[str] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, upload_file: typing.IO, ) -> File: """ Upload a file to S3. Parameters: - external_file_id: typing.Optional[str]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - upload_file: typing.IO. """ _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files"), params=remove_none_from_dict( {"external_file_id": external_file_id, "project_id": project_id, "organization_id": organization_id} ), data=jsonable_encoder({}), files={"upload_file": upload_file}, headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(File, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def generate_presigned_url( self, *, expires_at_seconds: typing.Optional[int] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, name: str, external_file_id: typing.Optional[str] = OMIT, file_size: typing.Optional[int] = OMIT, last_modified_at: typing.Optional[dt.datetime] = OMIT, resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateResourceInfoValue]]] = OMIT, permission_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreatePermissionInfoValue]]] = OMIT, data_source_id: typing.Optional[str] = OMIT, ) -> FileIdPresignedUrl: """ Create a presigned url for uploading a file. The presigned url is valid for a limited time period, after which it will expire. Be careful on accidental exposure of the presigned url, as it may allow unauthorized access to the file before the expiration. Parameters: - expires_at_seconds: typing.Optional[int]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - name: str. Name that will be used for created file. If possible, always include the file extension in the name. - external_file_id: typing.Optional[str]. - file_size: typing.Optional[int]. - last_modified_at: typing.Optional[dt.datetime]. - resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateResourceInfoValue]]]. - permission_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreatePermissionInfoValue]]]. - data_source_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.generate_presigned_url( name="string", ) """ _request: typing.Dict[str, typing.Any] = {"name": name} if external_file_id is not OMIT: _request["external_file_id"] = external_file_id if file_size is not OMIT: _request["file_size"] = file_size if last_modified_at is not OMIT: _request["last_modified_at"] = last_modified_at if resource_info is not OMIT: _request["resource_info"] = resource_info if permission_info is not OMIT: _request["permission_info"] = permission_info if data_source_id is not OMIT: _request["data_source_id"] = data_source_id _response = self._client_wrapper.httpx_client.request( "PUT", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files"), params=remove_none_from_dict( {"expires_at_seconds": expires_at_seconds, "project_id": project_id, "organization_id": organization_id} ), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(FileIdPresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def sync_files( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[File]: """ Sync Files API against file contents uploaded via S3 presigned urls. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.sync_files() """ _response = self._client_wrapper.httpx_client.request( "PUT", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files/sync"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[File], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def upload_file_from_url( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, name: typing.Optional[str] = OMIT, url: str, proxy_url: typing.Optional[str] = OMIT, request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, verify_ssl: typing.Optional[bool] = OMIT, follow_redirects: typing.Optional[bool] = OMIT, resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]] = OMIT, ) -> File: """ Upload a file to the project from a URL. If name is ommitted in the request payload, the file name will be extracted from the response Content-Disposition header if available or otherwise it will be derived from the URL path. If providing the name in the request payload, always suffix the file extension in the name if available. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - name: typing.Optional[str]. - url: str. URL of the file to download - proxy_url: typing.Optional[str]. - request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]]. - verify_ssl: typing.Optional[bool]. Whether to verify the SSL certificate when downloading the file - follow_redirects: typing.Optional[bool]. Whether to follow redirects when downloading the file - resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.upload_file_from_url( url="string", ) """ _request: typing.Dict[str, typing.Any] = {"url": url} if name is not OMIT: _request["name"] = name if proxy_url is not OMIT: _request["proxy_url"] = proxy_url if request_headers is not OMIT: _request["request_headers"] = request_headers if verify_ssl is not OMIT: _request["verify_ssl"] = verify_ssl if follow_redirects is not OMIT: _request["follow_redirects"] = follow_redirects if resource_info is not OMIT: _request["resource_info"] = resource_info _response = self._client_wrapper.httpx_client.request( "PUT", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files/upload_from_url"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(File, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def read_file_content( self, id: str, *, expires_at_seconds: typing.Optional[int] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> PresignedUrl: """ Returns a presigned url to read the file content. Parameters: - id: str. - expires_at_seconds: typing.Optional[int]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.read_file_content( id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/content"), params=remove_none_from_dict( {"expires_at_seconds": expires_at_seconds, "project_id": project_id, "organization_id": organization_id} ), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(PresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def list_file_page_screenshots( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[PageScreenshotMetadata]: """ List metadata for all screenshots of pages from a file. Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.list_file_page_screenshots( id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page_screenshots"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[PageScreenshotMetadata], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_file_page_screenshot( self, id: str, page_index: int, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Get screenshot of a page from a file. Parameters: - id: str. - page_index: int. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.get_file_page_screenshot( id="string", page_index=1, ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page_screenshots/{page_index}" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def list_file_pages_figures( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[PageFigureMetadata]: """ Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.list_file_pages_figures( id="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[PageFigureMetadata], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def list_file_page_figures( self, id: str, page_index: int, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.List[PageFigureMetadata]: """ Parameters: - id: str. - page_index: int. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.list_file_page_figures( id="string", page_index=1, ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures/{page_index}" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[PageFigureMetadata], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def get_file_page_figure( self, id: str, page_index: int, figure_name: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Parameters: - id: str. - page_index: int. - figure_name: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.get_file_page_figure( id="string", page_index=1, figure_name="string", ) """ _response = self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures/{page_index}/{figure_name}" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def generate_file_page_screenshot_presigned_url( self, id: str, page_index: int, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> PresignedUrl: """ Returns a presigned url to read a page screenshot. The presigned url is valid for a limited time period, after which it will expire. Be careful on accidental exposure of the presigned url, as it may allow unauthorized access to the file before the expiration. Parameters: - id: str. - page_index: int. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.generate_file_page_screenshot_presigned_url( id="string", page_index=1, ) """ _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page_screenshots/{page_index}/presigned_url", ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(PresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) def generate_file_page_figure_presigned_url( self, id: str, page_index: int, figure_name: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> PresignedUrl: """ Returns a presigned url to read a page figure. The presigned url is valid for a limited time period, after which it will expire. Be careful on accidental exposure of the presigned url, as it may allow unauthorized access to the file before the expiration. Parameters: - id: str. - page_index: int. - figure_name: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.generate_file_page_figure_presigned_url( id="string", page_index=1, figure_name="string", ) """ _response = self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures/{page_index}/{figure_name}/presigned_url", ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(PresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) class AsyncFilesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get_file( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> File: """ Read File metadata objects. Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.get_file( id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(File, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def delete_file( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> None: """ Delete the file from S3. Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.delete_file( id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "DELETE", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def list_files( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[File]: """ Read File metadata objects. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.list_files() """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[File], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def upload_file( self, *, external_file_id: typing.Optional[str] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, upload_file: typing.IO, ) -> File: """ Upload a file to S3. Parameters: - external_file_id: typing.Optional[str]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - upload_file: typing.IO. """ _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files"), params=remove_none_from_dict( {"external_file_id": external_file_id, "project_id": project_id, "organization_id": organization_id} ), data=jsonable_encoder({}), files={"upload_file": upload_file}, headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(File, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def generate_presigned_url( self, *, expires_at_seconds: typing.Optional[int] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, name: str, external_file_id: typing.Optional[str] = OMIT, file_size: typing.Optional[int] = OMIT, last_modified_at: typing.Optional[dt.datetime] = OMIT, resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateResourceInfoValue]]] = OMIT, permission_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreatePermissionInfoValue]]] = OMIT, data_source_id: typing.Optional[str] = OMIT, ) -> FileIdPresignedUrl: """ Create a presigned url for uploading a file. The presigned url is valid for a limited time period, after which it will expire. Be careful on accidental exposure of the presigned url, as it may allow unauthorized access to the file before the expiration. Parameters: - expires_at_seconds: typing.Optional[int]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - name: str. Name that will be used for created file. If possible, always include the file extension in the name. - external_file_id: typing.Optional[str]. - file_size: typing.Optional[int]. - last_modified_at: typing.Optional[dt.datetime]. - resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateResourceInfoValue]]]. - permission_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreatePermissionInfoValue]]]. - data_source_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.generate_presigned_url( name="string", ) """ _request: typing.Dict[str, typing.Any] = {"name": name} if external_file_id is not OMIT: _request["external_file_id"] = external_file_id if file_size is not OMIT: _request["file_size"] = file_size if last_modified_at is not OMIT: _request["last_modified_at"] = last_modified_at if resource_info is not OMIT: _request["resource_info"] = resource_info if permission_info is not OMIT: _request["permission_info"] = permission_info if data_source_id is not OMIT: _request["data_source_id"] = data_source_id _response = await self._client_wrapper.httpx_client.request( "PUT", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files"), params=remove_none_from_dict( {"expires_at_seconds": expires_at_seconds, "project_id": project_id, "organization_id": organization_id} ), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(FileIdPresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def sync_files( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[File]: """ Sync Files API against file contents uploaded via S3 presigned urls. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.sync_files() """ _response = await self._client_wrapper.httpx_client.request( "PUT", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files/sync"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[File], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def upload_file_from_url( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, name: typing.Optional[str] = OMIT, url: str, proxy_url: typing.Optional[str] = OMIT, request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, verify_ssl: typing.Optional[bool] = OMIT, follow_redirects: typing.Optional[bool] = OMIT, resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]] = OMIT, ) -> File: """ Upload a file to the project from a URL. If name is ommitted in the request payload, the file name will be extracted from the response Content-Disposition header if available or otherwise it will be derived from the URL path. If providing the name in the request payload, always suffix the file extension in the name if available. Parameters: - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. - name: typing.Optional[str]. - url: str. URL of the file to download - proxy_url: typing.Optional[str]. - request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]]. - verify_ssl: typing.Optional[bool]. Whether to verify the SSL certificate when downloading the file - follow_redirects: typing.Optional[bool]. Whether to follow redirects when downloading the file - resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.upload_file_from_url( url="string", ) """ _request: typing.Dict[str, typing.Any] = {"url": url} if name is not OMIT: _request["name"] = name if proxy_url is not OMIT: _request["proxy_url"] = proxy_url if request_headers is not OMIT: _request["request_headers"] = request_headers if verify_ssl is not OMIT: _request["verify_ssl"] = verify_ssl if follow_redirects is not OMIT: _request["follow_redirects"] = follow_redirects if resource_info is not OMIT: _request["resource_info"] = resource_info _response = await self._client_wrapper.httpx_client.request( "PUT", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", "api/v1/files/upload_from_url"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), json=jsonable_encoder(_request), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(File, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def read_file_content( self, id: str, *, expires_at_seconds: typing.Optional[int] = None, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> PresignedUrl: """ Returns a presigned url to read the file content. Parameters: - id: str. - expires_at_seconds: typing.Optional[int]. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.read_file_content( id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/content"), params=remove_none_from_dict( {"expires_at_seconds": expires_at_seconds, "project_id": project_id, "organization_id": organization_id} ), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(PresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def list_file_page_screenshots( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[PageScreenshotMetadata]: """ List metadata for all screenshots of pages from a file. Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.list_file_page_screenshots( id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page_screenshots"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[PageScreenshotMetadata], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_file_page_screenshot( self, id: str, page_index: int, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Get screenshot of a page from a file. Parameters: - id: str. - page_index: int. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.get_file_page_screenshot( id="string", page_index=1, ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page_screenshots/{page_index}" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def list_file_pages_figures( self, id: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None ) -> typing.List[PageFigureMetadata]: """ Parameters: - id: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.list_file_pages_figures( id="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin(f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures"), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[PageFigureMetadata], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def list_file_page_figures( self, id: str, page_index: int, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.List[PageFigureMetadata]: """ Parameters: - id: str. - page_index: int. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.list_file_page_figures( id="string", page_index=1, ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures/{page_index}" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.List[PageFigureMetadata], _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def get_file_page_figure( self, id: str, page_index: int, figure_name: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> typing.Any: """ Parameters: - id: str. - page_index: int. - figure_name: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.get_file_page_figure( id="string", page_index=1, figure_name="string", ) """ _response = await self._client_wrapper.httpx_client.request( "GET", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures/{page_index}/{figure_name}" ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(typing.Any, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def generate_file_page_screenshot_presigned_url( self, id: str, page_index: int, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> PresignedUrl: """ Returns a presigned url to read a page screenshot. The presigned url is valid for a limited time period, after which it will expire. Be careful on accidental exposure of the presigned url, as it may allow unauthorized access to the file before the expiration. Parameters: - id: str. - page_index: int. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.generate_file_page_screenshot_presigned_url( id="string", page_index=1, ) """ _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page_screenshots/{page_index}/presigned_url", ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(PresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json) async def generate_file_page_figure_presigned_url( self, id: str, page_index: int, figure_name: str, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, ) -> PresignedUrl: """ Returns a presigned url to read a page figure. The presigned url is valid for a limited time period, after which it will expire. Be careful on accidental exposure of the presigned url, as it may allow unauthorized access to the file before the expiration. Parameters: - id: str. - page_index: int. - figure_name: str. - project_id: typing.Optional[str]. - organization_id: typing.Optional[str]. --- from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.generate_file_page_figure_presigned_url( id="string", page_index=1, figure_name="string", ) """ _response = await self._client_wrapper.httpx_client.request( "POST", urllib.parse.urljoin( f"{self._client_wrapper.get_base_url()}/", f"api/v1/files/{id}/page-figures/{page_index}/{figure_name}/presigned_url", ), params=remove_none_from_dict({"project_id": project_id, "organization_id": organization_id}), headers=self._client_wrapper.get_headers(), timeout=60, ) if 200 <= _response.status_code < 300: return pydantic.parse_obj_as(PresignedUrl, _response.json()) # type: ignore if _response.status_code == 422: raise UnprocessableEntityError(pydantic.parse_obj_as(HttpValidationError, _response.json())) # type: ignore try: _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, body=_response.text) raise ApiError(status_code=_response.status_code, body=_response_json)