Attachments
The attachments module provides file upload functionality and URL refresh capabilities for the Airia platform. Files are uploaded to cloud storage and can be used across various platform features. See the Attachments API reference and response types for more details.
Upload Files
The upload_file method uploads files to cloud storage.
Synchronous Upload
from airia import AiriaClient
# Initialize client (API key can be passed directly or via AIRIA_API_KEY environment variable)
client = AiriaClient(api_key="your_api_key")
# Upload a file
upload_response = client.attachments.upload_file(
file_path="path/to/your/file.pdf"
)
print(f"Uploaded file ID: {upload_response.id}")
print(f"File URL: {upload_response.image_url}")
Asynchronous Upload
import asyncio
from airia import AiriaAsyncClient
async def main():
client = AiriaAsyncClient(api_key="your_api_key")
# Upload a file
upload_response = await client.attachments.upload_file(
file_path="path/to/your/file.pdf"
)
print(f"Uploaded file ID: {upload_response.id}")
print(f"File URL: {upload_response.image_url}")
asyncio.run(main())
Upload Multiple Files
from airia import AiriaClient
client = AiriaClient(api_key="your_api_key")
file_paths = [
"path/to/document.pdf",
"path/to/image.jpg",
"path/to/spreadsheet.xlsx"
]
uploaded_files = []
for file_path in file_paths:
upload_response = client.attachments.upload_file(file_path=file_path)
uploaded_files.append({
"path": file_path,
"id": upload_response.id,
"url": upload_response.image_url
})
for file_info in uploaded_files:
print(f"File: {file_info['path']}")
print(f"ID: {file_info['id']}")
print(f"URL: {file_info['url']}")
print("---")
Integration with Pipeline Execution
Uploaded files can be used directly in pipeline execution. The pipeline execution API supports both local files and remote URLs:
from airia import AiriaClient
client = AiriaClient(api_key="your_api_key")
# Option 1: Upload first, then use in pipeline
upload_response = client.attachments.upload_file(
file_path="path/to/document.pdf"
)
# Use the uploaded file URL in subsequent API calls if needed
print(f"File available at: {upload_response.image_url}")
# Option 2: Use files directly in pipeline execution (recommended)
# The SDK will automatically upload local files internally
response = client.pipeline_execution.execute_pipeline(
pipeline_id="your_pipeline_id",
user_input="Analyze this document",
files=["path/to/document.pdf"] # Automatic upload for local files
)
# Option 3: Use remote URLs directly (no upload needed)
response = client.pipeline_execution.execute_pipeline(
pipeline_id="your_pipeline_id",
user_input="Analyze this remote document",
files=["https://example.com/documents/report.pdf"] # Direct URL usage
)
# Option 4: Mix local files and URLs
response = client.pipeline_execution.execute_pipeline(
pipeline_id="your_pipeline_id",
user_input="Compare these documents",
files=[
"path/to/local-document.pdf", # Local file - uploaded automatically
"https://example.com/remote-document.pdf" # Remote URL - used directly
]
)
print(response.result)
Refresh File URL
The get_file_url method retrieves a refreshed signed URL for an existing attachment. This is useful when the original signed URL has expired or is about to expire, as signed URLs provide time-limited access to files.
Synchronous URL Refresh
from airia import AiriaClient
# Initialize client (API key can be passed directly or via AIRIA_API_KEY environment variable)
client = AiriaClient(api_key="your_api_key")
# First, upload a file
upload_response = client.attachments.upload_file(
file_path="path/to/your/file.pdf"
)
print(f"Uploaded file ID: {upload_response.id}")
print(f"Original URL: {upload_response.image_url}")
# Later, when the URL expires, get a refreshed URL
url_response = client.attachments.get_file_url(
file_id=upload_response.id
)
print(f"Refreshed URL: {url_response.signed_url}")
Asynchronous URL Refresh
import asyncio
from airia import AiriaAsyncClient
async def main():
client = AiriaAsyncClient(api_key="your_api_key")
# First, upload a file
upload_response = await client.attachments.upload_file(
file_path="path/to/your/file.pdf"
)
print(f"Uploaded file ID: {upload_response.id}")
print(f"Original URL: {upload_response.image_url}")
# Later, when the URL expires, get a refreshed URL
url_response = await client.attachments.get_file_url(
file_id=upload_response.id
)
print(f"Refreshed URL: {url_response.signed_url}")
asyncio.run(main())
Complete Workflow Example
from airia import AiriaClient
import time
client = AiriaClient(api_key="your_api_key")
# Step 1: Upload a file
upload_response = client.attachments.upload_file(
file_path="path/to/document.pdf"
)
file_id = upload_response.id
print(f"File uploaded with ID: {file_id}")
# Step 2: Use the file URL immediately
print(f"Initial URL: {upload_response.image_url}")
# Step 3: Later in your application, refresh the URL when needed
# (e.g., when the signed URL is about to expire)
url_response = client.attachments.get_file_url(file_id=file_id)
print(f"Refreshed URL: {url_response.signed_url}")
# The refreshed URL can now be used to access the file
# This is particularly useful for:
# - Long-running applications that need to access files over time
# - Sharing file access with other parts of your application
# - Providing fresh download links to users
Use Cases for URL Refresh
The get_file_url method is particularly useful in scenarios such as:
- Expired URLs: When signed URLs expire and you need fresh access to the file
- Long-Running Applications: Applications that need to access uploaded files over extended periods
- Sharing Access: Generating new URLs to share file access across different parts of your application
- Download Links: Providing users with fresh download links without re-uploading files
- URL Rotation: Security practices that require periodic rotation of access URLs