Feat: Implement upload file feature for UTApi and add support for the new UPLOADTHING_TOKEN

.gitignore

@@ -13,6 +13,7 @@
 .python-version
 .vscode/*
 
+.env
 /docs/site/*
 .mypy_cache
 

README.md

@@ -1,58 +1,146 @@
-# Installation
+# UploadThing Python SDK
+
+A Python SDK for [UploadThing](https://uploadthing.com) - the easiest way to add file uploads to your app.
+
+## Installation
 
 ```sh
 pip install uploadthing.py
 ```
 
-# Quickstart
+## Quick Start
+
+### Setting up your token
 
-## Using UTApi
+The SDK uses the `UPLOADTHING_TOKEN` environment variable. You can get this from your UploadThing dashboard.
 
-This is basically a 1:1 clone of the official [TypeScript SDK](https://docs.uploadthing.com/api-reference/ut-api)
+```bash
+export UPLOADTHING_TOKEN="your-token-here"
+```
+
+### Using UTApi
 
-```py
-import asyncio, os
+```python
+import asyncio
+import os
 
-from uploadthing_py import UTApi
+from uploadthing_py import UTApi, UTFile, UploadFiles
 
 
 async def main():
-    utapi = UTApi(os.getenv("UPLOADTHING_SECRET"))
+    # Initialize the client (reads UPLOADTHING_TOKEN from env)
+    utapi = UTApi()
 
-    # List the files in your app
-    res = await utapi.list_files()
-    print("List files:", res)
+    # Upload a file from bytes
+    file = UTFile.from_bytes(b"Hello, World!", "hello.txt")
+    result = await utapi.upload_files(file)
+
+    if result.is_success:
+        print(f"Uploaded: {result.data.url}")
+    else:
+        print(f"Error: {result.error.message}")
+
+    # Upload from a local file
+    file = UTFile.from_path("./image.png")
+    result = await utapi.upload_files(file)
+
+    # Upload from a URL
+    result = await utapi.upload_files_from_url(
+        "https://example.com/image.png"
+    )
+
+    # Upload multiple files concurrently
+    files = [
+        UTFile.from_bytes(b"File 1", "file1.txt"),
+        UTFile.from_bytes(b"File 2", "file2.txt"),
+    ]
+    results = await utapi.upload_files(
+        files,
+        options=UploadFiles.UploadFilesOptions(concurrency=2)
+    )
 
-    # Delete the first file from the list
-    key = res[0].key
-    res = await utapi.delete_file(key)
-    print("Delete file:", res)
+    # List files in your app
+    files = await utapi.list_files()
+    for f in files:
+        print(f"{f.name}: {f.key}")
 
-    # (TODO) Create a new file
-    # res = await utapi.upload_files()
+    # Delete files
+    await utapi.delete_files("file-key")
+
+    # Generate signed URL for private files (no API call needed)
+    signed = utapi.generate_signed_url("file-key")
+    print(signed.ufs_url)
 
 
 if __name__ == "__main__":
     asyncio.run(main())
 ```
 
-## Using FastAPI
+## API Reference
+
+### UTApi
+
+The main client for interacting with UploadThing.
+
+#### Constructor Options
+
+```python
+UTApi(
+    token: str | None = None,        # UPLOADTHING_TOKEN (reads from env if not provided)
+    key_type: str = "file_key",      # Default key type for operations
+    api_url: str | None = None,      # Override API URL
+    ingest_url: str | None = None,   # Override ingest URL
+    ufs_host: str | None = None,     # Override UFS host
+)
+```
+
+#### Methods
+
+| Method                                 | Description                     |
+| -------------------------------------- | ------------------------------- |
+| `upload_files(files, options)`         | Upload files to UploadThing     |
+| `upload_files_from_url(urls, options)` | Download from URLs and upload   |
+| `delete_files(keys, options)`          | Delete files by key or customId |
+| `list_files(options)`                  | List files in your app          |
+| `rename_files(updates)`                | Rename files                    |
+| `get_usage_info()`                     | Get storage usage stats         |
+| `generate_signed_url(key, options)`    | Generate signed URL locally     |
+| `get_signed_url(key, options)`         | Request signed URL via API      |
+| `update_acl(keys, acl, options)`       | Update file access control      |
+
+### UTFile
+
+A file wrapper for uploads.
 
-You can use FastAPI like any of the JavaScript backend adapters.
+```python
+# From bytes
+file = UTFile.from_bytes(b"content", "name.txt")
 
-> [!TIP]
->
-> You can use this example along with one of the [client examples](https://github.com/pingdotgg/uploadthing/tree/main/examples/backend-adapters)
->
-> ```sh
-> UPLOADTHING_SECRET=sk_foo poetry run uvicorn examples.fastapi:app --reload --port 3000
-> ```
+# From file path
+file = UTFile.from_path("./image.png")
+
+# With custom ID
+file = UTFile.from_bytes(b"content", "name.txt", custom_id="my-id")
+```
+
+### Upload Options
+
+```python
+UploadFiles.UploadFilesOptions(
+    content_disposition="inline",  # or "attachment"
+    acl="public-read",            # or "private"
+    concurrency=1,                # Max concurrent uploads (1-25)
+)
+```
+
+## Using with FastAPI
+
+You can use this SDK with FastAPI for client-side uploads:
 
 > [!WARNING]
->
-> This is a work in progress and not yet ready for production use.
+> The FastAPI integration is experimental and not yet production-ready.
 
-```py
+```python
 from fastapi import FastAPI, Request, Response
 from uploadthing_py import (
     UploadThingRequestBody,
@@ -72,29 +160,26 @@ app.add_middleware(
 
 f = create_uploadthing()
 
-
 upload_router = {
-    "videoAndImage": f(
+    "imageUploader": f(
         {
             "image/png": {"max_file_size": "4MB"},
-            "image/heic": {"max_file_size": "16MB"},
+            "image/jpeg": {"max_file_size": "4MB"},
         }
     )
-    .middleware(lambda req: {"user_id": req.headers["x-user-id"]})
-    .on_upload_complete(lambda file, metadata: print(f"Upload complete for {metadata['user_id']}"))
+    .middleware(lambda req: {"user_id": req.headers.get("x-user-id")})
+    .on_upload_complete(
+        lambda file, metadata: print(f"Upload complete for {metadata['user_id']}")
+    )
 }
+
 handlers = create_route_handler(
     router=upload_router,
-    api_key=os.getenv("UPLOADTHING_SECRET"),
+    api_key=os.getenv("UPLOADTHING_SECRET"),  # Legacy API key for handlers
     is_dev=os.getenv("ENVIRONMENT", "development") == "development",
 )
 
 
-@app.get("/api")
-async def greeting():
-    return "Hello from FastAPI"
-
-
 @app.get("/api/uploadthing")
 async def ut_get():
     return handlers["GET"]()
@@ -112,3 +197,26 @@ async def ut_post(
         body=body,
     )
 ```
+
+## Development
+
+```bash
+# Install dependencies
+poetry install
+
+# Run tests
+poetry run pytest
+
+# Run example
+UPLOADTHING_TOKEN="your-token" poetry run python examples/upload_example.py
+
+# Type checking
+poetry run mypy uploadthing_py
+
+# Linting
+poetry run ruff check uploadthing_py
+```
+
+## License
+
+MIT

examples/simple.py

@@ -5,23 +5,28 @@
 import asyncio
 import os
 
-from uploadthing_py import UTApi
+from uploadthing_py import UTApi, UTFile
 
 
 async def main():
-    utapi = UTApi(os.environ["UTAPI_KEY"])
+    utapi = UTApi()
 
     # List the files in your app
     res = await utapi.list_files()
     print("List files:", res)
 
-    # Delete the first file from the list
-    key = res[0].key
-    res = await utapi.delete_file(key)
-    print("Delete file:", res)
-
-    # (TODO) Create a new file
-    # res = await utapi.upload_files()
+    # Upload a file
+    file = UTFile.from_bytes(b"Hello from Python!", "hello.txt")
+    result = await utapi.upload_files(file)
+
+    if result.is_success:
+        print("Upload success:", result.data.url)
+
+        # Delete the uploaded file
+        res = await utapi.delete_files(result.data.key)
+        print("Delete file:", res)
+    else:
+        print("Upload error:", result.error.message)
 
 
 if __name__ == "__main__":

examples/upload_example.py

@@ -0,0 +1,101 @@
+"""
+Example: Upload files using UTApi.
+
+This example demonstrates how to upload files to UploadThing using the Python SDK.
+
+Requirements:
+    - Set UPLOADTHING_TOKEN environment variable with your token
+
+Usage:
+    poetry run python examples/upload_example.py
+"""
+
+import asyncio
+import os
+from pathlib import Path
+
+from uploadthing_py import UTApi, UTFile, UploadFiles
+
+
+async def main():
+    # Initialize the client (reads UPLOADTHING_TOKEN from environment)
+    utapi = UTApi()
+    print(f"Connected to app: {utapi.app_id}")
+
+    # Example 1: Upload a file from bytes
+    print("\n--- Example 1: Upload from bytes ---")
+    content = b"Hello, UploadThing!"
+    file = UTFile.from_bytes(content, "hello.txt")
+    result = await utapi.upload_files(file)
+
+    if result.is_success:
+        print(f"✅ Uploaded: {result.data.name}")
+        print(f"   URL: {result.data.ufs_url}")
+        print(f"   Key: {result.data.key}")
+    else:
+        print(f"❌ Error: {result.error.message}")
+
+    # Example 2: Upload multiple files concurrently
+    print("\n--- Example 2: Upload multiple files ---")
+    files = [
+        UTFile.from_bytes(b"File 1 content", "file1.txt"),
+        UTFile.from_bytes(b"File 2 content", "file2.txt"),
+        UTFile.from_bytes(b"File 3 content", "file3.txt"),
+    ]
+    results = await utapi.upload_files(
+        files,
+        options=UploadFiles.UploadFilesOptions(concurrency=3),
+    )
+
+    for r in results:
+        if r.is_success:
+            print(f"✅ {r.data.name} -> {r.data.ufs_url}")
+        else:
+            print(f"❌ Error: {r.error.message}")
+
+    # Example 3: Upload from a local file path
+    print("\n--- Example 3: Upload from file path ---")
+    # Create a temp file for demo
+    temp_file = Path("temp_upload_test.txt")
+    temp_file.write_text("This is a test file for upload.")
+
+    try:
+        file = UTFile.from_path(temp_file)
+        result = await utapi.upload_files(file)
+
+        if result.is_success:
+            print(f"✅ Uploaded: {result.data.name}")
+            print(f"   URL: {result.data.ufs_url}")
+        else:
+            print(f"❌ Error: {result.error.message}")
+    finally:
+        temp_file.unlink()  # Clean up
+
+    # Example 4: Upload from URL
+    print("\n--- Example 4: Upload from URL ---")
+    result = await utapi.upload_files_from_url(
+        "https://via.placeholder.com/150"
+    )
+
+    if result.is_success:
+        print(f"✅ Uploaded: {result.data.name}")
+        print(f"   URL: {result.data.ufs_url}")
+    else:
+        print(f"❌ Error: {result.error.message}")
+
+    # Example 5: Generate signed URL for private file access
+    print("\n--- Example 5: Generate signed URL ---")
+    if result.is_success:
+        signed = utapi.generate_signed_url(result.data.key)
+        print(f"Signed URL (valid for 5 min): {signed.ufs_url[:80]}...")
+
+    # Example 6: List and manage files
+    print("\n--- Example 6: List files ---")
+    files = await utapi.list_files()
+    print(f"Total files in app: {len(files)}")
+    for f in files[:3]:  # Show first 3
+        print(f"  - {f.name} ({f.key})")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())