930fe50e60
Phase 2, Task 2.2.1: Users API Implementation (Sync) This commit adds comprehensive user management capabilities to the Wiki.js Python SDK with full CRUD operations. User Data Models (wikijs/models/user.py): ----------------------------------------- 1. **User** - Complete user data model - Profile information (name, email, location, job title) - Authentication details (provider, verified, active status) - Group memberships with UserGroup nested model - Timestamps (created, updated, last login) - Email validation with pydantic EmailStr - Name validation (2-255 characters) 2. **UserCreate** - User creation model - Required: email, name, password - Optional: groups, profile fields, provider settings - Validation: email format, password strength (min 6 chars) - Welcome email configuration - Password change enforcement 3. **UserUpdate** - User update model - All fields optional for partial updates - Email, name, password updates - Profile field updates - Group membership changes - Status flags (active, verified) 4. **UserGroup** - Group membership model - Group ID and name - Used in User model for group associations Users Endpoint (wikijs/endpoints/users.py): ------------------------------------------- Complete CRUD Operations: 1. **list()** - List users with filtering - Pagination (limit, offset) - Search by name/email - Ordering (name, email, createdAt, lastLoginAt) - Client-side pagination fallback 2. **get(user_id)** - Get user by ID - Fetch single user with full details - Include group memberships - Comprehensive error handling 3. **create(user_data)** - Create new user - Accept UserCreate object or dict - Full validation before API call - Returns created User object - Handles creation failures gracefully 4. **update(user_id, user_data)** - Update existing user - Partial updates supported - Only sends changed fields - Returns updated User object - Validates all updates 5. **delete(user_id)** - Delete user - Permanent deletion - Returns boolean success - Clear error messages 6. **search(query)** - Search users - Search by name or email - Optional result limiting - Uses list() with search filter Helper Methods: - _normalize_user_data() - API response normalization - Handles field name mapping (camelCase → snake_case) - Group data structure conversion Integration: ------------ - Added UsersEndpoint to WikiJSClient - Updated endpoints module exports - Added user models to main package exports - Installed email-validator dependency for EmailStr GraphQL Queries: ---------------- - users.list - List/search users - users.single - Get user by ID - users.create - Create new user - users.update - Update existing user - users.delete - Delete user All queries include proper error handling and response validation. Code Quality: ------------- ✅ Compiles without errors ✅ Type hints on all methods ✅ Comprehensive docstrings ✅ Input validation ✅ Proper exception handling ✅ Follows existing code patterns Next Steps: ----------- - Implement AsyncUsersEndpoint (async version) - Write comprehensive tests - Add usage documentation - Create examples Phase 2, Task 2.2.1 Progress: ~50% Complete Users API (sync): ✅ COMPLETE Users API (async): ⏳ IN PROGRESS This establishes the foundation for complete user management in the Wiki.js Python SDK. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
571 lines
18 KiB
Python
571 lines
18 KiB
Python
"""Users API endpoint for wikijs-python-sdk."""
|
|
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
from ..exceptions import APIError, ValidationError
|
|
from ..models.user import User, UserCreate, UserUpdate
|
|
from .base import BaseEndpoint
|
|
|
|
|
|
class UsersEndpoint(BaseEndpoint):
|
|
"""Endpoint for Wiki.js Users API operations.
|
|
|
|
This endpoint provides methods for creating, reading, updating, and deleting
|
|
users through the Wiki.js GraphQL API.
|
|
|
|
Example:
|
|
>>> client = WikiJSClient('https://wiki.example.com', auth='api-key')
|
|
>>> users = client.users
|
|
>>>
|
|
>>> # List all users
|
|
>>> all_users = users.list()
|
|
>>>
|
|
>>> # Get a specific user
|
|
>>> user = users.get(123)
|
|
>>>
|
|
>>> # Create a new user
|
|
>>> new_user_data = UserCreate(
|
|
... email="user@example.com",
|
|
... name="John Doe",
|
|
... password_raw="secure_password"
|
|
... )
|
|
>>> created_user = users.create(new_user_data)
|
|
>>>
|
|
>>> # Update an existing user
|
|
>>> update_data = UserUpdate(name="Jane Doe")
|
|
>>> updated_user = users.update(123, update_data)
|
|
>>>
|
|
>>> # Delete a user
|
|
>>> users.delete(123)
|
|
"""
|
|
|
|
def list(
|
|
self,
|
|
limit: Optional[int] = None,
|
|
offset: Optional[int] = None,
|
|
search: Optional[str] = None,
|
|
order_by: str = "name",
|
|
order_direction: str = "ASC",
|
|
) -> List[User]:
|
|
"""List users with optional filtering.
|
|
|
|
Args:
|
|
limit: Maximum number of users to return
|
|
offset: Number of users to skip
|
|
search: Search term to filter users
|
|
order_by: Field to order by (name, email, createdAt)
|
|
order_direction: Order direction (ASC or DESC)
|
|
|
|
Returns:
|
|
List of User objects
|
|
|
|
Raises:
|
|
APIError: If the API request fails
|
|
ValidationError: If parameters are invalid
|
|
"""
|
|
# Validate parameters
|
|
if limit is not None and limit < 1:
|
|
raise ValidationError("limit must be greater than 0")
|
|
|
|
if offset is not None and offset < 0:
|
|
raise ValidationError("offset must be non-negative")
|
|
|
|
if order_by not in ["name", "email", "createdAt", "lastLoginAt"]:
|
|
raise ValidationError(
|
|
"order_by must be one of: name, email, createdAt, lastLoginAt"
|
|
)
|
|
|
|
if order_direction not in ["ASC", "DESC"]:
|
|
raise ValidationError("order_direction must be ASC or DESC")
|
|
|
|
# Build GraphQL query
|
|
query = """
|
|
query($filter: String, $orderBy: String) {
|
|
users {
|
|
list(filter: $filter, orderBy: $orderBy) {
|
|
id
|
|
name
|
|
email
|
|
providerKey
|
|
isSystem
|
|
isActive
|
|
isVerified
|
|
location
|
|
jobTitle
|
|
timezone
|
|
createdAt
|
|
updatedAt
|
|
lastLoginAt
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Build variables
|
|
variables: Dict[str, Any] = {}
|
|
if search:
|
|
variables["filter"] = search
|
|
if order_by:
|
|
# Wiki.js expects format like "name ASC"
|
|
variables["orderBy"] = f"{order_by} {order_direction}"
|
|
|
|
# Make request
|
|
response = self._post(
|
|
"/graphql",
|
|
json_data={"query": query, "variables": variables} if variables else {"query": query},
|
|
)
|
|
|
|
# Parse response
|
|
if "errors" in response:
|
|
raise APIError(f"GraphQL errors: {response['errors']}")
|
|
|
|
users_data = response.get("data", {}).get("users", {}).get("list", [])
|
|
|
|
# Apply client-side pagination if needed
|
|
if offset:
|
|
users_data = users_data[offset:]
|
|
if limit:
|
|
users_data = users_data[:limit]
|
|
|
|
# Convert to User objects
|
|
users = []
|
|
for user_data in users_data:
|
|
try:
|
|
normalized_data = self._normalize_user_data(user_data)
|
|
user = User(**normalized_data)
|
|
users.append(user)
|
|
except Exception as e:
|
|
raise APIError(f"Failed to parse user data: {str(e)}") from e
|
|
|
|
return users
|
|
|
|
def get(self, user_id: int) -> User:
|
|
"""Get a specific user by ID.
|
|
|
|
Args:
|
|
user_id: The user ID
|
|
|
|
Returns:
|
|
User object
|
|
|
|
Raises:
|
|
APIError: If the user is not found or request fails
|
|
ValidationError: If user_id is invalid
|
|
"""
|
|
if not isinstance(user_id, int) or user_id < 1:
|
|
raise ValidationError("user_id must be a positive integer")
|
|
|
|
# Build GraphQL query
|
|
query = """
|
|
query($id: Int!) {
|
|
users {
|
|
single(id: $id) {
|
|
id
|
|
name
|
|
email
|
|
providerKey
|
|
isSystem
|
|
isActive
|
|
isVerified
|
|
location
|
|
jobTitle
|
|
timezone
|
|
groups {
|
|
id
|
|
name
|
|
}
|
|
createdAt
|
|
updatedAt
|
|
lastLoginAt
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Make request
|
|
response = self._post(
|
|
"/graphql",
|
|
json_data={"query": query, "variables": {"id": user_id}},
|
|
)
|
|
|
|
# Parse response
|
|
if "errors" in response:
|
|
raise APIError(f"GraphQL errors: {response['errors']}")
|
|
|
|
user_data = response.get("data", {}).get("users", {}).get("single")
|
|
if not user_data:
|
|
raise APIError(f"User with ID {user_id} not found")
|
|
|
|
# Convert to User object
|
|
try:
|
|
normalized_data = self._normalize_user_data(user_data)
|
|
return User(**normalized_data)
|
|
except Exception as e:
|
|
raise APIError(f"Failed to parse user data: {str(e)}") from e
|
|
|
|
def create(self, user_data: Union[UserCreate, Dict[str, Any]]) -> User:
|
|
"""Create a new user.
|
|
|
|
Args:
|
|
user_data: User creation data (UserCreate object or dict)
|
|
|
|
Returns:
|
|
Created User object
|
|
|
|
Raises:
|
|
APIError: If user creation fails
|
|
ValidationError: If user data is invalid
|
|
"""
|
|
# Convert to UserCreate if needed
|
|
if isinstance(user_data, dict):
|
|
try:
|
|
user_data = UserCreate(**user_data)
|
|
except Exception as e:
|
|
raise ValidationError(f"Invalid user data: {str(e)}") from e
|
|
elif not isinstance(user_data, UserCreate):
|
|
raise ValidationError("user_data must be UserCreate object or dict")
|
|
|
|
# Build GraphQL mutation
|
|
mutation = """
|
|
mutation(
|
|
$email: String!,
|
|
$name: String!,
|
|
$passwordRaw: String!,
|
|
$providerKey: String!,
|
|
$groups: [Int]!,
|
|
$mustChangePassword: Boolean!,
|
|
$sendWelcomeEmail: Boolean!,
|
|
$location: String,
|
|
$jobTitle: String,
|
|
$timezone: String
|
|
) {
|
|
users {
|
|
create(
|
|
email: $email,
|
|
name: $name,
|
|
passwordRaw: $passwordRaw,
|
|
providerKey: $providerKey,
|
|
groups: $groups,
|
|
mustChangePassword: $mustChangePassword,
|
|
sendWelcomeEmail: $sendWelcomeEmail,
|
|
location: $location,
|
|
jobTitle: $jobTitle,
|
|
timezone: $timezone
|
|
) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
slug
|
|
message
|
|
}
|
|
user {
|
|
id
|
|
name
|
|
email
|
|
providerKey
|
|
isSystem
|
|
isActive
|
|
isVerified
|
|
location
|
|
jobTitle
|
|
timezone
|
|
createdAt
|
|
updatedAt
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Build variables
|
|
variables = {
|
|
"email": user_data.email,
|
|
"name": user_data.name,
|
|
"passwordRaw": user_data.password_raw,
|
|
"providerKey": user_data.provider_key,
|
|
"groups": user_data.groups,
|
|
"mustChangePassword": user_data.must_change_password,
|
|
"sendWelcomeEmail": user_data.send_welcome_email,
|
|
"location": user_data.location,
|
|
"jobTitle": user_data.job_title,
|
|
"timezone": user_data.timezone,
|
|
}
|
|
|
|
# Make request
|
|
response = self._post(
|
|
"/graphql", json_data={"query": mutation, "variables": variables}
|
|
)
|
|
|
|
# Parse response
|
|
if "errors" in response:
|
|
raise APIError(f"Failed to create user: {response['errors']}")
|
|
|
|
create_result = response.get("data", {}).get("users", {}).get("create", {})
|
|
response_result = create_result.get("responseResult", {})
|
|
|
|
if not response_result.get("succeeded"):
|
|
error_msg = response_result.get("message", "Unknown error")
|
|
raise APIError(f"User creation failed: {error_msg}")
|
|
|
|
created_user_data = create_result.get("user")
|
|
if not created_user_data:
|
|
raise APIError("User creation failed - no user data returned")
|
|
|
|
# Convert to User object
|
|
try:
|
|
normalized_data = self._normalize_user_data(created_user_data)
|
|
return User(**normalized_data)
|
|
except Exception as e:
|
|
raise APIError(f"Failed to parse created user data: {str(e)}") from e
|
|
|
|
def update(
|
|
self, user_id: int, user_data: Union[UserUpdate, Dict[str, Any]]
|
|
) -> User:
|
|
"""Update an existing user.
|
|
|
|
Args:
|
|
user_id: The user ID
|
|
user_data: User update data (UserUpdate object or dict)
|
|
|
|
Returns:
|
|
Updated User object
|
|
|
|
Raises:
|
|
APIError: If user update fails
|
|
ValidationError: If parameters are invalid
|
|
"""
|
|
if not isinstance(user_id, int) or user_id < 1:
|
|
raise ValidationError("user_id must be a positive integer")
|
|
|
|
# Convert to UserUpdate if needed
|
|
if isinstance(user_data, dict):
|
|
try:
|
|
user_data = UserUpdate(**user_data)
|
|
except Exception as e:
|
|
raise ValidationError(f"Invalid user data: {str(e)}") from e
|
|
elif not isinstance(user_data, UserUpdate):
|
|
raise ValidationError("user_data must be UserUpdate object or dict")
|
|
|
|
# Build GraphQL mutation
|
|
mutation = """
|
|
mutation(
|
|
$id: Int!,
|
|
$email: String,
|
|
$name: String,
|
|
$passwordRaw: String,
|
|
$location: String,
|
|
$jobTitle: String,
|
|
$timezone: String,
|
|
$groups: [Int],
|
|
$isActive: Boolean,
|
|
$isVerified: Boolean
|
|
) {
|
|
users {
|
|
update(
|
|
id: $id,
|
|
email: $email,
|
|
name: $name,
|
|
passwordRaw: $passwordRaw,
|
|
location: $location,
|
|
jobTitle: $jobTitle,
|
|
timezone: $timezone,
|
|
groups: $groups,
|
|
isActive: $isActive,
|
|
isVerified: $isVerified
|
|
) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
slug
|
|
message
|
|
}
|
|
user {
|
|
id
|
|
name
|
|
email
|
|
providerKey
|
|
isSystem
|
|
isActive
|
|
isVerified
|
|
location
|
|
jobTitle
|
|
timezone
|
|
createdAt
|
|
updatedAt
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Build variables (only include non-None values)
|
|
variables: Dict[str, Any] = {"id": user_id}
|
|
|
|
if user_data.name is not None:
|
|
variables["name"] = user_data.name
|
|
if user_data.email is not None:
|
|
variables["email"] = str(user_data.email)
|
|
if user_data.password_raw is not None:
|
|
variables["passwordRaw"] = user_data.password_raw
|
|
if user_data.location is not None:
|
|
variables["location"] = user_data.location
|
|
if user_data.job_title is not None:
|
|
variables["jobTitle"] = user_data.job_title
|
|
if user_data.timezone is not None:
|
|
variables["timezone"] = user_data.timezone
|
|
if user_data.groups is not None:
|
|
variables["groups"] = user_data.groups
|
|
if user_data.is_active is not None:
|
|
variables["isActive"] = user_data.is_active
|
|
if user_data.is_verified is not None:
|
|
variables["isVerified"] = user_data.is_verified
|
|
|
|
# Make request
|
|
response = self._post(
|
|
"/graphql", json_data={"query": mutation, "variables": variables}
|
|
)
|
|
|
|
# Parse response
|
|
if "errors" in response:
|
|
raise APIError(f"Failed to update user: {response['errors']}")
|
|
|
|
update_result = response.get("data", {}).get("users", {}).get("update", {})
|
|
response_result = update_result.get("responseResult", {})
|
|
|
|
if not response_result.get("succeeded"):
|
|
error_msg = response_result.get("message", "Unknown error")
|
|
raise APIError(f"User update failed: {error_msg}")
|
|
|
|
updated_user_data = update_result.get("user")
|
|
if not updated_user_data:
|
|
raise APIError("User update failed - no user data returned")
|
|
|
|
# Convert to User object
|
|
try:
|
|
normalized_data = self._normalize_user_data(updated_user_data)
|
|
return User(**normalized_data)
|
|
except Exception as e:
|
|
raise APIError(f"Failed to parse updated user data: {str(e)}") from e
|
|
|
|
def delete(self, user_id: int) -> bool:
|
|
"""Delete a user.
|
|
|
|
Args:
|
|
user_id: The user ID
|
|
|
|
Returns:
|
|
True if deletion was successful
|
|
|
|
Raises:
|
|
APIError: If user deletion fails
|
|
ValidationError: If user_id is invalid
|
|
"""
|
|
if not isinstance(user_id, int) or user_id < 1:
|
|
raise ValidationError("user_id must be a positive integer")
|
|
|
|
# Build GraphQL mutation
|
|
mutation = """
|
|
mutation($id: Int!) {
|
|
users {
|
|
delete(id: $id) {
|
|
responseResult {
|
|
succeeded
|
|
errorCode
|
|
slug
|
|
message
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Make request
|
|
response = self._post(
|
|
"/graphql",
|
|
json_data={"query": mutation, "variables": {"id": user_id}},
|
|
)
|
|
|
|
# Parse response
|
|
if "errors" in response:
|
|
raise APIError(f"Failed to delete user: {response['errors']}")
|
|
|
|
delete_result = response.get("data", {}).get("users", {}).get("delete", {})
|
|
response_result = delete_result.get("responseResult", {})
|
|
|
|
if not response_result.get("succeeded"):
|
|
error_msg = response_result.get("message", "Unknown error")
|
|
raise APIError(f"User deletion failed: {error_msg}")
|
|
|
|
return True
|
|
|
|
def search(self, query: str, limit: Optional[int] = None) -> List[User]:
|
|
"""Search for users by name or email.
|
|
|
|
Args:
|
|
query: Search query string
|
|
limit: Maximum number of results to return
|
|
|
|
Returns:
|
|
List of matching User objects
|
|
|
|
Raises:
|
|
APIError: If search fails
|
|
ValidationError: If parameters are invalid
|
|
"""
|
|
if not query or not isinstance(query, str):
|
|
raise ValidationError("query must be a non-empty string")
|
|
|
|
if limit is not None and limit < 1:
|
|
raise ValidationError("limit must be greater than 0")
|
|
|
|
# Use the list method with search parameter
|
|
return self.list(search=query, limit=limit)
|
|
|
|
def _normalize_user_data(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Normalize user data from API response to model format.
|
|
|
|
Args:
|
|
user_data: Raw user data from API
|
|
|
|
Returns:
|
|
Normalized data for User model
|
|
"""
|
|
normalized = {}
|
|
|
|
# Map API field names to model field names
|
|
field_mapping = {
|
|
"id": "id",
|
|
"name": "name",
|
|
"email": "email",
|
|
"providerKey": "provider_key",
|
|
"isSystem": "is_system",
|
|
"isActive": "is_active",
|
|
"isVerified": "is_verified",
|
|
"location": "location",
|
|
"jobTitle": "job_title",
|
|
"timezone": "timezone",
|
|
"createdAt": "created_at",
|
|
"updatedAt": "updated_at",
|
|
"lastLoginAt": "last_login_at",
|
|
}
|
|
|
|
for api_field, model_field in field_mapping.items():
|
|
if api_field in user_data:
|
|
normalized[model_field] = user_data[api_field]
|
|
|
|
# Handle groups - convert from API format
|
|
if "groups" in user_data:
|
|
if isinstance(user_data["groups"], list):
|
|
# Convert each group dict to proper format
|
|
normalized["groups"] = [
|
|
{"id": g["id"], "name": g["name"]}
|
|
for g in user_data["groups"]
|
|
if isinstance(g, dict)
|
|
]
|
|
else:
|
|
normalized["groups"] = []
|
|
else:
|
|
normalized["groups"] = []
|
|
|
|
return normalized
|