Get Upload Url

import os
import time
import requests
from opperai import Opper

# Initialize Opper client using your API key
opper = Opper(http_bearer=os.getenv("OPPER_API_KEY"))

# Step 1: Create a new knowledge base
unique_name = f"document_library_{int(time.time())}"
knowledge_base = opper.knowledge.create(
    name=unique_name,
    embedding_model="azure/text-embedding-3-large"
)

print(f"Created knowledge base: {knowledge_base.name}")
print(f"Knowledge base ID: {knowledge_base.id}")

# Step 2: Create a temporary file to upload
filename = "sample_policy.txt"
sample_content = """\
Employee Code of Conduct

1. Professional Behavior
All employees are expected to maintain professional conduct at all times.

2. Confidentiality
Employees must maintain confidentiality of company information and client data.

3. Dress Code
Business casual attire is required for office work.

4. Communication
Use professional communication channels for all business-related correspondence.
"""

with open(filename, "w") as f:
    f.write(sample_content)

print(f"Created sample file: {filename}")

try:
    # Step 3: Get upload URL from Opper
    upload_info = opper.knowledge.get_upload_url(
        knowledge_base_id=knowledge_base.id,
        filename=filename
    )

    print(f"Got upload URL for: {filename}")
    print(f"File ID: {upload_info.id}")

    # Step 4: Upload file to the presigned URL
    with open(filename, "rb") as f:
        files = {"file": f}
        data = upload_info.fields

        upload_response = requests.post(upload_info.url, data=data, files=files)
        upload_response.raise_for_status()

    print(f"Upload status: {upload_response.status_code}")

    # Step 5: Register the uploaded file
    print("Waiting a moment for file to be processed...")
    time.sleep(2)

    retries = 3
    registered_file = None

    while retries > 0:
        try:
            registered_file = opper.knowledge.register_file_upload(
                knowledge_base_id=knowledge_base.id,
                file_id=str(upload_info.id),
                filename=filename,
                content_type="text/plain"
            )
            break
        except Exception as error:
            retries -= 1
            print(f"Registration attempt failed, {retries} retries left. Error: {error}")
            if retries > 0:
                time.sleep(3)
            else:
                raise error

    if registered_file:
        print(f"Registered file: {registered_file.original_filename}")
        print(f"Document ID: {registered_file.document_id}")
    else:
        print("Failed to register file after all retries")

finally:
    if os.path.exists(filename):
        os.remove(filename)
        print(f"Cleaned up temporary file: {filename}")

{
  "url": "<string>",
  "fields": {},
  "id": "3c90c3cc-0d44-4b50-8888-8dd25736052a"
}