Merge pull request #4 from replicatedhq/example

add example

Author: marccampbell

examples/README.md

@@ -0,0 +1,49 @@
+# Replicated Python SDK Examples
+
+This directory contains examples demonstrating how to use the Replicated Python SDK.
+
+## Basic Example
+
+The `basic_example.py` script demonstrates the most fundamental usage of the SDK:
+- Initializing the Replicated client
+- Creating or retrieving a customer
+- Creating or retrieving an instance
+
+### Usage
+
+```bash
+python3 basic_example.py \
+    --publishable-key "your_publishable_key_here" \
+    --app-slug "your-app-slug" \
+    --customer-email "[email protected]" \
+    --customer-name "John Doe" \
+    --channel "stable"
+```
+
+### Options
+
+- `--publishable-key`: Your Replicated publishable key (required)
+- `--app-slug`: Your application slug (required) 
+- `--customer-email`: Customer email address (default: [email protected])
+- `--channel`: Channel for the customer (optional)
+- `--customer-name`: Customer name (optional)
+- `--base-url`: Base URL for the Replicated API (default: https://replicated.app)
+
+### Using Different Environments
+
+To use the SDK against a different environment (not production), use the `--base-url` flag:
+
+```bash
+python3 basic_example.py \
+    --publishable-key "your_publishable_key_here" \
+    --app-slug "your-app-slug" \
+    --base-url "https://staging.replicated.app" \
+    --customer-email "[email protected]" \
+    --customer-name "Jane Smith" \
+    --channel "beta"
+```
+
+## Other Examples
+
+- `sync_example.py`: More comprehensive synchronous example with metrics and status updates
+- `async_example.py`: Asynchronous version of the SDK usage

examples/basic_example.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""
+Basic example of using the Replicated Python SDK.
+This script initializes the replicated package, creates a customer and instance.
+"""
+
+import argparse
+
+from replicated import ReplicatedClient
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Basic Replicated SDK example")
+    parser.add_argument(
+        "--base-url",
+        default="https://replicated.app",
+        help="Base URL for the Replicated API (default: https://replicated.app)",
+    )
+    parser.add_argument(
+        "--publishable-key",
+        required=True,
+        help="Your Replicated publishable key (required)",
+    )
+    parser.add_argument(
+        "--app-slug", required=True, help="Your application slug (required)"
+    )
+    parser.add_argument(
+        "--customer-email",
+        default="[email protected]",
+        help="Customer email address (default: [email protected])",
+    )
+    parser.add_argument("--channel", help="Channel for the customer (optional)")
+    parser.add_argument("--customer-name", help="Customer name (optional)")
+
+    args = parser.parse_args()
+
+    print("Initializing Replicated client...")
+    print(f"Base URL: {args.base_url}")
+    print(f"App Slug: {args.app_slug}")
+
+    # Initialize the client
+    with ReplicatedClient(
+        publishable_key=args.publishable_key,
+        app_slug=args.app_slug,
+        base_url=args.base_url,
+    ) as client:
+        print("✓ Replicated client initialized successfully")
+
+        # Create or get customer
+        channel_info = f" (channel: {args.channel})" if args.channel else ""
+        name_info = f" (name: {args.customer_name})" if args.customer_name else ""
+        print(
+            f"\nCreating/getting customer with email: "
+            f"{args.customer_email}{channel_info}{name_info}"
+        )
+        customer = client.customer.get_or_create(
+            email_address=args.customer_email,
+            channel=args.channel,
+            name=args.customer_name,
+        )
+        print(f"✓ Customer created/retrieved - ID: {customer.customer_id}")
+
+        # Create or get instance
+        print("\nCreating/getting instance for customer...")
+        instance = customer.get_or_create_instance()
+        print(f"✓ Instance created/retrieved - ID: {instance.instance_id}")
+
+        print("\n🎉 Basic example completed successfully!")
+        print(f"Customer ID: {customer.customer_id}")
+        print(f"Instance ID: {instance.instance_id}")
+
+
+if __name__ == "__main__":
+    main()

replicated/http_client.py

@@ -51,6 +51,11 @@ def _handle_response(self, response: httpx.Response) -> Dict[str, Any]:
         error_message = json_body.get("message", default_msg)
         error_code = json_body.get("code")
 
+        # Debug: print the full error response
+        print(f"DEBUG: HTTP {response.status_code} Error Response:")
+        print(f"DEBUG: Response body: {response.text}")
+        print(f"DEBUG: JSON body: {json_body}")
+
         if response.status_code == 401:
             raise ReplicatedAuthError(
                 message=error_message,

replicated/services.py

@@ -17,37 +17,57 @@ def get_or_create(
         self,
         email_address: str,
         channel: Optional[str] = None,
+        name: Optional[str] = None,
     ) -> Customer:
         """Get or create a customer."""
-        # Check if customer ID is cached
+        # Check if customer ID is cached and email matches
         cached_customer_id = self._client.state_manager.get_customer_id()
-        if cached_customer_id:
+        cached_email = self._client.state_manager.get_customer_email()
+
+        if cached_customer_id and cached_email == email_address:
+            print(
+                f"DEBUG: Using cached customer ID {cached_customer_id} "
+                f"for email {email_address}"
+            )
             return Customer(
                 self._client,
                 cached_customer_id,
                 email_address,
                 channel,
             )
+        elif cached_customer_id and cached_email != email_address:
+            print(
+                f"DEBUG: Email changed from {cached_email} to "
+                f"{email_address}, clearing cache"
+            )
+            self._client.state_manager.clear_state()
 
         # Create or fetch customer
         response = self._client.http_client._make_request(
             "POST",
-            "/api/v1/customers",
+            "/v3/customer",
             json_data={
                 "email_address": email_address,
                 "channel": channel,
+                "name": name,
                 "app_slug": self._client.app_slug,
             },
             headers=self._client._get_auth_headers(),
         )
 
-        customer_id = response["id"]
+        print(f"DEBUG: API Response: {response}")
+        customer_id = response["customer"]["id"]
         self._client.state_manager.set_customer_id(customer_id)
+        self._client.state_manager.set_customer_email(email_address)
 
         # Store dynamic token if provided
         if "dynamic_token" in response:
             dynamic_token = response["dynamic_token"]
             self._client.state_manager.set_dynamic_token(dynamic_token)
+        elif "customer" in response and "serviceToken" in response["customer"]:
+            service_token = response["customer"]["serviceToken"]
+            self._client.state_manager.set_dynamic_token(service_token)
+            print(f"DEBUG: Stored service token: {service_token[:20]}...")
 
         response_data = response.copy()
         response_data.pop("email_address", None)
@@ -71,37 +91,57 @@ async def get_or_create(
         self,
         email_address: str,
         channel: Optional[str] = None,
+        name: Optional[str] = None,
     ) -> AsyncCustomer:
         """Get or create a customer."""
-        # Check if customer ID is cached
+        # Check if customer ID is cached and email matches
         cached_customer_id = self._client.state_manager.get_customer_id()
-        if cached_customer_id:
+        cached_email = self._client.state_manager.get_customer_email()
+
+        if cached_customer_id and cached_email == email_address:
+            print(
+                f"DEBUG: Using cached customer ID {cached_customer_id} "
+                f"for email {email_address}"
+            )
             return AsyncCustomer(
                 self._client,
                 cached_customer_id,
                 email_address,
                 channel,
             )
+        elif cached_customer_id and cached_email != email_address:
+            print(
+                f"DEBUG: Email changed from {cached_email} to "
+                f"{email_address}, clearing cache"
+            )
+            self._client.state_manager.clear_state()
 
         # Create or fetch customer
         response = await self._client.http_client._make_request_async(
             "POST",
-            "/api/v1/customers",
+            "/v3/customer",
             json_data={
                 "email_address": email_address,
                 "channel": channel,
+                "name": name,
                 "app_slug": self._client.app_slug,
             },
             headers=self._client._get_auth_headers(),
         )
 
-        customer_id = response["id"]
+        print(f"DEBUG: API Response: {response}")
+        customer_id = response["customer"]["id"]
         self._client.state_manager.set_customer_id(customer_id)
+        self._client.state_manager.set_customer_email(email_address)
 
         # Store dynamic token if provided
         if "dynamic_token" in response:
             dynamic_token = response["dynamic_token"]
             self._client.state_manager.set_dynamic_token(dynamic_token)
+        elif "customer" in response and "serviceToken" in response["customer"]:
+            service_token = response["customer"]["serviceToken"]
+            self._client.state_manager.set_dynamic_token(service_token)
+            print(f"DEBUG: Stored service token: {service_token[:20]}...")
 
         response_data = response.copy()
         response_data.pop("email_address", None)

replicated/state.py

@@ -92,6 +92,17 @@ def set_dynamic_token(self, token: str) -> None:
         state["dynamic_token"] = token
         self.save_state(state)
 
+    def get_customer_email(self) -> Optional[str]:
+        """Get the cached customer email."""
+        state = self.get_state()
+        return state.get("customer_email")
+
+    def set_customer_email(self, email: str) -> None:
+        """Set the customer email in state."""
+        state = self.get_state()
+        state["customer_email"] = email
+        self.save_state(state)
+
     def clear_state(self) -> None:
         """Clear all cached state."""
         if self._state_file.exists():

tests/test_client.py

@@ -23,8 +23,11 @@ def test_customer_creation(self, mock_httpx):
         mock_response = Mock()
         mock_response.is_success = True
         mock_response.json.return_value = {
-            "id": "customer_123",
-            "email_address": "[email protected]",
+            "customer": {
+                "id": "customer_123",
+                "email": "[email protected]",
+                "name": "test user",
+            }
         }
 
         mock_client = Mock()