#Python SDK

from filenest import FileNest, AsyncFileNest
 
# Synchronous
fn = FileNest(
    api_key="fn_live_...",
    project_id="proj_01j...",
)
 
# Asynchronous (context manager recommended)
async with AsyncFileNest(api_key="fn_live_...", project_id="proj_01j...") as fn:
    ...

# From bytes
with open("report.pdf", "rb") as f:
    file = fn.files.upload(
        filename="report.pdf",
        data=f.read(),
        mime_type="application/pdf",
        folder_id="fld_01j...",     # optional
        tags=["finance", "2026"],   # optional
        metadata={"year": "2026"},  # optional
    )
 
# From a filesystem path (MIME type auto-detected)
file = fn.files.upload_from_path(
    path="./report.pdf",
    folder_id="fld_01j...",
    tags=["finance"],
)

# Get a presigned URL
url_info = fn.files.get_download_url(file.id, ttl=3600)
print(url_info.url)
 
# Save directly to disk
fn.files.download_to_path(file.id, dest="./output.pdf")
 
# Load into memory
content = fn.files.download_to_bytes(file.id)

result = fn.files.list(
    status="ready",
    mime_type="application/pdf",
    tags=["finance"],
    sort_by="created_at",
    sort_order="desc",
    limit=50,
    offset=0,
)
for f in result.items:
    print(f.filename)

file = fn.files.get(file_id)
 
fn.files.update(
    file_id,
    filename="report-final.pdf",
    tags=["archived"],
    metadata={"reviewed": True},
)
 
fn.files.delete(file_id)   # soft delete — recoverable
 
# Restore a soft-deleted file
file = fn.files.restore(file_id)

versions = fn.files.list_versions(file_id)
 
# Restore to a previous version
fn.files.restore_version(file_id, version_id)

# Create a single folder
folder = fn.folders.create("invoices")
folder = fn.folders.create("q1", parent_folder_id=folder["id"], metadata={"dept": "finance"})
 
# Idempotent: create every missing path segment, return the leaf folder
folder = fn.folders.ensure_path("users/alice/uploads")

# Returns None if the path does not exist
folder = fn.folders.get_by_path("users/alice/uploads")
 
folder = fn.folders.get(folder_id)

result = fn.folders.list()
result = fn.folders.list(name="uploads")   # name filter
 
# List files inside a folder with filters
files = fn.folders.list_files(
    folder_id,
    q="invoice",           # filename search
    tags=["finance"],
    category="document",
    status="ready",
    limit=20,
    cursor=last_cursor,    # cursor-based pagination
)

# Fails with an error if the folder contains files or subfolders
fn.folders.delete(folder_id)

# 1. Create a session and store upload_id somewhere durable
session = fn.uploads.create(
    filename="large-video.mp4",
    size_bytes=len(data),
    mime_type="video/mp4",
    folder_id="fld_01j...",
    tags=["video", "raw"],
    metadata={"uploaded_by": "user_abc"},
)
# session["upload_id"] → persist this to resume later
 
# 2. Upload all parts (or resume from where you left off)
file = fn.uploads.resume(
    session["upload_id"],
    data=data,
    on_progress=lambda pct: print(f"{pct}%"),
)
 
# 3. Abort if no longer needed
fn.uploads.abort(session["upload_id"])

token = fn.upload_tokens.create(
    folder_id=folder["id"],
    owner_user_id="user_abc",
    owner_org_id="org_xyz",
    max_size=50 * 1024 * 1024,                           # 50 MB per file
    max_files=10,
    allowed_mime_types=["image/*", "application/pdf"],
    metadata={"uploaded_from": "profile-settings"},      # stamped on every file
    tags=["avatar", "user-content"],                     # merged onto every file
    expires_in=3600,
)
print(token.token)       # fn_upload_token_...  — return this to the browser
print(token.expires_at)  # ISO timestamp — return this for client-side refresh

# Create an endpoint
wh = fn.webhooks.create(
    name="prod-receiver",
    url="https://app.example.com/webhooks/filenest",
    events=["file.uploaded", "file.ready", "file.failed"],
)
# wh["signing_secret"] — shown only once, store safely
 
result = fn.webhooks.list()
wh = fn.webhooks.get(webhook_id)
fn.webhooks.update(webhook_id, is_active=False)
fn.webhooks.delete(webhook_id)
 
deliveries = fn.webhooks.list_deliveries(webhook_id, limit=20)
 
# Verify a payload
is_valid = fn.webhooks.verify(raw_body, signature_header, secret)

from filenest.namespaces.webhooks import verify_webhook_signature
 
is_valid = verify_webhook_signature(raw_body, signature_header, secret)

results = fn.search.query(
    q="invoice",
    filters={"status": "ready", "tags": ["finance"]},
    facets=["mime_type", "tags"],
    sort_by="created_at",
    sort_order="desc",
    limit=20,
)
 
# Iterate over all results with automatic pagination
for file in fn.search.iterate(q="invoice"):
    print(file.filename)

from fastapi import FastAPI, UploadFile, Request, HTTPException
from filenest import AsyncFileNest
from filenest.namespaces.webhooks import verify_webhook_signature
 
app = FastAPI()
fn = AsyncFileNest(api_key="fn_live_...", project_id="proj_01j...")
 
 
@app.post("/upload")
async def upload(file: UploadFile, user_id: str):
    folder = await fn.folders.ensure_path(f"users/{user_id}/uploads")
    result = await fn.files.upload(
        filename=file.filename,
        data=await file.read(),
        mime_type=file.content_type,
        folder_id=folder["id"],
    )
    return {"file_id": result.id, "status": result.status}
 
 
@app.post("/webhooks/filenest")
async def webhook(request: Request):
    body = await request.body()
    sig = request.headers.get("x-filenest-signature", "")
    if not verify_webhook_signature(body, sig, secret="..."):
        raise HTTPException(status_code=401, detail="Invalid signature")
    event = await request.json()
    print(event["type"], event["data"])
    return {"ok": True}

async with AsyncFileNest(api_key="fn_live_...", project_id="proj_...") as fn:
    folder = await fn.folders.ensure_path("users/alice/uploads")
 
    token = await fn.upload_tokens.create(
        folder_id=folder["id"],
        owner_user_id="alice",
        tags=["user-content"],
        expires_in=3600,
    )
 
    files = await fn.folders.list_files(folder["id"], limit=20)
 
    # Resumable multipart
    session = await fn.uploads.create(filename="video.mp4", size_bytes=len(data))
    file = await fn.uploads.resume(session["upload_id"], data=data)