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"
}Get upload URL for a knowledge base by its id
Uploading files is a three step process:
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"
}Bearer authentication header of the form Bearer <token>, where <token> is your auth token.
The id of the knowledge base to get the upload URL for
The filename of the file to upload
"example.pdf"