# This file was auto-generated by Fern from our API Definition. 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_create import FileCreate from ...types.file_id_presigned_url import FileIdPresignedUrl from ...types.http_validation_error import HttpValidationError from ...types.presigned_url import PresignedUrl from .types.file_create_from_url_resource_info_value import FileCreateFromUrlResourceInfoValue 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 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, request: FileCreate, ) -> 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]. - request: FileCreate. --- from llama_cloud import FileCreate from llama_cloud.client import LlamaCloud client = LlamaCloud( token="YOUR_TOKEN", ) client.files.generate_presigned_url( request=FileCreate( name="string", ), ) """ _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 upload_file_from_url( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, follow_redirects: typing.Optional[bool] = OMIT, name: typing.Optional[str] = OMIT, proxy_url: typing.Optional[str] = OMIT, request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]] = OMIT, url: str, verify_ssl: typing.Optional[bool] = 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]. - follow_redirects: typing.Optional[bool]. Whether to follow redirects when downloading the file - name: typing.Optional[str]. - proxy_url: typing.Optional[str]. - request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]]. - resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]]. - url: str. URL of the file to download - verify_ssl: typing.Optional[bool]. Whether to verify the SSL certificate when downloading the file --- 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 follow_redirects is not OMIT: _request["follow_redirects"] = follow_redirects 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 resource_info is not OMIT: _request["resource_info"] = resource_info if verify_ssl is not OMIT: _request["verify_ssl"] = verify_ssl _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 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 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) class AsyncFilesClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper 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, request: FileCreate, ) -> 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]. - request: FileCreate. --- from llama_cloud import FileCreate from llama_cloud.client import AsyncLlamaCloud client = AsyncLlamaCloud( token="YOUR_TOKEN", ) await client.files.generate_presigned_url( request=FileCreate( name="string", ), ) """ _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 upload_file_from_url( self, *, project_id: typing.Optional[str] = None, organization_id: typing.Optional[str] = None, follow_redirects: typing.Optional[bool] = OMIT, name: typing.Optional[str] = OMIT, proxy_url: typing.Optional[str] = OMIT, request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT, resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]] = OMIT, url: str, verify_ssl: typing.Optional[bool] = 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]. - follow_redirects: typing.Optional[bool]. Whether to follow redirects when downloading the file - name: typing.Optional[str]. - proxy_url: typing.Optional[str]. - request_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]]. - resource_info: typing.Optional[typing.Dict[str, typing.Optional[FileCreateFromUrlResourceInfoValue]]]. - url: str. URL of the file to download - verify_ssl: typing.Optional[bool]. Whether to verify the SSL certificate when downloading the file --- 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 follow_redirects is not OMIT: _request["follow_redirects"] = follow_redirects 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 resource_info is not OMIT: _request["resource_info"] = resource_info if verify_ssl is not OMIT: _request["verify_ssl"] = verify_ssl _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 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 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)