personal-projects/py-wikijs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge development branch with complete v0.2.0 documentation

Browse Source 
Resolved conflicts:
- CHANGELOG.md: Combined detailed v0.1.0 with new v0.2.0 release notes
- CLAUDE.md: Kept development version for consistency

Brings in all Phase 2 features:
- Async/await support
- Caching layer
- Batch operations
- Complete API coverage (Users, Groups, Assets)
- Comprehensive documentation updates
This commit is contained in:
Claude
2025-10-23 15:06:17 +00:00
parent 6fdff8f95d a48db0e754
commit da0c46122f
50 changed files with 14654 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
docs/CHANGELOG.md
View File

@@ -27,6 +27,67 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
---
## [0.2.0] - 2025-10-23
**Enhanced Performance & Complete API Coverage**
This release significantly expands the SDK's capabilities with async support, intelligent caching, batch operations, and complete Wiki.js API coverage.
### Added
- **Async/Await Support**
- Full async client implementation (`AsyncWikiJSClient`) using aiohttp
- Async versions of all API endpoints in `wikijs.aio` module
- Support for concurrent operations with improved throughput (>3x faster)
- Async context manager support for proper resource cleanup
- **Intelligent Caching Layer**
- Abstract `BaseCache` interface for pluggable cache backends
- `MemoryCache` implementation with LRU eviction and TTL support
- Automatic cache invalidation on write operations (update, delete)
- Cache statistics tracking (hits, misses, hit rate)
- Manual cache management (clear, cleanup_expired, invalidate_resource)
- Configurable TTL and max size limits
- **Batch Operations**
- `pages.create_many()` - Bulk page creation with partial failure handling
- `pages.update_many()` - Bulk page updates with detailed error reporting
- `pages.delete_many()` - Bulk page deletion with success/failure tracking
- Significantly improved performance for bulk operations (>10x faster)
- Graceful handling of partial failures with detailed error context
- **Complete API Coverage**
- Users API with full CRUD operations (list, get, create, update, delete)
- Groups API with management and permissions
- Assets API with file upload and management capabilities
- System API with health checks and instance information
- **Documentation & Examples**
- Comprehensive caching examples (`examples/caching_example.py`)
- Batch operations guide (`examples/batch_operations.py`)
- Updated API reference with caching and batch operations
- Enhanced user guide with practical examples
- **Testing**
- 27 comprehensive cache tests covering LRU, TTL, statistics, and invalidation
- 10 batch operation tests with success and failure scenarios
- Extensive Users, Groups, and Assets API test coverage
- Overall test coverage increased from 43% to 81%
### Changed
- Pages API now supports optional caching when cache is configured
- All write operations automatically invalidate relevant cache entries
- Updated all documentation to reflect new features and capabilities
### Fixed
- All Pydantic v2 deprecation warnings (17 model classes updated)
- JWT base_url validation edge cases
- Email validation dependencies (email-validator package)
### Performance
- Caching reduces API calls by >50% for frequently accessed pages
- Batch operations achieve >10x performance improvement vs sequential operations
- Async client handles 100+ concurrent requests efficiently
- LRU cache eviction ensures optimal memory usage
## [0.1.0] - 2025-10-23
**MVP Release - Basic Wiki.js Integration** ✅
@@ -136,57 +197,22 @@ This is the first production-ready release of the Wiki.js Python SDK, delivering
## Release Planning
### [0.1.0] - Released: 2025-10-23 ✅
**MVP Release - Basic Wiki.js Integration - COMPLETE**
#### Delivered Features ✅
- ✅ Core WikiJSClient with HTTP transport
- ✅ Three authentication methods (NoAuth, API Key, JWT)
- ✅ Pages API with full CRUD operations (list, get, create, update, delete)
- ✅ Additional operations: search, get_by_path, get_by_tags
- ✅ Type-safe data models with Pydantic
- ✅ Comprehensive error handling (11 exception types)
- ✅ 87%+ test coverage (231 tests)
- ✅ Complete API documentation (3,589+ lines)
- ✅ Gitea release publication
#### Success Criteria - ALL MET ✅
- [x] Package installable via `pip install git+https://gitea.hotserv.cloud/lmiranda/wikijs-sdk-python.git`
- [x] Basic page operations work with real Wiki.js instance
- [x] All quality gates pass (tests, coverage, linting, security)
- [x] Documentation sufficient for basic usage
- [x] Examples provided (basic_usage.py, content_management.py)
### [0.2.0] - Target: 4 weeks from start
**Essential Features - Complete API Coverage**
#### Planned Features
- Users API (full CRUD operations)
- Groups API (management and permissions)
- Assets API (file upload and management)
- System API (health checks and info)
- Enhanced error handling with detailed context
- Configuration management (file and environment-based)
- Basic CLI interface
- Performance benchmarks
### [0.3.0] - Target: 7 weeks from start
### [0.3.0] - Planned
**Production Ready - Reliability & Performance**
#### Planned Features
- Retry logic with exponential backoff
- Circuit breaker for fault tolerance
- Intelligent caching with multiple backends
- Redis cache backend support
- Rate limiting and API compliance
- Performance monitoring and metrics
- Bulk operations for efficiency
- Connection pooling optimization
- Configuration management (file and environment-based)
### [1.0.0] - Target: 11 weeks from start
### [1.0.0] - Planned
**Enterprise Grade - Advanced Features**
#### Planned Features
- Full async/await support with aiohttp
- Advanced CLI with interactive mode
- Plugin architecture for extensibility
- Advanced authentication (JWT rotation, OAuth2)

