Client Level API

Client Level API Key + Tests

Author: justinwlin

.gitignore

@@ -8,6 +8,7 @@ test_logging.py
 /example_project
 IGNORE.py
 /quick-test
+/local-test
 .DS_Store
 
 # Byte-compiled / optimized / DLL files

README.md

@@ -30,18 +30,37 @@ Welcome to the official Python library for Runpod API & SDK.
 
 ## 💻 | Installation
 
+### Install from PyPI (Stable Release)
+
 ```bash
-# Install the latest release version
+# Install with pip
 pip install runpod
 
-# Using uv (faster alternative)
+# Install with uv (faster alternative)
 uv add runpod
+```
+
+### Install from GitHub (Latest Changes)
 
-# Install the latest development version (main branch)
+To get the latest changes that haven't been released to PyPI yet:
+
+```bash
+# Install latest development version from main branch with pip
 pip install git+https://github.com/runpod/runpod-python.git
 
-# Or with uv
+# Install with uv
 uv add git+https://github.com/runpod/runpod-python.git
+
+# Install a specific branch
+pip install git+https://github.com/runpod/runpod-python.git@branch-name
+
+# Install a specific tag/release
+pip install git+https://github.com/runpod/[email protected]
+
+# Install in editable mode for development
+git clone https://github.com/runpod/runpod-python.git
+cd runpod-python
+pip install -e .
 ```
 
 *Python 3.8 or higher is required to use the latest version of this package.*
@@ -101,6 +120,8 @@ runpod.api_key = "your_runpod_api_key_found_under_settings"
 
 You can interact with Runpod endpoints via a `run` or `run_sync` method.
 
+#### Basic Usage
+
 ```python
 endpoint = runpod.Endpoint("ENDPOINT_ID")
 
@@ -126,6 +147,99 @@ run_request = endpoint.run_sync(
 print(run_request )
 ```
 
