feat: add batch upload, CLI, and exception hierarchy

- Add RbxError base class with AuthError, RateLimitError, UploadError,
  and AssetNotFoundError subclasses
- Add BatchUploadItem and BatchResult dataclasses
- Add batch_upload method with semaphore-limited concurrency (2 at a time)
- Add configurable max_attempts and poll_interval to upload_clothing_image
- Add CLI with upload and onsale commands (rbx-upload[cli] extra)
- Improve auth error handling across all client methods
- Bump version to 0.2.0

README.md

@@ -8,6 +8,12 @@ Async Python client for (re)uploading and managing Roblox assets.
 uv add rbx-upload
 ```
 
+With CLI support:
+
+```bash
+uv add "rbx-upload[cli]"
+```
+
 ## Usage
 
 ```python
@@ -43,17 +49,82 @@ async with RobloxClient(
     )
 ```
 
+### Batch upload
+
+```python
+from rbx_upload import RobloxClient, RbxAssetType, BatchUploadItem
+
+items = [
+    BatchUploadItem(image=img1, name="Shirt A", asset_type=RbxAssetType.SHIRT, group_id=67890),
+    BatchUploadItem(image=img2, name="Shirt B", asset_type=RbxAssetType.SHIRT, group_id=67890),
+    BatchUploadItem(image=img3, name="Pants A", asset_type=RbxAssetType.PANTS, group_id=67890),
+]
+
+async with RobloxClient(roblosecurity="...", publisher_user_id=12345) as client:
+    result = await client.batch_upload(items)
+
+    for item, upload in result.succeeded:
+        print(f"{item.name}: asset_id={upload['asset_id']}")
+
+    for item, error in result.failed:
+        print(f"{item.name} failed: {error}")
+```
+
+### Error handling
+
+```python
+from rbx_upload import AuthError, RateLimitError, UploadError, AssetNotFoundError
+
+try:
+    result = await client.upload_clothing_image(...)
+except AuthError:
+    # Invalid or expired ROBLOSECURITY token
+    ...
+except RateLimitError:
+    # Hit rate limit, back off and retry
+    ...
+except UploadError:
+    # Upload operation timed out or failed
+    ...
+```
+
+## CLI
+
+Requires `pip install "rbx-upload[cli]"`. Set `ROBLOSECURITY` in your environment.
+
+```bash
+# Upload a clothing image
+rbx-upload upload shirt.png --name "My Shirt" --group 67890 --publisher 12345
+
+# Put an asset on sale
+rbx-upload onsale 123456789 --name "My Shirt" --group 67890 --publisher 12345 --price 10
+```
+
+Run `rbx-upload --help` or `rbx-upload <command> --help` for all options.
+
 ## API
 
 ### `RobloxClient(roblosecurity, publisher_user_id, proxy=None)`
 
-All methods are async
+All methods are async.
 
 | Method | Description |
 |---|---|
 | `asset_from_id(asset_id)` | Fetch asset info. Returns `ClothingAsset` for shirts/pants, `RbxAsset` otherwise. |
 | `fetch_clothing_image(asset)` | Fetch raw PNG bytes for a clothing asset. |
-| `upload_clothing_image(image, name, description, asset_type, group_id)` | Upload a clothing image. Returns `{"asset_id": int}`. |
+| `upload_clothing_image(image, name, description, asset_type, group_id, max_attempts=10, poll_interval=1.0)` | Upload a clothing image. Returns `{"asset_id": int}`. |
+| `batch_upload(items, max_attempts=10, poll_interval=1.0)` | Upload multiple items (2 at a time). Returns `BatchResult`. |
 | `onsale_asset(asset_id, name, description, group_id, price=5)` | Put an asset on sale. |
 
-`proxy` replaces `roblox.com` in all URLs
+`proxy` replaces `roblox.com` in all URLs.
+
+### Exceptions
+
+All exceptions inherit from `RbxError`.
+
+| Exception | When raised |
+|---|---|
+| `AuthError` | Invalid/expired token or not authorized |
+| `RateLimitError` | HTTP 429 from Roblox |
+| `UploadError` | Upload operation timed out |
+| `AssetNotFoundError` | Asset ID does not exist |

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "rbx-upload"
-version = "0.1.0"
+version = "0.2.0"
 description = "Roblox asset upload client"
 requires-python = ">=3.13"
 dependencies = [
@@ -8,6 +8,12 @@ dependencies = [
 ]
 license = "MIT"
 
+[project.optional-dependencies]
+cli = ["click>=8.0"]
+
+[project.scripts]
+rbx-upload = "rbx_upload.cli:main"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

src/rbx_upload/__init__.py

@@ -1,9 +1,27 @@
-from .client import RobloxClient, RateLimitError
+from .client import RobloxClient
-from .models import RbxAsset, ClothingAsset, RbxCreator, RbxAssetType
+from .models import (
+    AssetNotFoundError,
+    AuthError,
+    BatchResult,
+    BatchUploadItem,
+    ClothingAsset,
+    RateLimitError,
+    RbxAsset,
+    RbxAssetType,
+    RbxCreator,
+    RbxError,
+    UploadError,
+)
 
 __all__ = [
     "RobloxClient",
+    "RbxError",
+    "AuthError",
     "RateLimitError",
+    "UploadError",
+    "AssetNotFoundError",
+    "BatchUploadItem",
+    "BatchResult",
     "RbxAsset",
     "ClothingAsset",
     "RbxCreator",

src/rbx_upload/cli.py

@@ -0,0 +1,103 @@
+import asyncio
+import os
+import sys
+
+try:
+    import click
+except ImportError:
+    print(
+        "CLI dependencies are not installed. Run: pip install rbx-upload[cli]",
+        file=sys.stderr,
+    )
+    sys.exit(1)
+
+from .client import RobloxClient
+from .models import RbxAssetType
+
+
+def _get_roblosecurity() -> str:
+    token = os.environ.get("ROBLOSECURITY")
+    if not token:
+        raise click.ClickException(
+            "ROBLOSECURITY environment variable is not set."
+        )
+    return token
+
+
+@click.group()
+def cli():
+    """rbx-upload — Roblox asset upload tool."""
+    pass
+
+
+@cli.command()
+@click.argument("image", type=click.Path(exists=True, readable=True))
+@click.option("--name", "-n", required=True, help="Asset display name.")
+@click.option("--description", "-d", default="", show_default=True, help="Asset description.")
+@click.option(
+    "--type", "-t", "asset_type",
+    type=click.Choice(["shirt", "pants"], case_sensitive=False),
+    default="shirt",
+    show_default=True,
+    help="Asset type.",
+)
+@click.option("--group", "-g", "group_id", required=True, type=int, help="Group ID to upload to.")
+@click.option("--publisher", "-p", "publisher_user_id", required=True, type=int, help="Publisher user ID.")
+@click.option("--max-attempts", default=10, show_default=True, help="Max polling attempts.")
+@click.option("--poll-interval", default=1.0, show_default=True, help="Seconds between polls.")
+def upload(image, name, description, asset_type, group_id, publisher_user_id, max_attempts, poll_interval):
+    """Upload a clothing image to Roblox."""
+    roblosecurity = _get_roblosecurity()
+    asset_type_enum = RbxAssetType.SHIRT if asset_type == "shirt" else RbxAssetType.PANTS
+
+    with open(image, "rb") as f:
+        image_bytes = f.read()
+
+    async def _run():
+        async with RobloxClient(roblosecurity, publisher_user_id) as client:
+            result = await client.upload_clothing_image(
+                image=image_bytes,
+                name=name,
+                description=description,
+                asset_type=asset_type_enum,
+                group_id=group_id,
+                max_attempts=max_attempts,
+                poll_interval=poll_interval,
+            )
+        return result
+
+    result = asyncio.run(_run())
+    asset_id = result.get("asset_id")
+    if asset_id:
+        click.echo(f"Uploaded successfully. Asset ID: {asset_id}")
+    else:
+        click.echo(f"Upload result: {result}")
+
+
+@cli.command()
+@click.argument("asset_id", type=int)
+@click.option("--name", "-n", required=True, help="Asset display name.")
+@click.option("--description", "-d", default="", show_default=True, help="Asset description.")
+@click.option("--group", "-g", "group_id", required=True, type=int, help="Group ID.")
+@click.option("--publisher", "-p", "publisher_user_id", required=True, type=int, help="Publisher user ID.")
+@click.option("--price", default=5, show_default=True, help="Price in Robux.")
+def onsale(asset_id, name, description, group_id, publisher_user_id, price):
+    """Put an asset on sale."""
+    roblosecurity = _get_roblosecurity()
+
+    async def _run():
+        async with RobloxClient(roblosecurity, publisher_user_id) as client:
+            return await client.onsale_asset(
+                asset_id=asset_id,
+                name=name,
+                description=description,
+                group_id=group_id,
+                price=price,
+            )
+
+    asyncio.run(_run())
+    click.echo(f"Asset {asset_id} put on sale for {price} Robux.")
+
+
+def main():
+    cli()

src/rbx_upload/client.py

@@ -5,13 +5,18 @@ import xml.etree.ElementTree
 
 import httpx
 
-from .models import ClothingAsset, RbxAsset, RbxAssetType, RbxCreator
+from .models import (
-
+    AssetNotFoundError,
-
+    AuthError,
-class RateLimitError(Exception):
+    BatchResult,
-    """Raised when hitting Roblox rate limits (HTTP 429)."""
+    BatchUploadItem,
-
+    ClothingAsset,
-    pass
+    RateLimitError,
+    RbxAsset,
+    RbxAssetType,
+    RbxCreator,
+    UploadError,
+)
 
 
 class RobloxClient:
@@ -49,11 +54,9 @@ class RobloxClient:
         )
         csrf = response.headers.get("X-CSRF-TOKEN")
         if not csrf:
-            raise httpx.HTTPStatusError(
+            if response.status_code in (401, 403):
-                "Failed to retrieve X-CSRF-TOKEN.",
+                raise AuthError("Invalid or expired ROBLOSECURITY token.")
-                request=response.request,
+            raise AuthError("Failed to retrieve X-CSRF-TOKEN.")
-                response=response,
-            )
         return csrf
 
     async def _economy_request(self, asset_id: int) -> httpx.Response:
@@ -89,6 +92,10 @@ class RobloxClient:
     async def asset_from_id(self, asset_id: int) -> RbxAsset:
         """Fetch asset information from Roblox by asset ID."""
         response = await self._economy_request(asset_id)
+        if response.status_code == 404:
+            raise AssetNotFoundError(f"Asset {asset_id} not found.")
+        if response.status_code in (401, 403):
+            raise AuthError("Not authorized to fetch this asset.")
         response.raise_for_status()
         asset_info = response.json()
         creator_info = asset_info["Creator"]
@@ -129,8 +136,20 @@ class RobloxClient:
         description: str,
         asset_type: RbxAssetType,
         group_id: int,
+        max_attempts: int = 10,
+        poll_interval: float = 1.0,
     ) -> dict:
-        """Upload a clothing image to Roblox and return the operation result."""
+        """Upload a clothing image to Roblox and return the operation result.
+
+        Args:
+            image: Raw PNG bytes of the clothing image.
+            name: Display name for the asset.
+            description: Description for the asset.
+            asset_type: RbxAssetType.SHIRT or RbxAssetType.PANTS.
+            group_id: ID of the group to upload the asset to.
+            max_attempts: Number of times to poll the operation status. Defaults to 10.
+            poll_interval: Seconds to wait between polls. Defaults to 1.0.
+        """
         csrf = await self._get_csrf_token()
         upload_url = self._proxy_url(
             "https://apis.roblox.com/assets/user-auth/v1/assets"
@@ -166,15 +185,17 @@ class RobloxClient:
         )
 
         if response.status_code == 429:
-            raise RateLimitError("Rate limit hit during upload")
+            raise RateLimitError("Rate limit hit during upload.")
+        if response.status_code in (401, 403):
+            raise AuthError("Not authorized to upload assets.")
 
         response.raise_for_status()
         data = response.json()
 
         operation_id = data.get("operationId")
         if operation_id:
-            for _ in range(10):
+            for _ in range(max_attempts):
-                await asyncio.sleep(1)
+                await asyncio.sleep(poll_interval)
                 op_response = await self._http.get(
                     self._proxy_url(
                         f"https://apis.roblox.com/assets/user-auth/v1/operations/{operation_id}"
@@ -188,9 +209,50 @@ class RobloxClient:
                     if op_data.get("response", {}).get("assetId"):
                         return {"asset_id": op_data["response"]["assetId"]}
                     return op_data
+            raise UploadError(
+                f"Upload operation did not complete after {max_attempts} attempts."
+            )
 
         return data
 
+    async def batch_upload(
+        self,
+        items: list[BatchUploadItem],
+        max_attempts: int = 10,
+        poll_interval: float = 1.0,
+    ) -> BatchResult:
+        """Upload multiple clothing images with limited concurrency.
+
+        Processes items 2 at a time. Continues on failure and reports all
+        failures in the returned BatchResult.
+
+        Args:
+            items: List of BatchUploadItem to upload.
+            max_attempts: Passed to each upload_clothing_image call.
+            poll_interval: Passed to each upload_clothing_image call.
+        """
+        result = BatchResult()
+        semaphore = asyncio.Semaphore(2)
+
+        async def _upload_one(item: BatchUploadItem):
+            async with semaphore:
+                try:
+                    upload_result = await self.upload_clothing_image(
+                        image=item.image,
+                        name=item.name,
+                        description=item.description,
+                        asset_type=item.asset_type,
+                        group_id=item.group_id,
+                        max_attempts=max_attempts,
+                        poll_interval=poll_interval,
+                    )
+                    result.succeeded.append((item, upload_result))
+                except Exception as e:
+                    result.failed.append((item, e))
+
+        await asyncio.gather(*[_upload_one(item) for item in items])
+        return result
+
     async def onsale_asset(
         self,
         asset_id: int,
@@ -232,7 +294,9 @@ class RobloxClient:
         )
 
         if response.status_code == 429:
-            raise RateLimitError("Rate limit hit during onsale")
+            raise RateLimitError("Rate limit hit during onsale.")
+        if response.status_code in (401, 403):
+            raise AuthError("Not authorized to put this asset on sale.")
 
         response.raise_for_status()
         return response.json()

src/rbx_upload/models.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass, field
 from enum import IntEnum
 from typing import Literal
 
@@ -15,6 +16,31 @@ ClothingAssetType = Literal[
 CreatorType = Literal["User", "Group"]
 
 
+class RbxError(Exception):
+    """Base exception for all rbx-upload errors."""
+    pass
+
+
+class AuthError(RbxError):
+    """Raised when authentication fails or the ROBLOSECURITY token is invalid."""
+    pass
+
+
+class RateLimitError(RbxError):
+    """Raised when hitting Roblox rate limits (HTTP 429)."""
+    pass
+
+
+class UploadError(RbxError):
+    """Raised when an asset upload fails."""
+    pass
+
+
+class AssetNotFoundError(RbxError):
+    """Raised when an asset cannot be found."""
+    pass
+
+
 class RbxCreator:
     def __init__(self, creator_id: int, username: str, creator_type: CreatorType):
         self.creator_id = creator_id
@@ -54,3 +80,22 @@ class ClothingAsset(RbxAsset):
             description=description,
             asset_type=asset_type,
         )
+
+
+@dataclass
+class BatchUploadItem:
+    image: bytes
+    name: str
+    asset_type: RbxAssetType
+    group_id: int
+    description: str = ""
+
+
+@dataclass
+class BatchResult:
+    succeeded: list[tuple[BatchUploadItem, dict]] = field(default_factory=list)
+    failed: list[tuple[BatchUploadItem, Exception]] = field(default_factory=list)
+
+    @property
+    def all_succeeded(self) -> bool:
+        return len(self.failed) == 0