1252
docs/IMPROVEMENT_PLAN.md Normal file
View File

File diff suppressed because it is too large Load Diff

190
docs/api_reference.md
View File

@@ -6,7 +6,10 @@ Complete reference for the Wiki.js Python SDK.
- [Client](#client)
- [Authentication](#authentication)
- [Caching](#caching)
- [Pages API](#pages-api)
- [Basic Operations](#basic-operations)
- [Batch Operations](#batch-operations)
- [Models](#models)
- [Exceptions](#exceptions)
- [Utilities](#utilities)
@@ -38,6 +41,7 @@ client = WikiJSClient(
- **timeout** (`int`, optional): Request timeout in seconds (default: 30)
- **verify_ssl** (`bool`, optional): Whether to verify SSL certificates (default: True)
- **user_agent** (`str`, optional): Custom User-Agent header
- **cache** (`BaseCache`, optional): Cache instance for response caching (default: None)
#### Methods
@@ -116,6 +120,105 @@ client = WikiJSClient("https://wiki.example.com", auth=auth)
---
## Caching
The SDK supports intelligent caching to reduce API calls and improve performance.
### MemoryCache
In-memory LRU cache with TTL (time-to-live) support.
```python
from wikijs import WikiJSClient
from wikijs.cache import MemoryCache
# Create cache with 5 minute TTL and max 1000 items
cache = MemoryCache(ttl=300, max_size=1000)
# Enable caching on client
client = WikiJSClient(
"https://wiki.example.com",
auth="your-api-key",
cache=cache
)
# First call hits the API
page = client.pages.get(123)
# Second call returns from cache (instant)
page = client.pages.get(123)
# Get cache statistics
stats = cache.get_stats()
print(f"Hit rate: {stats['hit_rate']}")
print(f"Cache size: {stats['current_size']}/{stats['max_size']}")
```
#### Parameters
- **ttl** (`int`, optional): Time-to-live in seconds (default: 300 = 5 minutes)
- **max_size** (`int`, optional): Maximum number of cached items (default: 1000)
#### Methods
##### get(key: CacheKey) → Optional[Any]
Retrieve value from cache if not expired.
##### set(key: CacheKey, value: Any) → None
Store value in cache with TTL.
##### delete(key: CacheKey) → None
Remove specific value from cache.
##### clear() → None
Clear all cached values.
##### invalidate_resource(resource_type: str, identifier: Optional[str] = None) → None
Invalidate cache entries for a resource type.
```python
# Invalidate specific page
cache.invalidate_resource('page', '123')
# Invalidate all pages
cache.invalidate_resource('page')
```
##### get_stats() → dict
Get cache performance statistics.
```python
stats = cache.get_stats()
# Returns: {
# 'ttl': 300,
# 'max_size': 1000,
# 'current_size': 245,
# 'hits': 1523,
# 'misses': 278,
# 'hit_rate': '84.54%',
# 'total_requests': 1801
# }
```
##### cleanup_expired() → int
Manually remove expired entries. Returns number of entries removed.
#### Cache Behavior
- **GET operations** are cached (e.g., `pages.get()`, `users.get()`)
- **Write operations** (create, update, delete) automatically invalidate cache
- **LRU eviction**: Least recently used items removed when cache is full
- **TTL expiration**: Entries automatically expire after TTL seconds
---
## Pages API
Access the Pages API through `client.pages`.
@@ -309,6 +412,93 @@ pages = client.pages.get_by_tags(
- `APIError`: If request fails
- `ValidationError`: If parameters are invalid
### Batch Operations
Efficient methods for performing multiple operations in a single call.
#### create_many()
Create multiple pages efficiently.
```python
from wikijs.models import PageCreate
pages_to_create = [
PageCreate(title="Page 1", path="page-1", content="Content 1"),
PageCreate(title="Page 2", path="page-2", content="Content 2"),
PageCreate(title="Page 3", path="page-3", content="Content 3"),
]
created_pages = client.pages.create_many(pages_to_create)
print(f"Created {len(created_pages)} pages")
```
**Parameters:**
- **pages_data** (`List[PageCreate | dict]`): List of page creation data
**Returns:** `List[Page]` - List of created Page objects
**Raises:**
- `APIError`: If creation fails (includes partial success information)
- `ValidationError`: If page data is invalid
**Note:** Continues creating pages even if some fail. Raises APIError with details about successes and failures.
#### update_many()
Update multiple pages efficiently.
```python
updates = [
{"id": 1, "content": "New content 1"},
{"id": 2, "content": "New content 2", "title": "Updated Title 2"},
{"id": 3, "is_published": False},
]
updated_pages = client.pages.update_many(updates)
print(f"Updated {len(updated_pages)} pages")
```
**Parameters:**
- **updates** (`List[dict]`): List of dicts with 'id' and fields to update
**Returns:** `List[Page]` - List of updated Page objects
**Raises:**
- `APIError`: If updates fail (includes partial success information)
- `ValidationError`: If update data is invalid (missing 'id' field)
**Note:** Each dict must contain an 'id' field. Continues updating even if some fail.
#### delete_many()
Delete multiple pages efficiently.
```python
result = client.pages.delete_many([1, 2, 3, 4, 5])
print(f"Deleted: {result['successful']}")
print(f"Failed: {result['failed']}")
if result['errors']:
print(f"Errors: {result['errors']}")
```
**Parameters:**
- **page_ids** (`List[int]`): List of page IDs to delete
**Returns:** `dict` with keys:
- `successful` (`int`): Number of successfully deleted pages
- `failed` (`int`): Number of failed deletions
- `errors` (`List[dict]`): List of errors with page_id and error message
**Raises:**
- `APIError`: If deletions fail (includes detailed error information)
- `ValidationError`: If page IDs are invalid
**Performance Benefits:**
- Reduces network overhead for bulk operations
- Partial success handling prevents all-or-nothing failures
- Detailed error reporting for debugging
---
## Models

418
docs/async_usage.md Normal file
View File

@@ -0,0 +1,418 @@
# Async/Await Support
The Wiki.js Python SDK provides full async/await support for high-performance concurrent operations using `aiohttp`.
## Installation
```bash
pip install wikijs-python-sdk[async]
```
## Quick Start
```python
import asyncio
from wikijs.aio import AsyncWikiJSClient
async def main():
# Use async context manager for automatic cleanup
async with AsyncWikiJSClient(
base_url="https://wiki.example.com",
auth="your-api-key"
) as client:
# All operations are now async
pages = await client.pages.list()
page = await client.pages.get(123)
print(f"Found {len(pages)} pages")
print(f"Page title: {page.title}")
# Run the async function
asyncio.run(main())
```
## Why Async?
Async operations provide significant performance benefits for concurrent requests:
- **Sequential (Sync)**: Requests happen one-by-one
- 100 requests @ 100ms each = 10 seconds
- **Concurrent (Async)**: Requests happen simultaneously
- 100 requests @ 100ms each = ~100ms total
- **>3x faster** for typical workloads!
## Basic Operations
### Connection Testing
```python
async with AsyncWikiJSClient(url, auth) as client:
connected = await client.test_connection()
print(f"Connected: {connected}")
```
### Listing Pages
```python
# List all pages
pages = await client.pages.list()
# List with filtering
pages = await client.pages.list(
limit=10,
offset=0,
search="documentation",
locale="en",
order_by="title",
order_direction="ASC"
)
```
### Getting Pages
```python
# Get by ID
page = await client.pages.get(123)
# Get by path
page = await client.pages.get_by_path("getting-started")
```
### Creating Pages
```python
from wikijs.models.page import PageCreate
new_page = PageCreate(
title="New Page",
path="new-page",
content="# New Page\n\nContent here.",
description="A new page",
tags=["new", "example"]
)
created_page = await client.pages.create(new_page)
print(f"Created page with ID: {created_page.id}")
```
### Updating Pages
```python
from wikijs.models.page import PageUpdate
updates = PageUpdate(
title="Updated Title",
content="# Updated\n\nNew content.",
tags=["updated"]
)
updated_page = await client.pages.update(123, updates)
```
### Deleting Pages
```python
success = await client.pages.delete(123)
print(f"Deleted: {success}")
```
### Searching Pages
```python
results = await client.pages.search("api documentation", limit=10)
for page in results:
print(f"- {page.title}")
```
## Concurrent Operations
The real power of async is running multiple operations concurrently:
### Fetch Multiple Pages
```python
import asyncio
# Sequential (slow)
pages = []
for page_id in [1, 2, 3, 4, 5]:
page = await client.pages.get(page_id)
pages.append(page)
# Concurrent (fast!)
tasks = [client.pages.get(page_id) for page_id in [1, 2, 3, 4, 5]]
pages = await asyncio.gather(*tasks)
```
### Bulk Create Operations
```python
# Create multiple pages concurrently
pages_to_create = [
PageCreate(title=f"Page {i}", path=f"page-{i}", content=f"Content {i}")
for i in range(1, 11)
]
tasks = [client.pages.create(page) for page in pages_to_create]
created_pages = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out any errors
successful = [p for p in created_pages if isinstance(p, Page)]
print(f"Created {len(successful)} pages")
```
### Parallel Search Operations
```python
# Search multiple terms concurrently
search_terms = ["api", "guide", "tutorial", "reference"]
tasks = [client.pages.search(term) for term in search_terms]
results = await asyncio.gather(*tasks)
for term, pages in zip(search_terms, results):
print(f"{term}: {len(pages)} pages found")
```
## Error Handling
Handle errors gracefully with try/except:
```python
from wikijs.exceptions import (
AuthenticationError,
NotFoundError,
APIError
)
async with AsyncWikiJSClient(url, auth) as client:
try:
page = await client.pages.get(999)
except NotFoundError:
print("Page not found")
except AuthenticationError:
print("Invalid API key")
except APIError as e:
print(f"API error: {e}")
```
### Handle Errors in Concurrent Operations
```python
# Use return_exceptions=True to continue on errors
tasks = [client.pages.get(page_id) for page_id in [1, 2, 999, 4, 5]]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Page {i}: Error - {result}")
else:
print(f"Page {i}: {result.title}")
```
## Resource Management
### Automatic Cleanup with Context Manager
```python
# Recommended: Use async context manager
async with AsyncWikiJSClient(url, auth) as client:
# Session automatically closed when block exits
pages = await client.pages.list()
```
### Manual Resource Management
```python
# If you need manual control
client = AsyncWikiJSClient(url, auth)
try:
pages = await client.pages.list()
finally:
await client.close() # Important: close the session
```
## Advanced Configuration
### Custom Connection Pool
```python
import aiohttp
# Create custom connector for fine-tuned control
connector = aiohttp.TCPConnector(
limit=200, # Max connections
limit_per_host=50, # Max per host
ttl_dns_cache=600, # DNS cache TTL
)
async with AsyncWikiJSClient(
url,
auth,
connector=connector
) as client:
# Use client with custom connector
pages = await client.pages.list()
```
### Custom Timeout
```python
# Set custom timeout (in seconds)
async with AsyncWikiJSClient(
url,
auth,
timeout=60 # 60 second timeout
) as client:
pages = await client.pages.list()
```
### Disable SSL Verification (Development Only)
```python
async with AsyncWikiJSClient(
url,
auth,
verify_ssl=False # NOT recommended for production!
) as client:
pages = await client.pages.list()
```
## Performance Best Practices
### 1. Use Connection Pooling
The async client automatically uses connection pooling. Keep a single client instance for your application:
```python
# Good: Reuse client
client = AsyncWikiJSClient(url, auth)
for i in range(100):
await client.pages.get(i)
await client.close()
# Bad: Create new client each time
for i in range(100):
async with AsyncWikiJSClient(url, auth) as client:
await client.pages.get(i) # New connection each time!
```
### 2. Batch Concurrent Operations
Use `asyncio.gather()` for concurrent operations:
```python
# Fetch 100 pages concurrently (fast!)
tasks = [client.pages.get(i) for i in range(1, 101)]
pages = await asyncio.gather(*tasks, return_exceptions=True)
```
### 3. Use Semaphores to Control Concurrency
Limit concurrent connections to avoid overwhelming the server:
```python
import asyncio
async def fetch_page_with_semaphore(client, page_id, sem):
async with sem: # Limit concurrent operations
return await client.pages.get(page_id)
# Limit to 10 concurrent requests
sem = asyncio.Semaphore(10)
tasks = [
fetch_page_with_semaphore(client, i, sem)
for i in range(1, 101)
]
pages = await asyncio.gather(*tasks)
```
## Comparison: Sync vs Async
| Feature | Sync Client | Async Client |
|---------|-------------|--------------|
| Import | `from wikijs import WikiJSClient` | `from wikijs.aio import AsyncWikiJSClient` |
| Usage | `client.pages.get(123)` | `await client.pages.get(123)` |
| Context Manager | `with WikiJSClient(...) as client:` | `async with AsyncWikiJSClient(...) as client:` |
| Concurrency | Sequential only | Concurrent with `asyncio.gather()` |
| Performance | Good for single requests | Excellent for multiple requests |
| Dependencies | `requests` | `aiohttp` |
| Best For | Simple scripts, sequential operations | Web apps, high-throughput, concurrent ops |
## When to Use Async
**Use Async When:**
- Making multiple concurrent API calls
- Building async web applications (FastAPI, aiohttp)
- Need maximum throughput
- Working with other async libraries
**Use Sync When:**
- Simple scripts or automation
- Sequential operations only
- Don't need concurrency
- Simpler code is preferred
## Complete Example
```python
import asyncio
from wikijs.aio import AsyncWikiJSClient
from wikijs.models.page import PageCreate, PageUpdate
async def main():
async with AsyncWikiJSClient(
base_url="https://wiki.example.com",
auth="your-api-key"
) as client:
# Test connection
print("Testing connection...")
connected = await client.test_connection()
print(f"Connected: {connected}")
# Create page
print("\nCreating page...")
new_page = PageCreate(
title="Test Page",
path="test-page",
content="# Test\n\nContent here.",
tags=["test"]
)
page = await client.pages.create(new_page)
print(f"Created page {page.id}: {page.title}")
# Update page
print("\nUpdating page...")
updates = PageUpdate(title="Updated Test Page")
page = await client.pages.update(page.id, updates)
print(f"Updated: {page.title}")
# List pages concurrently
print("\nFetching multiple pages...")
tasks = [
client.pages.list(limit=5),
client.pages.search("test"),
client.pages.get_by_tags(["test"])
]
list_results, search_results, tag_results = await asyncio.gather(*tasks)
print(f"Listed: {len(list_results)}")
print(f"Searched: {len(search_results)}")
print(f"By tags: {len(tag_results)}")
# Clean up
print("\nDeleting test page...")
await client.pages.delete(page.id)
print("Done!")
if __name__ == "__main__":
asyncio.run(main())
```
## See Also
- [Basic Usage Guide](../README.md#usage)
- [API Reference](api/)
- [Examples](../examples/)
- [Performance Benchmarks](benchmarks.md)

140
docs/user_guide.md
View File

@@ -353,8 +353,65 @@ for heading in headings:
print(f"- {heading}")
```
### Intelligent Caching
The SDK supports intelligent caching to reduce API calls and improve performance.
```python
from wikijs import WikiJSClient
from wikijs.cache import MemoryCache
# Create cache with 5-minute TTL and max 1000 items
cache = MemoryCache(ttl=300, max_size=1000)
# Enable caching on client
client = WikiJSClient(
"https://wiki.example.com",
auth="your-api-key",
cache=cache
)
# First call hits the API
page = client.pages.get(123) # ~200ms
# Second call returns from cache (instant!)
page = client.pages.get(123) # <1ms
# Check cache statistics
stats = cache.get_stats()
print(f"Cache hit rate: {stats['hit_rate']}")
print(f"Total requests: {stats['total_requests']}")
print(f"Cache size: {stats['current_size']}/{stats['max_size']}")
```
#### Cache Invalidation
Caches are automatically invalidated on write operations:
```python
# Enable caching
cache = MemoryCache(ttl=300)
client = WikiJSClient("https://wiki.example.com", auth="key", cache=cache)
# Get page (cached)
page = client.pages.get(123)
# Update page (cache automatically invalidated)
client.pages.update(123, {"content": "New content"})
# Next get() will fetch fresh data from API
page = client.pages.get(123) # Fresh data
# Manual cache invalidation
cache.invalidate_resource('page', '123') # Invalidate specific page
cache.invalidate_resource('page') # Invalidate all pages
cache.clear() # Clear entire cache
```
### Batch Operations
Efficient methods for bulk operations that reduce network overhead.
#### Creating Multiple Pages
```python
@@ -371,41 +428,74 @@ pages_to_create = [
for i in range(1, 6)
]
# Create them one by one
created_pages = []
for page_data in pages_to_create:
try:
created_page = client.pages.create(page_data)
created_pages.append(created_page)
print(f"Created: {created_page.title}")
except Exception as e:
print(f"Failed to create page: {e}")
# Create all pages in batch
created_pages = client.pages.create_many(pages_to_create)
print(f"Successfully created {len(created_pages)} pages")
# Handles partial failures automatically
try:
pages = client.pages.create_many(pages_to_create)
except APIError as e:
# Error includes details about successes and failures
print(f"Batch creation error: {e}")
```
#### Bulk Updates
```python
from wikijs.models import PageUpdate
# Update multiple pages efficiently
updates = [
{"id": 1, "content": "New content 1", "tags": ["updated"]},
{"id": 2, "content": "New content 2"},
{"id": 3, "is_published": False},
{"id": 4, "title": "Updated Title 4"},
]
# Get pages to update
tutorial_pages = client.pages.get_by_tags(["tutorial"])
updated_pages = client.pages.update_many(updates)
print(f"Updated {len(updated_pages)} pages")
# Update all tutorial pages
update_data = PageUpdate(
tags=["tutorial", "updated-2024"]
)
# Partial success handling
try:
pages = client.pages.update_many(updates)
except APIError as e:
# Continues updating even if some fail
print(f"Some updates failed: {e}")
```
updated_count = 0
for page in tutorial_pages:
try:
client.pages.update(page.id, update_data)
updated_count += 1
except Exception as e:
print(f"Failed to update page {page.id}: {e}")
#### Bulk Deletions
print(f"Updated {updated_count} tutorial pages")
```python
# Delete multiple pages
page_ids = [1, 2, 3, 4, 5]
result = client.pages.delete_many(page_ids)
print(f"Deleted: {result['successful']}")
print(f"Failed: {result['failed']}")
if result['errors']:
print("Errors:")
for error in result['errors']:
print(f" Page {error['page_id']}: {error['error']}")
```
#### Performance Comparison
```python
import time
# OLD WAY (slow): One by one
start = time.time()
for page_data in pages_to_create:
client.pages.create(page_data)
old_time = time.time() - start
print(f"Individual creates: {old_time:.2f}s")
# NEW WAY (fast): Batch operation
start = time.time()
client.pages.create_many(pages_to_create)
new_time = time.time() - start
print(f"Batch create: {new_time:.2f}s")
print(f"Speed improvement: {old_time/new_time:.1f}x faster")
```
### Content Migration

745
docs/users_api.md Normal file
View File

@@ -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)