+#### API Key Management
+
+The SDK supports multiple ways to set API keys:
+
+**1. Global API Key** (Default)
+```python
+import runpod
+
+# Set global API key
+runpod.api_key = "your_runpod_api_key"
+
+# All endpoints will use this key by default
+endpoint = runpod.Endpoint("ENDPOINT_ID")
+result = endpoint.run_sync({"input": "data"})
+```
+
+**2. Endpoint-Specific API Key**
+```python
+# Create endpoint with its own API key
+endpoint = runpod.Endpoint("ENDPOINT_ID", api_key="specific_api_key")
+
+# This endpoint will always use the provided API key
+result = endpoint.run_sync({"input": "data"})
+```
+
+**3. Per-Request API Key Override**
+```python
+# Override API key for a specific request
+endpoint = runpod.Endpoint("ENDPOINT_ID")  # Uses global key
+result = endpoint.run_sync({"input": "data"}, api_key="override_api_key")
+```
+
+#### API Key Precedence
+
+The SDK uses this precedence order (highest to lowest):
+1. Per-request API key (if provided to `run()` or `run_sync()`)
+2. Endpoint instance API key (if provided to `Endpoint()`)
+3. Global API key (set via `runpod.api_key`)
+
+```python
+import runpod
+
+# Example showing precedence
+runpod.api_key = "GLOBAL_KEY"
+
+# This endpoint uses GLOBAL_KEY
+endpoint1 = runpod.Endpoint("ENDPOINT_ID")
+
+# This endpoint uses ENDPOINT_KEY (overrides global)
+endpoint2 = runpod.Endpoint("ENDPOINT_ID", api_key="ENDPOINT_KEY")
+
+# This request uses REQUEST_KEY (overrides endpoint key)
+result = endpoint2.run_sync({"input": "data"}, api_key="REQUEST_KEY")
+```
+
+#### Conflict Prevention
+
+To prevent accidental API key misuse, the SDK will raise an error if you try to mix different keys:
+
+```python
+# This will raise a ValueError
+endpoint = runpod.Endpoint("ENDPOINT_ID", api_key="KEY_A")
+result = endpoint.run_sync({"input": "data"}, api_key="KEY_B")  # Error!
+
+# This is fine - same key used
+endpoint = runpod.Endpoint("ENDPOINT_ID", api_key="KEY_A")
+result = endpoint.run_sync({"input": "data"}, api_key="KEY_A")  # OK
+```
+
+#### Thread-Safe Operations
+
+Each `Endpoint` instance maintains its own API key, making concurrent operations safe:
+
+```python
+import threading
+import runpod
+
+def process_request(api_key, endpoint_id, input_data):
+    # Each thread gets its own Endpoint instance
+    endpoint = runpod.Endpoint(endpoint_id, api_key=api_key)
+    return endpoint.run_sync(input_data)
+
+# Safe concurrent usage with different API keys
+threads = []
+for customer in customers:
+    t = threading.Thread(
+        target=process_request,
+        args=(customer["api_key"], customer["endpoint_id"], customer["input"])
+    )
+    threads.append(t)
+    t.start()
+```
+
 ### GPU Cloud (Pods)
 
 ```python

runpod/api/ctl_commands.py

@@ -18,43 +18,59 @@
 from .queries import user as user_queries
 
 
-def get_user() -> dict:
+def get_user(api_key: Optional[str] = None) -> dict:
     """
-    Get the current user
+    Get the current user with optional API key override.
+    
+    Args:
+        api_key: Optional API key to use for this query.
     """
-    raw_response = run_graphql_query(user_queries.QUERY_USER)
+    raw_response = run_graphql_query(user_queries.QUERY_USER, api_key=api_key)
     cleaned_return = raw_response["data"]["myself"]
     return cleaned_return
 
 
-def update_user_settings(pubkey: str) -> dict:
+def update_user_settings(pubkey: str, api_key: Optional[str] = None) -> dict:
     """
     Update the current user
 
-    :param pubkey: the public key of the user
+    Args:
+        pubkey: the public key of the user
+        api_key: Optional API key to use for this query.
     """
-    raw_response = run_graphql_query(user_mutations.generate_user_mutation(pubkey))
+    raw_response = run_graphql_query(
+        user_mutations.generate_user_mutation(pubkey), 
+        api_key=api_key
+    )
     cleaned_return = raw_response["data"]["updateUserSettings"]
     return cleaned_return
 
 
-def get_gpus() -> dict:
+def get_gpus(api_key: Optional[str] = None) -> dict:
     """
     Get all GPU types
+    
+    Args:
+        api_key: Optional API key to use for this query.
     """
-    raw_response = run_graphql_query(gpus.QUERY_GPU_TYPES)
+    raw_response = run_graphql_query(gpus.QUERY_GPU_TYPES, api_key=api_key)
     cleaned_return = raw_response["data"]["gpuTypes"]
     return cleaned_return
 
 
-def get_gpu(gpu_id: str, gpu_quantity: int = 1):
+def get_gpu(gpu_id: str, gpu_quantity: int = 1, api_key: Optional[str] = None):
     """
     Get a specific GPU type
 
-    :param gpu_id: the id of the gpu
-    :param gpu_quantity: how many of the gpu should be returned
+    Args:
+        gpu_id: the id of the gpu
+        gpu_quantity: how many of the gpu should be returned
+        api_key: Optional API key to use for this query.
     """
-    raw_response = run_graphql_query(gpus.generate_gpu_query(gpu_id, gpu_quantity))
+    raw_response = run_graphql_query(
+        gpus.generate_gpu_query(gpu_id, gpu_quantity), 
+        api_key=api_key
+    )
 
     cleaned_return = raw_response["data"]["gpuTypes"]
 
@@ -67,22 +83,30 @@ def get_gpu(gpu_id: str, gpu_quantity: int = 1):
     return cleaned_return[0]
 
 
-def get_pods() -> dict:
+def get_pods(api_key: Optional[str] = None) -> dict:
     """
     Get all pods
+    
+    Args:
+        api_key: Optional API key to use for this query.
     """
-    raw_return = run_graphql_query(pod_queries.QUERY_POD)
+    raw_return = run_graphql_query(pod_queries.QUERY_POD, api_key=api_key)
     cleaned_return = raw_return["data"]["myself"]["pods"]
     return cleaned_return
 
 
-def get_pod(pod_id: str):
+def get_pod(pod_id: str, api_key: Optional[str] = None):
     """
     Get a specific pod
 
-    :param pod_id: the id of the pod
+    Args:
+        pod_id: the id of the pod
+        api_key: Optional API key to use for this query.
     """
-    raw_response = run_graphql_query(pod_queries.generate_pod_query(pod_id))
+    raw_response = run_graphql_query(
+        pod_queries.generate_pod_query(pod_id), 
+        api_key=api_key
+    )
     return raw_response["data"]["pod"]
 
 

runpod/api/graphql.py

@@ -4,7 +4,7 @@
 
 import json
 import os
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 import requests
 
@@ -14,19 +14,29 @@
 HTTP_STATUS_UNAUTHORIZED = 401
 
 
-def run_graphql_query(query: str) -> Dict[str, Any]:
+def run_graphql_query(query: str, api_key: Optional[str] = None) -> Dict[str, Any]:
     """
-    Run a GraphQL query
+    Run a GraphQL query with optional API key override.
+    
+    Args:
+        query: The GraphQL query to execute.
+        api_key: Optional API key to use for this query.
     """
-    from runpod import api_key  # pylint: disable=import-outside-toplevel, cyclic-import
+    from runpod import api_key as global_api_key  # pylint: disable=import-outside-toplevel, cyclic-import
+    
+    # Use provided API key or fall back to global
+    effective_api_key = api_key or global_api_key
+    
+    if not effective_api_key:
+        raise error.AuthenticationError("No API key provided")
 
     api_url_base = os.environ.get("RUNPOD_API_BASE_URL", "https://api.runpod.io")
     url = f"{api_url_base}/graphql"
 
     headers = {
         "Content-Type": "application/json",
         "User-Agent": USER_AGENT,
-        "Authorization": f"Bearer {api_key}",
+        "Authorization": f"Bearer {effective_api_key}",
     }
 
     data = json.dumps({"query": query})