From 5ad98e469e561131d2bf09728e7668cf970a6d10 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Oct 2025 20:21:45 +0000 Subject: [PATCH] Add comprehensive Users API documentation and examples MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Documentation: - Complete Users API guide (docs/users_api.md) - User models overview with validation rules - Sync and async usage examples - CRUD operations guide - Advanced patterns (pagination, bulk ops, concurrent) - Error handling best practices - Complete API reference Examples: - Basic sync operations (examples/users_basic.py) - List, search, CRUD operations - Group management - Bulk operations - Pagination patterns - Error handling demonstrations - Async operations (examples/users_async.py) - Concurrent user fetching - Bulk user creation/updates - Performance comparisons (sync vs async) - Batch updates with progress tracking - Advanced error handling patterns Both examples are production-ready with comprehensive error handling and demonstrate real-world usage patterns. šŸ¤– Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- docs/users_api.md | 745 ++++++++++++++++++++++++++++++++++++++++ examples/users_async.py | 398 +++++++++++++++++++++ examples/users_basic.py | 301 ++++++++++++++++ 3 files changed, 1444 insertions(+) create mode 100644 docs/users_api.md create mode 100644 examples/users_async.py create mode 100644 examples/users_basic.py diff --git a/docs/users_api.md b/docs/users_api.md new file mode 100644 index 0000000..b57dd12 --- /dev/null +++ b/docs/users_api.md @@ -0,0 +1,745 @@ +# Users API Guide + +Comprehensive guide for managing Wiki.js users through the SDK. + +## Table of Contents + +- [Overview](#overview) +- [User Models](#user-models) +- [Basic Operations](#basic-operations) +- [Async Operations](#async-operations) +- [Advanced Usage](#advanced-usage) +- [Error Handling](#error-handling) +- [Best Practices](#best-practices) + +## Overview + +The Users API provides complete user management capabilities for Wiki.js, including: + +- **CRUD Operations**: Create, read, update, and delete users +- **User Search**: Find users by name or email +- **User Listing**: List all users with filtering and pagination +- **Group Management**: Assign users to groups +- **Profile Management**: Update user profiles and settings + +Both **synchronous** and **asynchronous** clients are supported with identical interfaces. + +## User Models + +### User + +Represents a complete Wiki.js user with all profile information. + +```python +from wikijs.models import User + +# User fields +user = User( + id=1, + name="John Doe", + email="john@example.com", + provider_key="local", # Authentication provider + is_system=False, # System user flag + is_active=True, # Account active status + is_verified=True, # Email verified + location="New York", # Optional location + job_title="Developer", # Optional job title + timezone="America/New_York", # Optional timezone + groups=[ # User's groups + {"id": 1, "name": "Administrators"}, + {"id": 2, "name": "Editors"} + ], + created_at="2024-01-01T00:00:00Z", + updated_at="2024-01-01T00:00:00Z", + last_login_at="2024-01-15T12:00:00Z" +) +``` + +### UserCreate + +Model for creating new users. + +```python +from wikijs.models import UserCreate + +# Minimal user creation +new_user = UserCreate( + email="newuser@example.com", + name="New User", + password_raw="SecurePassword123" +) + +# Complete user creation +new_user = UserCreate( + email="newuser@example.com", + name="New User", + password_raw="SecurePassword123", + provider_key="local", # Default: "local" + groups=[1, 2], # Group IDs + must_change_password=False, # Force password change on first login + send_welcome_email=True, # Send welcome email + location="San Francisco", + job_title="Software Engineer", + timezone="America/Los_Angeles" +) +``` + +**Validation Rules:** +- Email must be valid format +- Name must be 2-255 characters +- Password must be 6-255 characters +- Groups must be list of integer IDs + +### UserUpdate + +Model for updating existing users. All fields are optional. + +```python +from wikijs.models import UserUpdate + +# Partial update - only specified fields are changed +update_data = UserUpdate( + name="Jane Doe", + location="Los Angeles" +) + +# Complete update +update_data = UserUpdate( + name="Jane Doe", + email="jane@example.com", + password_raw="NewPassword123", + location="Los Angeles", + job_title="Senior Developer", + timezone="America/Los_Angeles", + groups=[1, 2, 3], # Replace all groups + is_active=True, + is_verified=True +) +``` + +**Notes:** +- Only non-None fields are sent to the API +- Partial updates are fully supported +- Password is optional (only include if changing) + +### UserGroup + +Represents a user's group membership. + +```python +from wikijs.models import UserGroup + +group = UserGroup( + id=1, + name="Administrators" +) +``` + +## Basic Operations + +### Synchronous Client + +```python +from wikijs import WikiJSClient +from wikijs.models import UserCreate, UserUpdate + +# Initialize client +client = WikiJSClient( + base_url="https://wiki.example.com", + auth="your-api-key" +) + +# List all users +users = client.users.list() +for user in users: + print(f"{user.name} ({user.email})") + +# List with filtering +users = client.users.list( + limit=10, + offset=0, + search="john", + order_by="name", + order_direction="ASC" +) + +# Get a specific user +user = client.users.get(user_id=1) +print(f"User: {user.name}") +print(f"Email: {user.email}") +print(f"Groups: {[g.name for g in user.groups]}") + +# Create a new user +new_user_data = UserCreate( + email="newuser@example.com", + name="New User", + password_raw="SecurePassword123", + groups=[1, 2] +) +created_user = client.users.create(new_user_data) +print(f"Created user: {created_user.id}") + +# Update a user +update_data = UserUpdate( + name="Updated Name", + location="New Location" +) +updated_user = client.users.update( + user_id=created_user.id, + user_data=update_data +) + +# Search for users +results = client.users.search("john", limit=5) +for user in results: + print(f"Found: {user.name} ({user.email})") + +# Delete a user +success = client.users.delete(user_id=created_user.id) +if success: + print("User deleted successfully") +``` + +## Async Operations + +### Async Client + +```python +import asyncio +from wikijs.aio import AsyncWikiJSClient +from wikijs.models import UserCreate, UserUpdate + +async def manage_users(): + # Initialize async client + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", + auth="your-api-key" + ) as client: + # List users + users = await client.users.list() + + # Get specific user + user = await client.users.get(user_id=1) + + # Create user + new_user_data = UserCreate( + email="newuser@example.com", + name="New User", + password_raw="SecurePassword123" + ) + created_user = await client.users.create(new_user_data) + + # Update user + update_data = UserUpdate(name="Updated Name") + updated_user = await client.users.update( + user_id=created_user.id, + user_data=update_data + ) + + # Search users + results = await client.users.search("john", limit=5) + + # Delete user + success = await client.users.delete(user_id=created_user.id) + +# Run async function +asyncio.run(manage_users()) +``` + +### Concurrent Operations + +Process multiple users concurrently for better performance: + +```python +import asyncio +from wikijs.aio import AsyncWikiJSClient +from wikijs.models import UserUpdate + +async def update_users_concurrently(): + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", + auth="your-api-key" + ) as client: + # Get all users + users = await client.users.list() + + # Update all users concurrently + update_data = UserUpdate(is_verified=True) + + tasks = [ + client.users.update(user.id, update_data) + for user in users + if not user.is_verified + ] + + # Execute all updates concurrently + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Process results + success_count = sum(1 for r in results if not isinstance(r, Exception)) + print(f"Updated {success_count}/{len(tasks)} users") + +asyncio.run(update_users_concurrently()) +``` + +## Advanced Usage + +### Using Dictionaries Instead of Models + +You can use dictionaries instead of model objects: + +```python +# Create user from dict +user_dict = { + "email": "user@example.com", + "name": "Test User", + "password_raw": "SecurePassword123", + "groups": [1, 2] +} +created_user = client.users.create(user_dict) + +# Update user from dict +update_dict = { + "name": "Updated Name", + "location": "New Location" +} +updated_user = client.users.update(user_id=1, user_data=update_dict) +``` + +### Pagination + +Handle large user lists with pagination: + +```python +# Fetch users in batches +def fetch_all_users(client, batch_size=50): + all_users = [] + offset = 0 + + while True: + batch = client.users.list( + limit=batch_size, + offset=offset, + order_by="id", + order_direction="ASC" + ) + + if not batch: + break + + all_users.extend(batch) + offset += batch_size + + print(f"Fetched {len(all_users)} users so far...") + + return all_users + +# Async pagination +async def fetch_all_users_async(client, batch_size=50): + all_users = [] + offset = 0 + + while True: + batch = await client.users.list( + limit=batch_size, + offset=offset, + order_by="id", + order_direction="ASC" + ) + + if not batch: + break + + all_users.extend(batch) + offset += batch_size + + return all_users +``` + +### Group Management + +Manage user group assignments: + +```python +from wikijs.models import UserUpdate + +# Add user to groups +update_data = UserUpdate(groups=[1, 2, 3]) # Group IDs +updated_user = client.users.update(user_id=1, user_data=update_data) + +# Remove user from all groups +update_data = UserUpdate(groups=[]) +updated_user = client.users.update(user_id=1, user_data=update_data) + +# Get user's current groups +user = client.users.get(user_id=1) +print("User groups:") +for group in user.groups: + print(f" - {group.name} (ID: {group.id})") +``` + +### Bulk User Creation + +Create multiple users efficiently: + +```python +from wikijs.models import UserCreate + +# Sync bulk creation +def create_users_bulk(client, user_data_list): + created_users = [] + + for user_data in user_data_list: + try: + user = client.users.create(user_data) + created_users.append(user) + print(f"Created: {user.name}") + except Exception as e: + print(f"Failed to create {user_data['name']}: {e}") + + return created_users + +# Async bulk creation (concurrent) +async def create_users_bulk_async(client, user_data_list): + tasks = [ + client.users.create(user_data) + for user_data in user_data_list + ] + + results = await asyncio.gather(*tasks, return_exceptions=True) + + created_users = [ + r for r in results if not isinstance(r, Exception) + ] + + print(f"Created {len(created_users)}/{len(user_data_list)} users") + return created_users +``` + +## Error Handling + +### Common Exceptions + +```python +from wikijs.exceptions import ( + ValidationError, + APIError, + AuthenticationError, + ConnectionError, + TimeoutError +) + +try: + # Create user with invalid data + user_data = UserCreate( + email="invalid-email", # Invalid format + name="Test", + password_raw="123" # Too short + ) +except ValidationError as e: + print(f"Validation error: {e}") + +try: + # Get non-existent user + user = client.users.get(user_id=99999) +except APIError as e: + print(f"API error: {e}") + +try: + # Invalid authentication + client = WikiJSClient( + base_url="https://wiki.example.com", + auth="invalid-key" + ) + users = client.users.list() +except AuthenticationError as e: + print(f"Authentication failed: {e}") +``` + +### Robust Error Handling + +```python +from wikijs.exceptions import ValidationError, APIError + +def create_user_safely(client, user_data): + """Create user with comprehensive error handling.""" + try: + # Validate data first + validated_data = UserCreate(**user_data) + + # Create user + user = client.users.create(validated_data) + print(f"āœ“ Created user: {user.name} (ID: {user.id})") + return user + + except ValidationError as e: + print(f"āœ— Validation error: {e}") + # Handle validation errors (e.g., fix data and retry) + return None + + except APIError as e: + if "already exists" in str(e).lower(): + print(f"āœ— User already exists: {user_data['email']}") + # Handle duplicate user + return None + else: + print(f"āœ— API error: {e}") + raise + + except Exception as e: + print(f"āœ— Unexpected error: {e}") + raise + +# Async version +async def create_user_safely_async(client, user_data): + try: + validated_data = UserCreate(**user_data) + user = await client.users.create(validated_data) + print(f"āœ“ Created user: {user.name} (ID: {user.id})") + return user + except ValidationError as e: + print(f"āœ— Validation error: {e}") + return None + except APIError as e: + if "already exists" in str(e).lower(): + print(f"āœ— User already exists: {user_data['email']}") + return None + else: + print(f"āœ— API error: {e}") + raise +``` + +## Best Practices + +### 1. Use Models for Type Safety + +Always use Pydantic models for better validation and IDE support: + +```python +# Good - type safe with validation +user_data = UserCreate( + email="user@example.com", + name="Test User", + password_raw="SecurePassword123" +) +user = client.users.create(user_data) + +# Acceptable - but less type safe +user_dict = { + "email": "user@example.com", + "name": "Test User", + "password_raw": "SecurePassword123" +} +user = client.users.create(user_dict) +``` + +### 2. Handle Pagination for Large Datasets + +Always paginate when dealing with many users: + +```python +# Good - paginated +all_users = [] +offset = 0 +batch_size = 50 + +while True: + batch = client.users.list(limit=batch_size, offset=offset) + if not batch: + break + all_users.extend(batch) + offset += batch_size + +# Bad - loads all users at once +all_users = client.users.list() # May be slow for large user bases +``` + +### 3. Use Async for Concurrent Operations + +Use async client for better performance when processing multiple users: + +```python +# Good - concurrent async operations +async with AsyncWikiJSClient(...) as client: + tasks = [client.users.get(id) for id in user_ids] + users = await asyncio.gather(*tasks) + +# Less efficient - sequential sync operations +for user_id in user_ids: + user = client.users.get(user_id) +``` + +### 4. Validate Before API Calls + +Catch validation errors early: + +```python +# Good - validate first +try: + user_data = UserCreate(**raw_data) + user = client.users.create(user_data) +except ValidationError as e: + print(f"Invalid data: {e}") + # Fix data before API call + +# Less efficient - validation happens during API call +user = client.users.create(raw_data) +``` + +### 5. Use Partial Updates + +Only update fields that changed: + +```python +# Good - only update changed fields +update_data = UserUpdate(name="New Name") +user = client.users.update(user_id=1, user_data=update_data) + +# Wasteful - updates all fields +update_data = UserUpdate( + name="New Name", + email=user.email, + location=user.location, + # ... all other fields +) +user = client.users.update(user_id=1, user_data=update_data) +``` + +### 6. Implement Retry Logic for Production + +```python +import time +from wikijs.exceptions import ConnectionError, TimeoutError + +def create_user_with_retry(client, user_data, max_retries=3): + """Create user with automatic retry on transient failures.""" + for attempt in range(max_retries): + try: + return client.users.create(user_data) + except (ConnectionError, TimeoutError) as e: + if attempt < max_retries - 1: + wait_time = 2 ** attempt # Exponential backoff + print(f"Retry {attempt + 1}/{max_retries} after {wait_time}s...") + time.sleep(wait_time) + else: + raise +``` + +### 7. Secure Password Handling + +```python +import getpass +from wikijs.models import UserCreate + +# Good - prompt for password securely +password = getpass.getpass("Enter password: ") +user_data = UserCreate( + email="user@example.com", + name="Test User", + password_raw=password +) + +# Bad - hardcoded passwords +user_data = UserCreate( + email="user@example.com", + name="Test User", + password_raw="password123" # Never do this! +) +``` + +## Examples + +See the `examples/` directory for complete working examples: + +- `examples/users_basic.py` - Basic user management operations +- `examples/users_async.py` - Async user management with concurrency +- `examples/users_bulk_import.py` - Bulk user import from CSV + +## API Reference + +### UsersEndpoint / AsyncUsersEndpoint + +#### `list(limit=None, offset=None, search=None, order_by="name", order_direction="ASC")` + +List users with optional filtering and pagination. + +**Parameters:** +- `limit` (int, optional): Maximum number of users to return +- `offset` (int, optional): Number of users to skip +- `search` (str, optional): Search term (filters by name or email) +- `order_by` (str): Field to sort by (`name`, `email`, `createdAt`, `lastLoginAt`) +- `order_direction` (str): Sort direction (`ASC` or `DESC`) + +**Returns:** `List[User]` + +**Raises:** `ValidationError`, `APIError` + +#### `get(user_id)` + +Get a specific user by ID. + +**Parameters:** +- `user_id` (int): User ID + +**Returns:** `User` + +**Raises:** `ValidationError`, `APIError` + +#### `create(user_data)` + +Create a new user. + +**Parameters:** +- `user_data` (UserCreate or dict): User creation data + +**Returns:** `User` + +**Raises:** `ValidationError`, `APIError` + +#### `update(user_id, user_data)` + +Update an existing user. + +**Parameters:** +- `user_id` (int): User ID +- `user_data` (UserUpdate or dict): User update data + +**Returns:** `User` + +**Raises:** `ValidationError`, `APIError` + +#### `delete(user_id)` + +Delete a user. + +**Parameters:** +- `user_id` (int): User ID + +**Returns:** `bool` (True if successful) + +**Raises:** `ValidationError`, `APIError` + +#### `search(query, limit=None)` + +Search for users by name or email. + +**Parameters:** +- `query` (str): Search query +- `limit` (int, optional): Maximum number of results + +**Returns:** `List[User]` + +**Raises:** `ValidationError`, `APIError` + +## Related Documentation + +- [Async Usage Guide](async_usage.md) +- [Authentication Guide](../README.md#authentication) +- [API Reference](../README.md#api-documentation) +- [Examples](../examples/) + +## Support + +For issues and questions: +- GitHub Issues: [wikijs-python-sdk/issues](https://github.com/yourusername/wikijs-python-sdk/issues) +- Documentation: [Full Documentation](../README.md) diff --git a/examples/users_async.py b/examples/users_async.py new file mode 100644 index 0000000..62bacfe --- /dev/null +++ b/examples/users_async.py @@ -0,0 +1,398 @@ +"""Async users management example for wikijs-python-sdk. + +This example demonstrates: +- Async user operations +- Concurrent user processing +- Bulk operations with asyncio.gather +- Performance comparison with sync operations +""" + +import asyncio +import time +from typing import List + +from wikijs.aio import AsyncWikiJSClient +from wikijs.exceptions import APIError, ValidationError +from wikijs.models import User, UserCreate, UserUpdate + + +async def basic_async_operations(): + """Demonstrate basic async user operations.""" + print("=" * 60) + print("Async Users API - Basic Operations") + print("=" * 60) + + # Initialize async client with context manager + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", auth="your-api-key-here" + ) as client: + + # 1. List users + print("\n1. Listing all users...") + try: + users = await client.users.list() + print(f" Found {len(users)} users") + for user in users[:5]: + print(f" - {user.name} ({user.email})") + except APIError as e: + print(f" Error: {e}") + + # 2. Search users + print("\n2. Searching for users...") + try: + results = await client.users.search("admin", limit=5) + print(f" Found {len(results)} matching users") + except APIError as e: + print(f" Error: {e}") + + # 3. Create user + print("\n3. Creating a new user...") + try: + new_user_data = UserCreate( + email="asynctest@example.com", + name="Async Test User", + password_raw="SecurePassword123", + location="Remote", + job_title="Engineer", + ) + + created_user = await client.users.create(new_user_data) + print(f" āœ“ Created: {created_user.name} (ID: {created_user.id})") + test_user_id = created_user.id + + except (ValidationError, APIError) as e: + print(f" āœ— Error: {e}") + return + + # 4. Get user + print(f"\n4. Getting user {test_user_id}...") + try: + user = await client.users.get(test_user_id) + print(f" User: {user.name}") + print(f" Email: {user.email}") + print(f" Location: {user.location}") + except APIError as e: + print(f" Error: {e}") + + # 5. Update user + print(f"\n5. Updating user...") + try: + update_data = UserUpdate( + name="Updated Async User", location="San Francisco" + ) + updated_user = await client.users.update(test_user_id, update_data) + print(f" āœ“ Updated: {updated_user.name}") + print(f" Location: {updated_user.location}") + except APIError as e: + print(f" Error: {e}") + + # 6. Delete user + print(f"\n6. Deleting test user...") + try: + await client.users.delete(test_user_id) + print(f" āœ“ User deleted") + except APIError as e: + print(f" Error: {e}") + + +async def concurrent_user_fetch(): + """Demonstrate concurrent user fetching for better performance.""" + print("\n" + "=" * 60) + print("Concurrent User Fetching") + print("=" * 60) + + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", auth="your-api-key-here" + ) as client: + + # Get list of user IDs + print("\n1. Getting list of users...") + users = await client.users.list(limit=10) + user_ids = [user.id for user in users] + print(f" Will fetch {len(user_ids)} users concurrently") + + # Fetch all users concurrently + print("\n2. Fetching users concurrently...") + start_time = time.time() + + tasks = [client.users.get(user_id) for user_id in user_ids] + fetched_users = await asyncio.gather(*tasks, return_exceptions=True) + + elapsed = time.time() - start_time + + # Process results + successful = [u for u in fetched_users if isinstance(u, User)] + failed = [u for u in fetched_users if isinstance(u, Exception)] + + print(f" āœ“ Fetched {len(successful)} users successfully") + print(f" āœ— Failed: {len(failed)}") + print(f" ā± Time: {elapsed:.2f}s") + print(f" šŸ“Š Average: {elapsed/len(user_ids):.3f}s per user") + + +async def bulk_user_creation(): + """Demonstrate bulk user creation with concurrent operations.""" + print("\n" + "=" * 60) + print("Bulk User Creation (Concurrent)") + print("=" * 60) + + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", auth="your-api-key-here" + ) as client: + + # Prepare user data + print("\n1. Preparing user data...") + users_to_create = [ + UserCreate( + email=f"bulkuser{i}@example.com", + name=f"Bulk User {i}", + password_raw=f"SecurePass{i}123", + location="Test Location", + job_title="Test Engineer", + ) + for i in range(1, 6) + ] + print(f" Prepared {len(users_to_create)} users") + + # Create all users concurrently + print("\n2. Creating users concurrently...") + start_time = time.time() + + tasks = [client.users.create(user_data) for user_data in users_to_create] + results = await asyncio.gather(*tasks, return_exceptions=True) + + elapsed = time.time() - start_time + + # Process results + created_users = [r for r in results if isinstance(r, User)] + failed = [r for r in results if isinstance(r, Exception)] + + print(f" āœ“ Created: {len(created_users)} users") + print(f" āœ— Failed: {len(failed)}") + print(f" ā± Time: {elapsed:.2f}s") + + # Show created users + for user in created_users: + print(f" - {user.name} (ID: {user.id})") + + # Update all users concurrently + if created_users: + print("\n3. Updating all users concurrently...") + update_data = UserUpdate(location="Updated Location", is_verified=True) + + tasks = [ + client.users.update(user.id, update_data) for user in created_users + ] + updated_users = await asyncio.gather(*tasks, return_exceptions=True) + + successful_updates = [u for u in updated_users if isinstance(u, User)] + print(f" āœ“ Updated: {len(successful_updates)} users") + + # Delete all test users + if created_users: + print("\n4. Cleaning up (deleting test users)...") + tasks = [client.users.delete(user.id) for user in created_users] + results = await asyncio.gather(*tasks, return_exceptions=True) + + successful_deletes = [r for r in results if r is True] + print(f" āœ“ Deleted: {len(successful_deletes)} users") + + +async def performance_comparison(): + """Compare sync vs async performance.""" + print("\n" + "=" * 60) + print("Performance Comparison: Sync vs Async") + print("=" * 60) + + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", auth="your-api-key-here" + ) as async_client: + + # Get list of user IDs + users = await async_client.users.list(limit=20) + user_ids = [user.id for user in users[:10]] # Use first 10 + + print(f"\nFetching {len(user_ids)} users...") + + # Async concurrent fetching + print("\n1. Async (concurrent):") + start_time = time.time() + + tasks = [async_client.users.get(user_id) for user_id in user_ids] + async_results = await asyncio.gather(*tasks, return_exceptions=True) + + async_time = time.time() - start_time + + async_successful = len([r for r in async_results if isinstance(r, User)]) + print(f" Fetched: {async_successful} users") + print(f" Time: {async_time:.2f}s") + print(f" Rate: {len(user_ids)/async_time:.1f} users/sec") + + # Async sequential fetching (for comparison) + print("\n2. Async (sequential):") + start_time = time.time() + + sequential_results = [] + for user_id in user_ids: + try: + user = await async_client.users.get(user_id) + sequential_results.append(user) + except Exception as e: + sequential_results.append(e) + + sequential_time = time.time() - start_time + + seq_successful = len([r for r in sequential_results if isinstance(r, User)]) + print(f" Fetched: {seq_successful} users") + print(f" Time: {sequential_time:.2f}s") + print(f" Rate: {len(user_ids)/sequential_time:.1f} users/sec") + + # Calculate speedup + speedup = sequential_time / async_time + print(f"\nšŸ“Š Performance Summary:") + print(f" Concurrent speedup: {speedup:.1f}x faster") + print(f" Time saved: {sequential_time - async_time:.2f}s") + + +async def batch_user_updates(): + """Demonstrate batch updates with progress tracking.""" + print("\n" + "=" * 60) + print("Batch User Updates with Progress Tracking") + print("=" * 60) + + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", auth="your-api-key-here" + ) as client: + + # Get users to update + print("\n1. Finding users to update...") + users = await client.users.list(limit=10) + print(f" Found {len(users)} users") + + # Update all users concurrently with progress + print("\n2. Updating users...") + update_data = UserUpdate(is_verified=True) + + async def update_with_progress(user: User, index: int, total: int): + """Update user and show progress.""" + try: + updated = await client.users.update(user.id, update_data) + print(f" [{index}/{total}] āœ“ Updated: {updated.name}") + return updated + except Exception as e: + print(f" [{index}/{total}] āœ— Failed: {user.name} - {e}") + return e + + tasks = [ + update_with_progress(user, i + 1, len(users)) + for i, user in enumerate(users) + ] + + results = await asyncio.gather(*tasks) + + # Summary + successful = len([r for r in results if isinstance(r, User)]) + failed = len([r for r in results if isinstance(r, Exception)]) + + print(f"\n Summary:") + print(f" āœ“ Successful: {successful}") + print(f" āœ— Failed: {failed}") + + +async def advanced_error_handling(): + """Demonstrate advanced error handling patterns.""" + print("\n" + "=" * 60) + print("Advanced Error Handling") + print("=" * 60) + + async with AsyncWikiJSClient( + base_url="https://wiki.example.com", auth="your-api-key-here" + ) as client: + + print("\n1. Individual error handling:") + + # Try to create multiple users with mixed valid/invalid data + test_users = [ + { + "email": "valid1@example.com", + "name": "Valid User 1", + "password_raw": "SecurePass123", + }, + { + "email": "invalid-email", + "name": "Invalid Email", + "password_raw": "SecurePass123", + }, + { + "email": "valid2@example.com", + "name": "Valid User 2", + "password_raw": "123", + }, # Weak password + { + "email": "valid3@example.com", + "name": "Valid User 3", + "password_raw": "SecurePass123", + }, + ] + + async def create_user_safe(user_data: dict): + """Create user with error handling.""" + try: + validated_data = UserCreate(**user_data) + user = await client.users.create(validated_data) + print(f" āœ“ Created: {user.name}") + return user + except ValidationError as e: + print(f" āœ— Validation error for {user_data.get('email')}: {e}") + return None + except APIError as e: + print(f" āœ— API error for {user_data.get('email')}: {e}") + return None + + results = await asyncio.gather(*[create_user_safe(u) for u in test_users]) + + # Clean up created users + created = [r for r in results if r is not None] + if created: + print(f"\n2. Cleaning up {len(created)} created users...") + await asyncio.gather(*[client.users.delete(u.id) for u in created]) + print(" āœ“ Cleanup complete") + + +async def main(): + """Run all async examples.""" + try: + # Basic operations + await basic_async_operations() + + # Concurrent operations + await concurrent_user_fetch() + + # Bulk operations + await bulk_user_creation() + + # Performance comparison + await performance_comparison() + + # Batch updates + await batch_user_updates() + + # Error handling + await advanced_error_handling() + + print("\n" + "=" * 60) + print("All examples completed!") + print("=" * 60) + + except KeyboardInterrupt: + print("\n\nInterrupted by user") + except Exception as e: + print(f"\n\nUnexpected error: {e}") + import traceback + + traceback.print_exc() + + +if __name__ == "__main__": + # Run all examples + asyncio.run(main()) diff --git a/examples/users_basic.py b/examples/users_basic.py new file mode 100644 index 0000000..7c498aa --- /dev/null +++ b/examples/users_basic.py @@ -0,0 +1,301 @@ +"""Basic users management example for wikijs-python-sdk. + +This example demonstrates: +- Creating users +- Reading user information +- Updating users +- Deleting users +- Searching users +- Managing user groups +""" + +from wikijs import WikiJSClient +from wikijs.exceptions import APIError, ValidationError +from wikijs.models import UserCreate, UserUpdate + + +def main(): + """Run basic user management operations.""" + # Initialize client + client = WikiJSClient( + base_url="https://wiki.example.com", + auth="your-api-key-here", # Replace with your actual API key + ) + + print("=" * 60) + print("Wiki.js Users API - Basic Operations Example") + print("=" * 60) + + # 1. List all users + print("\n1. Listing all users...") + try: + users = client.users.list() + print(f" Found {len(users)} users") + for user in users[:5]: # Show first 5 + print(f" - {user.name} ({user.email}) - Active: {user.is_active}") + except APIError as e: + print(f" Error listing users: {e}") + + # 2. List users with filtering + print("\n2. Listing users with pagination and ordering...") + try: + users = client.users.list( + limit=10, offset=0, order_by="email", order_direction="ASC" + ) + print(f" Found {len(users)} users (first 10)") + except APIError as e: + print(f" Error: {e}") + + # 3. Search for users + print("\n3. Searching for users...") + try: + search_term = "admin" + results = client.users.search(search_term, limit=5) + print(f" Found {len(results)} users matching '{search_term}'") + for user in results: + print(f" - {user.name} ({user.email})") + except APIError as e: + print(f" Error searching: {e}") + + # 4. Create a new user + print("\n4. Creating a new user...") + try: + new_user_data = UserCreate( + email="testuser@example.com", + name="Test User", + password_raw="SecurePassword123", + groups=[1], # Assign to group with ID 1 + location="San Francisco", + job_title="QA Engineer", + timezone="America/Los_Angeles", + send_welcome_email=False, # Don't send email for test user + must_change_password=True, + ) + + created_user = client.users.create(new_user_data) + print(f" āœ“ Created user: {created_user.name}") + print(f" ID: {created_user.id}") + print(f" Email: {created_user.email}") + print(f" Active: {created_user.is_active}") + print(f" Verified: {created_user.is_verified}") + + # Save user ID for later operations + test_user_id = created_user.id + + except ValidationError as e: + print(f" āœ— Validation error: {e}") + return + except APIError as e: + print(f" āœ— API error: {e}") + if "already exists" in str(e).lower(): + print(" Note: User might already exist from previous run") + return + + # 5. Get specific user + print(f"\n5. Getting user by ID ({test_user_id})...") + try: + user = client.users.get(test_user_id) + print(f" User: {user.name}") + print(f" Email: {user.email}") + print(f" Location: {user.location}") + print(f" Job Title: {user.job_title}") + print(f" Groups: {[g.name for g in user.groups]}") + except APIError as e: + print(f" Error: {e}") + + # 6. Update user information + print(f"\n6. Updating user...") + try: + update_data = UserUpdate( + name="Updated Test User", + location="New York", + job_title="Senior QA Engineer", + is_verified=True, + ) + + updated_user = client.users.update(test_user_id, update_data) + print(f" āœ“ Updated user: {updated_user.name}") + print(f" New location: {updated_user.location}") + print(f" New job title: {updated_user.job_title}") + print(f" Verified: {updated_user.is_verified}") + except APIError as e: + print(f" Error: {e}") + + # 7. Update user password + print(f"\n7. Updating user password...") + try: + password_update = UserUpdate(password_raw="NewSecurePassword456") + + updated_user = client.users.update(test_user_id, password_update) + print(f" āœ“ Password updated for user: {updated_user.name}") + except APIError as e: + print(f" Error: {e}") + + # 8. Manage user groups + print(f"\n8. Managing user groups...") + try: + # Add user to multiple groups + group_update = UserUpdate(groups=[1, 2, 3]) + updated_user = client.users.update(test_user_id, group_update) + print(f" āœ“ User groups updated") + print(f" Groups: {[g.name for g in updated_user.groups]}") + except APIError as e: + print(f" Error: {e}") + + # 9. Deactivate user + print(f"\n9. Deactivating user...") + try: + deactivate_update = UserUpdate(is_active=False) + updated_user = client.users.update(test_user_id, deactivate_update) + print(f" āœ“ User deactivated: {updated_user.name}") + print(f" Active: {updated_user.is_active}") + except APIError as e: + print(f" Error: {e}") + + # 10. Reactivate user + print(f"\n10. Reactivating user...") + try: + reactivate_update = UserUpdate(is_active=True) + updated_user = client.users.update(test_user_id, reactivate_update) + print(f" āœ“ User reactivated: {updated_user.name}") + print(f" Active: {updated_user.is_active}") + except APIError as e: + print(f" Error: {e}") + + # 11. Delete user + print(f"\n11. Deleting test user...") + try: + success = client.users.delete(test_user_id) + if success: + print(f" āœ“ User deleted successfully") + except APIError as e: + print(f" Error: {e}") + if "system user" in str(e).lower(): + print(" Note: Cannot delete system users") + + # 12. Demonstrate error handling + print("\n12. Demonstrating error handling...") + + # Try to create user with invalid email + print(" a) Invalid email validation:") + try: + invalid_user = UserCreate( + email="not-an-email", name="Test", password_raw="password123" + ) + client.users.create(invalid_user) + except ValidationError as e: + print(f" āœ“ Caught validation error: {e}") + + # Try to create user with weak password + print(" b) Weak password validation:") + try: + weak_password_user = UserCreate( + email="test@example.com", name="Test User", password_raw="123" # Too short + ) + client.users.create(weak_password_user) + except ValidationError as e: + print(f" āœ“ Caught validation error: {e}") + + # Try to get non-existent user + print(" c) Non-existent user:") + try: + user = client.users.get(99999) + except APIError as e: + print(f" āœ“ Caught API error: {e}") + + print("\n" + "=" * 60) + print("Example completed!") + print("=" * 60) + + +def demonstrate_bulk_operations(): + """Demonstrate bulk user operations.""" + client = WikiJSClient(base_url="https://wiki.example.com", auth="your-api-key-here") + + print("\n" + "=" * 60) + print("Bulk Operations Example") + print("=" * 60) + + # Create multiple users + print("\n1. Creating multiple users...") + users_to_create = [ + { + "email": f"user{i}@example.com", + "name": f"User {i}", + "password_raw": f"SecurePass{i}123", + "job_title": "Team Member", + } + for i in range(1, 4) + ] + + created_users = [] + for user_data in users_to_create: + try: + user = client.users.create(UserCreate(**user_data)) + created_users.append(user) + print(f" āœ“ Created: {user.name}") + except (ValidationError, APIError) as e: + print(f" āœ— Failed to create {user_data['name']}: {e}") + + # Update all created users + print("\n2. Updating all created users...") + update_data = UserUpdate(location="Team Location", is_verified=True) + + for user in created_users: + try: + updated_user = client.users.update(user.id, update_data) + print(f" āœ“ Updated: {updated_user.name}") + except APIError as e: + print(f" āœ— Failed to update {user.name}: {e}") + + # Delete all created users + print("\n3. Cleaning up (deleting test users)...") + for user in created_users: + try: + client.users.delete(user.id) + print(f" āœ“ Deleted: {user.name}") + except APIError as e: + print(f" āœ— Failed to delete {user.name}: {e}") + + +def demonstrate_pagination(): + """Demonstrate pagination for large user lists.""" + client = WikiJSClient(base_url="https://wiki.example.com", auth="your-api-key-here") + + print("\n" + "=" * 60) + print("Pagination Example") + print("=" * 60) + + # Fetch all users in batches + print("\nFetching all users in batches of 50...") + all_users = [] + offset = 0 + batch_size = 50 + + while True: + try: + batch = client.users.list( + limit=batch_size, offset=offset, order_by="id", order_direction="ASC" + ) + + if not batch: + break + + all_users.extend(batch) + offset += batch_size + print(f" Fetched batch: {len(batch)} users (total: {len(all_users)})") + + except APIError as e: + print(f" Error fetching batch: {e}") + break + + print(f"\nTotal users fetched: {len(all_users)}") + + +if __name__ == "__main__": + # Run main example + main() + + # Uncomment to run additional examples: + # demonstrate_bulk_operations() + # demonstrate_pagination()