Overview
The FileStore class provides secure file storage with automatic validation, MIME type detection, and configurable limits. It’s designed to handle attachments and files in conversational AI applications.
Creating a FileStore
Recommended: Factory Method
from narrator import FileStore
# Default configuration
store = await FileStore.create()
# Custom directory
store = await FileStore.create("/path/to/files")
# Full configuration
store = await FileStore.create(
base_path="/var/app/files",
max_file_size=100*1024*1024, # 100MB
max_storage_size=10*1024*1024*1024, # 10GB
allowed_mime_types={"image/jpeg", "image/png", "application/pdf"}
)
Direct Constructor
# Creates store without validation
store = FileStore("/path/to/files")
# Validation happens on first use
file_id = await store.save(content, "file.pdf") # Validates here
Configuration Options
base_path
string
default:"./narrator_files"
Base directory for file storage
Maximum file size in bytes (default: 50MB)
Maximum total storage in bytes (default: 5GB)
allowed_mime_types
Set[str]
default:"common types"
Set of allowed MIME types. Default includes documents, images, archives, and audio formats.
Default Allowed File Types
# Documents
'application/pdf'
'application/msword'
'application/vnd.openxmlformats-officedocument.wordprocessingml.document'
'text/plain'
'text/csv'
'application/json'
# Images
'image/jpeg'
'image/png'
'image/gif'
'image/webp'
'image/svg+xml'
# Archives
'application/zip'
'application/x-tar'
'application/gzip'
# Audio
'audio/mpeg'
'audio/mp3'
'audio/wav'
'audio/ogg'
# ... and more
Saving Files
# Save from bytes
file_info = await store.save(
content=pdf_bytes,
filename="report.pdf",
mime_type="application/pdf" # Optional, auto-detected if not provided
)
print(f"File ID: {file_info['id']}")
print(f"Stored at: {file_info['storage_path']}")
print(f"Size: {file_info['size']} bytes")
# Save from base64
import base64
encoded = base64.b64encode(image_bytes).decode()
file_info = await store.save(
content=encoded,
filename="photo.jpg"
)
# Save with metadata
file_info = await store.save(
content=data,
filename="document.docx",
metadata={
"author": "John Doe",
"department": "Sales",
"version": "1.2"
}
)
Retrieving Files
# Get file content
file_id = "abc123..."
storage_path = "2024/01/15/abc123_report.pdf"
content = await store.get(file_id, storage_path)
# Returns bytes
# Get file URL for web access
file_url = FileStore.get_file_url(storage_path)
# Returns: "/files/2024/01/15/abc123_report.pdf"
# Get file metadata
metadata = await store.get_metadata(file_id, storage_path)
print(f"Filename: {metadata['filename']}")
print(f"Size: {metadata['size']}")
print(f"MIME type: {metadata['mime_type']}")
print(f"Created: {metadata['created_at']}")
Deleting Files
# Delete a file
deleted = await store.delete(file_id, storage_path)
print(f"Deleted: {deleted}")
# Delete all files (careful!)
await store.delete_all()
Storage Management
# Check storage usage
current_size = await store.get_storage_size()
print(f"Using {current_size / 1024 / 1024:.2f} MB")
# List all files
files = await store.list_files()
for file_info in files:
print(f"{file_info['filename']}: {file_info['size']} bytes")
# Get file statistics
stats = await store.get_stats()
print(f"Total files: {stats['total_files']}")
print(f"Total size: {stats['total_size_mb']:.2f} MB")
print(f"Average size: {stats['average_size_mb']:.2f} MB")
print(f"Storage used: {stats['storage_percentage']:.1f}%")
# File type breakdown
for mime_type, count in stats['mime_type_counts'].items():
print(f"{mime_type}: {count} files")
Error handling
FileStore includes specific exceptions for different error cases:
from narrator.storage.file_store import (
FileTooLargeError,
UnsupportedFileTypeError,
StorageFullError,
FileNotFoundError
)
try:
# Save a large file
await store.save(large_content, "huge.zip")
except FileTooLargeError as e:
print(f"File too large: {e}")
try:
# Save unsupported type
await store.save(exe_content, "app.exe")
except UnsupportedFileTypeError as e:
print(f"File type not allowed: {e}")
try:
# Storage limit exceeded
await store.save(content, "file.pdf")
except StorageFullError as e:
print(f"Storage full: {e}")
try:
# Retrieve non-existent file
await store.get("bad-id", "bad-path")
except FileNotFoundError as e:
print(f"File not found: {e}")
Integration with Attachments
FileStore is designed to work with the Attachment class:
from narrator import Attachment, FileStore
# Create store
store = await FileStore.create("./uploads")
# Process attachment
attachment = Attachment(
filename="presentation.pdf",
content=pdf_bytes
)
# Store the attachment
await attachment.process_and_store(store)
# Attachment now has storage info
print(f"Stored at: {attachment.storage_path}")
print(f"File ID: {attachment.file_id}")
print(f"URL: {attachment.attributes.get('url')}")
# Later: retrieve content
content = await attachment.get_content_bytes(file_store=store)
File Organization
Files are organized by date for easy management:
narrator_files/
├── 2024/
│ ├── 01/
│ │ ├── 15/
│ │ │ ├── abc123_report.pdf
│ │ │ ├── def456_image.jpg
│ │ │ └── metadata/
│ │ │ ├── abc123.json
│ │ │ └── def456.json
Example: Document Management System
from narrator import FileStore
import asyncio
class DocumentManager:
def __init__(self, store: FileStore):
self.store = store
async def upload_document(self, content: bytes, filename: str,
department: str, doc_type: str):
"""Upload a document with metadata"""
try:
file_info = await self.store.save(
content=content,
filename=filename,
metadata={
"department": department,
"doc_type": doc_type,
"uploaded_by": "current_user",
"upload_time": datetime.now().isoformat()
}
)
return {
"success": True,
"file_id": file_info["id"],
"url": FileStore.get_file_url(file_info["storage_path"])
}
except FileTooLargeError:
return {"success": False, "error": "File exceeds size limit"}
except UnsupportedFileTypeError:
return {"success": False, "error": "File type not allowed"}
async def get_department_documents(self, department: str):
"""Get all documents for a department"""
all_files = await self.store.list_files()
dept_files = []
for file_info in all_files:
metadata = await self.store.get_metadata(
file_info["id"],
file_info["storage_path"]
)
if metadata.get("metadata", {}).get("department") == department:
dept_files.append({
"filename": metadata["filename"],
"size": metadata["size"],
"uploaded": metadata["created_at"],
"type": metadata.get("metadata", {}).get("doc_type"),
"url": FileStore.get_file_url(file_info["storage_path"])
})
return dept_files
async def cleanup_old_files(self, days: int = 30):
"""Remove files older than specified days"""
cutoff = datetime.now() - timedelta(days=days)
all_files = await self.store.list_files()
deleted = 0
for file_info in all_files:
metadata = await self.store.get_metadata(
file_info["id"],
file_info["storage_path"]
)
if metadata["created_at"] < cutoff:
await self.store.delete(
file_info["id"],
file_info["storage_path"]
)
deleted += 1
return deleted
# Usage
store = await FileStore.create(
base_path="/var/documents",
max_file_size=25*1024*1024, # 25MB
max_storage_size=100*1024*1024*1024 # 100GB
)
doc_manager = DocumentManager(store)
# Upload a document
result = await doc_manager.upload_document(
content=pdf_bytes,
filename="Q4_Report.pdf",
department="Finance",
doc_type="quarterly_report"
)
# Get department documents
finance_docs = await doc_manager.get_department_documents("Finance")
for doc in finance_docs:
print(f"{doc['filename']} - {doc['type']} - {doc['url']}")
# Storage maintenance
stats = await store.get_stats()
if stats["storage_percentage"] > 80:
deleted = await doc_manager.cleanup_old_files(60)
print(f"Cleaned up {deleted} old files")
Best practices
- Use the factory method for immediate validation of storage access
- Set appropriate limits based on your application needs
- Monitor storage usage to prevent running out of space
- Use metadata to organize and search files
- Handle errors gracefully - files can fail validation
- Clean up old files periodically to manage storage
- Back up the file directory - FileStore doesn’t handle backups
