Implement Users API with complete CRUD operations

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>
This commit is contained in:
Claude
2025-10-22 18:21:46 +00:00
parent 3b11b09cde
commit 930fe50e60
5 changed files with 772 additions and 3 deletions

View File

@@ -8,7 +8,7 @@ from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry from urllib3.util.retry import Retry
from .auth import APIKeyAuth, AuthHandler from .auth import APIKeyAuth, AuthHandler
from .endpoints import PagesEndpoint from .endpoints import PagesEndpoint, UsersEndpoint
from .exceptions import ( from .exceptions import (
APIError, APIError,
AuthenticationError, AuthenticationError,
@@ -90,8 +90,8 @@ class WikiJSClient:
# Endpoint handlers # Endpoint handlers
self.pages = PagesEndpoint(self) self.pages = PagesEndpoint(self)
self.users = UsersEndpoint(self)
# Future endpoints: # Future endpoints:
# self.users = UsersEndpoint(self)
# self.groups = GroupsEndpoint(self) # self.groups = GroupsEndpoint(self)
def _create_session(self) -> requests.Session: def _create_session(self) -> requests.Session:

View File

@@ -5,9 +5,9 @@ Wiki.js API endpoints.
Implemented: Implemented:
- Pages API (CRUD operations) ✅ - Pages API (CRUD operations) ✅
- Users API (user management) ✅
Future implementations: Future implementations:
- Users API (user management)
- Groups API (group management) - Groups API (group management)
- Assets API (file management) - Assets API (file management)
- System API (system information) - System API (system information)
@@ -15,8 +15,10 @@ Future implementations:
from .base import BaseEndpoint from .base import BaseEndpoint
from .pages import PagesEndpoint from .pages import PagesEndpoint
from .users import UsersEndpoint
__all__ = [ __all__ = [
"BaseEndpoint", "BaseEndpoint",
"PagesEndpoint", "PagesEndpoint",
"UsersEndpoint",
] ]

570
wikijs/endpoints/users.py Normal file
View File

@@ -0,0 +1,570 @@
"""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

View File

@@ -2,10 +2,15 @@
from .base import BaseModel from .base import BaseModel
from .page import Page, PageCreate, PageUpdate from .page import Page, PageCreate, PageUpdate
from .user import User, UserCreate, UserGroup, UserUpdate
__all__ = [ __all__ = [
"BaseModel", "BaseModel",
"Page", "Page",
"PageCreate", "PageCreate",
"PageUpdate", "PageUpdate",
"User",
"UserCreate",
"UserUpdate",
"UserGroup",
] ]

192
wikijs/models/user.py Normal file
View File

@@ -0,0 +1,192 @@
"""User-related data models for wikijs-python-sdk."""
import re
from typing import List, Optional
from pydantic import EmailStr, Field, field_validator
from .base import BaseModel, TimestampedModel
class UserGroup(BaseModel):
"""Represents a user's group membership.
This model contains information about a user's membership
in a specific group.
"""
id: int = Field(..., description="Group ID")
name: str = Field(..., description="Group name")
class User(TimestampedModel):
"""Represents a Wiki.js user.
This model contains all user data including profile information,
authentication details, and group memberships.
"""
id: int = Field(..., description="Unique user identifier")
name: str = Field(..., description="User's full name")
email: EmailStr = Field(..., description="User's email address")
# Authentication and status
provider_key: Optional[str] = Field(None, alias="providerKey", description="Auth provider key")
is_system: bool = Field(False, alias="isSystem", description="Whether user is system user")
is_active: bool = Field(True, alias="isActive", description="Whether user is active")
is_verified: bool = Field(False, alias="isVerified", description="Whether email is verified")
# Profile information
location: Optional[str] = Field(None, description="User's location")
job_title: Optional[str] = Field(None, alias="jobTitle", description="User's job title")
timezone: Optional[str] = Field(None, description="User's timezone")
# Permissions and groups
groups: List[UserGroup] = Field(default_factory=list, description="User's groups")
# Timestamps handled by TimestampedModel
last_login_at: Optional[str] = Field(None, alias="lastLoginAt", description="Last login timestamp")
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Validate user name."""
if not v or not v.strip():
raise ValueError("Name cannot be empty")
# Check length
if len(v) < 2:
raise ValueError("Name must be at least 2 characters long")
if len(v) > 255:
raise ValueError("Name cannot exceed 255 characters")
return v.strip()
class Config:
"""Pydantic model configuration."""
populate_by_name = True
str_strip_whitespace = True
class UserCreate(BaseModel):
"""Model for creating a new user.
This model contains all required and optional fields
for creating a new Wiki.js user.
"""
email: EmailStr = Field(..., description="User's email address")
name: str = Field(..., description="User's full name")
password_raw: str = Field(..., alias="passwordRaw", description="User's password")
# Optional fields
provider_key: str = Field("local", alias="providerKey", description="Auth provider key")
groups: List[int] = Field(default_factory=list, description="Group IDs to assign")
must_change_password: bool = Field(False, alias="mustChangePassword", description="Force password change")
send_welcome_email: bool = Field(True, alias="sendWelcomeEmail", description="Send welcome email")
# Profile information
location: Optional[str] = Field(None, description="User's location")
job_title: Optional[str] = Field(None, alias="jobTitle", description="User's job title")
timezone: Optional[str] = Field(None, description="User's timezone")
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Validate user name."""
if not v or not v.strip():
raise ValueError("Name cannot be empty")
if len(v) < 2:
raise ValueError("Name must be at least 2 characters long")
if len(v) > 255:
raise ValueError("Name cannot exceed 255 characters")
return v.strip()
@field_validator("password_raw")
@classmethod
def validate_password(cls, v: str) -> str:
"""Validate password strength."""
if not v:
raise ValueError("Password cannot be empty")
if len(v) < 6:
raise ValueError("Password must be at least 6 characters long")
if len(v) > 255:
raise ValueError("Password cannot exceed 255 characters")
return v
class Config:
"""Pydantic model configuration."""
populate_by_name = True
str_strip_whitespace = True
class UserUpdate(BaseModel):
"""Model for updating an existing user.
This model contains optional fields that can be updated
for an existing Wiki.js user. All fields are optional.
"""
name: Optional[str] = Field(None, description="User's full name")
email: Optional[EmailStr] = Field(None, description="User's email address")
password_raw: Optional[str] = Field(None, alias="passwordRaw", description="New password")
# Profile information
location: Optional[str] = Field(None, description="User's location")
job_title: Optional[str] = Field(None, alias="jobTitle", description="User's job title")
timezone: Optional[str] = Field(None, description="User's timezone")
# Group assignments
groups: Optional[List[int]] = Field(None, description="Group IDs to assign")
# Status flags
is_active: Optional[bool] = Field(None, alias="isActive", description="Whether user is active")
is_verified: Optional[bool] = Field(None, alias="isVerified", description="Whether email is verified")
@field_validator("name")
@classmethod
def validate_name(cls, v: Optional[str]) -> Optional[str]:
"""Validate user name if provided."""
if v is None:
return v
if not v.strip():
raise ValueError("Name cannot be empty")
if len(v) < 2:
raise ValueError("Name must be at least 2 characters long")
if len(v) > 255:
raise ValueError("Name cannot exceed 255 characters")
return v.strip()
@field_validator("password_raw")
@classmethod
def validate_password(cls, v: Optional[str]) -> Optional[str]:
"""Validate password strength if provided."""
if v is None:
return v
if len(v) < 6:
raise ValueError("Password must be at least 6 characters long")
if len(v) > 255:
raise ValueError("Password cannot exceed 255 characters")
return v
class Config:
"""Pydantic model configuration."""
populate_by_name = True
str_strip_whitespace = True