Upload Files

Set up environment

1%pip install -U athena-intelligence
1import os
2import io
3import asyncio
4from athena.client import Athena
5
6ATHENA_API_KEY = os.environ["ATHENA_API_KEY"]
7
8athena = Athena(
9    api_key=ATHENA_API_KEY,
10)

Upload a file

To upload a file using the Athena SDK, you can use the upload.upload_documents() method. This method accepts a list of file tuples, where each tuple contains the filename, file content as a BytesIO object, and the MIME type.

Here’s an example of how to upload an Excel file:

1async def upload_file():
2    # Prepare the file for upload
3    file_bytes = io.BytesIO()
4    with open("your_file.xlsx", "rb") as f:
5        file_bytes.write(f.read())
6        file_bytes.seek(0)  # Reset the cursor of the BytesIO object
7
8    # Create the file tuple
9    file_tuple = (
10        "your_file.xlsx",
11        file_bytes,
12        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
13    )
14
15    # Upload the file
16    result = await athena.upload.upload_documents(files=[file_tuple])
17    print(result)
18
19# Run the async function
20await upload_file()

In this example:

  1. We open the file and read its contents into a BytesIO object.
  2. We create a tuple containing the filename, the BytesIO object, and the MIME type.
  3. We call athena.upload.upload_documents() with a list containing our file tuple.
  4. The function returns the result of the upload operation.

Upload multiple files

You can upload multiple files in a single request by adding more file tuples to the list:

1async def upload_multiple_files():
2    files = [
3        ("file1.xlsx", file1_bytes, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"),
4        ("file2.csv", file2_bytes, "text/csv"),
5        ("file3.pdf", file3_bytes, "application/pdf")
6    ]
7
8    result = await athena.upload.upload_documents(files=files)
9    print(result)
10
11await upload_multiple_files()

Using with FastAPI

If you’re using FastAPI and want to upload files received from a client, you can use the UploadFile object:

1from fastapi import FastAPI, UploadFile
2
3app = FastAPI()
4
5@app.post("/upload/")
6async def upload_file(file: UploadFile):
7    file_tuple = (file.filename, file.file, file.content_type)
8    result = await athena.upload.upload_documents(files=[file_tuple])
9    return {"message": "File uploaded successfully", "result": result}

This endpoint will accept file uploads and forward them to the Athena API using the SDK.

Remember to handle exceptions and implement proper error checking in your production code.