Implement Groups API with complete CRUD operations

Implementation:
- Group data models (wikijs/models/group.py)
  - Group, GroupCreate, GroupUpdate models
  - GroupPermission, GroupPageRule, GroupUser models
  - GroupAssignUser, GroupUnassignUser models
  - Field validation and normalization

- Sync GroupsEndpoint (wikijs/endpoints/groups.py)
  - list() - List all groups with users
  - get(group_id) - Get single group
  - create(group_data) - Create new group
  - update(group_id, group_data) - Update existing group
  - delete(group_id) - Delete group
  - assign_user(group_id, user_id) - Add user to group
  - unassign_user(group_id, user_id) - Remove user from group

- Async AsyncGroupsEndpoint (wikijs/aio/endpoints/groups.py)
  - Complete async implementation
  - Identical interface to sync version
  - All CRUD operations + user management

- Integration with clients
  - WikiJSClient.groups
  - AsyncWikiJSClient.groups

GraphQL operations for all group management features.

🤖 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 20:27:22 +00:00
parent 5ad98e469e
commit fc96472d55
8 changed files with 1353 additions and 5 deletions

View File

@@ -27,7 +27,7 @@ from ..utils import (
parse_wiki_response, parse_wiki_response,
) )
from ..version import __version__ from ..version import __version__
from .endpoints import AsyncPagesEndpoint, AsyncUsersEndpoint from .endpoints import AsyncGroupsEndpoint, AsyncPagesEndpoint, AsyncUsersEndpoint
class AsyncWikiJSClient: class AsyncWikiJSClient:
@@ -104,8 +104,9 @@ class AsyncWikiJSClient:
# Endpoint handlers (will be initialized when session is created) # Endpoint handlers (will be initialized when session is created)
self.pages = AsyncPagesEndpoint(self) self.pages = AsyncPagesEndpoint(self)
self.users = AsyncUsersEndpoint(self) self.users = AsyncUsersEndpoint(self)
self.groups = AsyncGroupsEndpoint(self)
# Future endpoints: # Future endpoints:
# self.groups = AsyncGroupsEndpoint(self) # self.assets = AsyncAssetsEndpoint(self)
def _get_session(self) -> aiohttp.ClientSession: def _get_session(self) -> aiohttp.ClientSession:
"""Get or create aiohttp session. """Get or create aiohttp session.

View File

@@ -1,11 +1,13 @@
"""Async endpoint handlers for Wiki.js API.""" """Async endpoint handlers for Wiki.js API."""
from .base import AsyncBaseEndpoint from .base import AsyncBaseEndpoint
from .groups import AsyncGroupsEndpoint
from .pages import AsyncPagesEndpoint from .pages import AsyncPagesEndpoint
from .users import AsyncUsersEndpoint from .users import AsyncUsersEndpoint
__all__ = [ __all__ = [
"AsyncBaseEndpoint", "AsyncBaseEndpoint",
"AsyncGroupsEndpoint",
"AsyncPagesEndpoint", "AsyncPagesEndpoint",
"AsyncUsersEndpoint", "AsyncUsersEndpoint",
] ]

View File

@@ -0,0 +1,558 @@
"""Async groups endpoint for Wiki.js API."""
from typing import Dict, List, Union
from ...exceptions import APIError, ValidationError
from ...models import Group, GroupCreate, GroupUpdate
from .base import AsyncBaseEndpoint
class AsyncGroupsEndpoint(AsyncBaseEndpoint):
"""Async endpoint for managing Wiki.js groups.
Provides async methods to:
- List all groups
- Get a specific group by ID
- Create new groups
- Update existing groups
- Delete groups
- Assign users to groups
- Remove users from groups
"""
async def list(self) -> List[Group]:
"""List all groups asynchronously.
Returns:
List of Group objects
Raises:
APIError: If the API request fails
Example:
>>> async with AsyncWikiJSClient(...) as client:
... groups = await client.groups.list()
... for group in groups:
... print(f"{group.name}: {len(group.users)} users")
"""
query = """
query {
groups {
list {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
users {
id
name
email
}
createdAt
updatedAt
}
}
}
"""
response = await self._post("/graphql", json_data={"query": query})
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Extract and normalize groups
groups_data = response.get("data", {}).get("groups", {}).get("list", [])
return [Group(**self._normalize_group_data(g)) for g in groups_data]
async def get(self, group_id: int) -> Group:
"""Get a specific group by ID asynchronously.
Args:
group_id: The group ID
Returns:
Group object with user list
Raises:
ValidationError: If group_id is invalid
APIError: If the group is not found or API request fails
Example:
>>> async with AsyncWikiJSClient(...) as client:
... group = await client.groups.get(1)
... print(f"{group.name}: {group.permissions}")
"""
# Validate group_id
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
query = """
query ($id: Int!) {
groups {
single(id: $id) {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
users {
id
name
email
}
createdAt
updatedAt
}
}
}
"""
response = await self._post(
"/graphql", json_data={"query": query, "variables": {"id": group_id}}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Extract group data
group_data = response.get("data", {}).get("groups", {}).get("single")
if not group_data:
raise APIError(f"Group with ID {group_id} not found")
return Group(**self._normalize_group_data(group_data))
async def create(self, group_data: Union[GroupCreate, Dict]) -> Group:
"""Create a new group asynchronously.
Args:
group_data: GroupCreate object or dict with group data
Returns:
Created Group object
Raises:
ValidationError: If group data is invalid
APIError: If the API request fails
Example:
>>> from wikijs.models import GroupCreate
>>> async with AsyncWikiJSClient(...) as client:
... group_data = GroupCreate(
... name="Editors",
... permissions=["read:pages", "write:pages"]
... )
... group = await client.groups.create(group_data)
"""
# Validate and convert to dict
if isinstance(group_data, dict):
try:
group_data = GroupCreate(**group_data)
except Exception as e:
raise ValidationError(f"Invalid group data: {e}")
elif not isinstance(group_data, GroupCreate):
raise ValidationError("group_data must be a GroupCreate object or dict")
# Build mutation
mutation = """
mutation ($name: String!, $redirectOnLogin: String, $permissions: [String]!, $pageRules: [PageRuleInput]!) {
groups {
create(
name: $name
redirectOnLogin: $redirectOnLogin
permissions: $permissions
pageRules: $pageRules
) {
responseResult {
succeeded
errorCode
slug
message
}
group {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
createdAt
updatedAt
}
}
}
}
"""
variables = {
"name": group_data.name,
"redirectOnLogin": group_data.redirect_on_login or "/",
"permissions": group_data.permissions,
"pageRules": group_data.page_rules,
}
response = await self._post(
"/graphql", json_data={"query": mutation, "variables": variables}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("create", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to create group: {error_msg}")
# Extract and return created group
group_data = result.get("group")
if not group_data:
raise APIError("Group created but no data returned")
return Group(**self._normalize_group_data(group_data))
async def update(
self, group_id: int, group_data: Union[GroupUpdate, Dict]
) -> Group:
"""Update an existing group asynchronously.
Args:
group_id: The group ID
group_data: GroupUpdate object or dict with fields to update
Returns:
Updated Group object
Raises:
ValidationError: If group_id or group_data is invalid
APIError: If the API request fails
Example:
>>> from wikijs.models import GroupUpdate
>>> async with AsyncWikiJSClient(...) as client:
... update_data = GroupUpdate(
... name="Senior Editors",
... permissions=["read:pages", "write:pages", "delete:pages"]
... )
... group = await client.groups.update(1, update_data)
"""
# Validate group_id
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
# Validate and convert to dict
if isinstance(group_data, dict):
try:
group_data = GroupUpdate(**group_data)
except Exception as e:
raise ValidationError(f"Invalid group data: {e}")
elif not isinstance(group_data, GroupUpdate):
raise ValidationError("group_data must be a GroupUpdate object or dict")
# Build mutation with only non-None fields
mutation = """
mutation ($id: Int!, $name: String, $redirectOnLogin: String, $permissions: [String], $pageRules: [PageRuleInput]) {
groups {
update(
id: $id
name: $name
redirectOnLogin: $redirectOnLogin
permissions: $permissions
pageRules: $pageRules
) {
responseResult {
succeeded
errorCode
slug
message
}
group {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
createdAt
updatedAt
}
}
}
}
"""
variables = {"id": group_id}
# Add only non-None fields to variables
if group_data.name is not None:
variables["name"] = group_data.name
if group_data.redirect_on_login is not None:
variables["redirectOnLogin"] = group_data.redirect_on_login
if group_data.permissions is not None:
variables["permissions"] = group_data.permissions
if group_data.page_rules is not None:
variables["pageRules"] = group_data.page_rules
response = await self._post(
"/graphql", json_data={"query": mutation, "variables": variables}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("update", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to update group: {error_msg}")
# Extract and return updated group
group_data_response = result.get("group")
if not group_data_response:
raise APIError("Group updated but no data returned")
return Group(**self._normalize_group_data(group_data_response))
async def delete(self, group_id: int) -> bool:
"""Delete a group asynchronously.
Args:
group_id: The group ID
Returns:
True if deletion was successful
Raises:
ValidationError: If group_id is invalid
APIError: If the API request fails
Example:
>>> async with AsyncWikiJSClient(...) as client:
... success = await client.groups.delete(5)
... if success:
... print("Group deleted")
"""
# Validate group_id
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
mutation = """
mutation ($id: Int!) {
groups {
delete(id: $id) {
responseResult {
succeeded
errorCode
slug
message
}
}
}
}
"""
response = await self._post(
"/graphql", json_data={"query": mutation, "variables": {"id": group_id}}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("delete", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to delete group: {error_msg}")
return True
async def assign_user(self, group_id: int, user_id: int) -> bool:
"""Assign a user to a group asynchronously.
Args:
group_id: The group ID
user_id: The user ID
Returns:
True if assignment was successful
Raises:
ValidationError: If group_id or user_id is invalid
APIError: If the API request fails
Example:
>>> async with AsyncWikiJSClient(...) as client:
... success = await client.groups.assign_user(group_id=1, user_id=5)
... if success:
... print("User assigned to group")
"""
# Validate IDs
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
if not isinstance(user_id, int) or user_id <= 0:
raise ValidationError("user_id must be a positive integer")
mutation = """
mutation ($groupId: Int!, $userId: Int!) {
groups {
assignUser(groupId: $groupId, userId: $userId) {
responseResult {
succeeded
errorCode
slug
message
}
}
}
}
"""
response = await self._post(
"/graphql",
json_data={
"query": mutation,
"variables": {"groupId": group_id, "userId": user_id},
},
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("assignUser", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to assign user to group: {error_msg}")
return True
async def unassign_user(self, group_id: int, user_id: int) -> bool:
"""Remove a user from a group asynchronously.
Args:
group_id: The group ID
user_id: The user ID
Returns:
True if removal was successful
Raises:
ValidationError: If group_id or user_id is invalid
APIError: If the API request fails
Example:
>>> async with AsyncWikiJSClient(...) as client:
... success = await client.groups.unassign_user(group_id=1, user_id=5)
... if success:
... print("User removed from group")
"""
# Validate IDs
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
if not isinstance(user_id, int) or user_id <= 0:
raise ValidationError("user_id must be a positive integer")
mutation = """
mutation ($groupId: Int!, $userId: Int!) {
groups {
unassignUser(groupId: $groupId, userId: $userId) {
responseResult {
succeeded
errorCode
slug
message
}
}
}
}
"""
response = await self._post(
"/graphql",
json_data={
"query": mutation,
"variables": {"groupId": group_id, "userId": user_id},
},
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("unassignUser", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to remove user from group: {error_msg}")
return True
def _normalize_group_data(self, data: Dict) -> Dict:
"""Normalize group data from API response to Python naming convention.
Args:
data: Raw group data from API
Returns:
Normalized group data with snake_case field names
"""
normalized = {
"id": data.get("id"),
"name": data.get("name"),
"is_system": data.get("isSystem", False),
"redirect_on_login": data.get("redirectOnLogin"),
"permissions": data.get("permissions", []),
"page_rules": data.get("pageRules", []),
"users": data.get("users", []),
"created_at": data.get("createdAt"),
"updated_at": data.get("updatedAt"),
}
return normalized

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, UsersEndpoint from .endpoints import GroupsEndpoint, PagesEndpoint, UsersEndpoint
from .exceptions import ( from .exceptions import (
APIError, APIError,
AuthenticationError, AuthenticationError,
@@ -91,8 +91,9 @@ class WikiJSClient:
# Endpoint handlers # Endpoint handlers
self.pages = PagesEndpoint(self) self.pages = PagesEndpoint(self)
self.users = UsersEndpoint(self) self.users = UsersEndpoint(self)
self.groups = GroupsEndpoint(self)
# Future endpoints: # Future endpoints:
# self.groups = GroupsEndpoint(self) # self.assets = AssetsEndpoint(self)
def _create_session(self) -> requests.Session: def _create_session(self) -> requests.Session:
"""Create configured HTTP session with retry strategy. """Create configured HTTP session with retry strategy.

View File

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

549
wikijs/endpoints/groups.py Normal file
View File

@@ -0,0 +1,549 @@
"""Groups endpoint for Wiki.js API."""
from typing import Dict, List, Union
from ..exceptions import APIError, ValidationError
from ..models import Group, GroupCreate, GroupUpdate
from .base import BaseEndpoint
class GroupsEndpoint(BaseEndpoint):
"""Endpoint for managing Wiki.js groups.
Provides methods to:
- List all groups
- Get a specific group by ID
- Create new groups
- Update existing groups
- Delete groups
- Assign users to groups
- Remove users from groups
"""
def list(self) -> List[Group]:
"""List all groups.
Returns:
List of Group objects
Raises:
APIError: If the API request fails
Example:
>>> groups = client.groups.list()
>>> for group in groups:
... print(f"{group.name}: {len(group.users)} users")
"""
query = """
query {
groups {
list {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
users {
id
name
email
}
createdAt
updatedAt
}
}
}
"""
response = self._post("/graphql", json_data={"query": query})
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Extract and normalize groups
groups_data = response.get("data", {}).get("groups", {}).get("list", [])
return [Group(**self._normalize_group_data(g)) for g in groups_data]
def get(self, group_id: int) -> Group:
"""Get a specific group by ID.
Args:
group_id: The group ID
Returns:
Group object with user list
Raises:
ValidationError: If group_id is invalid
APIError: If the group is not found or API request fails
Example:
>>> group = client.groups.get(1)
>>> print(f"{group.name}: {group.permissions}")
"""
# Validate group_id
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
query = """
query ($id: Int!) {
groups {
single(id: $id) {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
users {
id
name
email
}
createdAt
updatedAt
}
}
}
"""
response = self._post(
"/graphql", json_data={"query": query, "variables": {"id": group_id}}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Extract group data
group_data = response.get("data", {}).get("groups", {}).get("single")
if not group_data:
raise APIError(f"Group with ID {group_id} not found")
return Group(**self._normalize_group_data(group_data))
def create(self, group_data: Union[GroupCreate, Dict]) -> Group:
"""Create a new group.
Args:
group_data: GroupCreate object or dict with group data
Returns:
Created Group object
Raises:
ValidationError: If group data is invalid
APIError: If the API request fails
Example:
>>> from wikijs.models import GroupCreate
>>> group_data = GroupCreate(
... name="Editors",
... permissions=["read:pages", "write:pages"]
... )
>>> group = client.groups.create(group_data)
"""
# Validate and convert to dict
if isinstance(group_data, dict):
try:
group_data = GroupCreate(**group_data)
except Exception as e:
raise ValidationError(f"Invalid group data: {e}")
elif not isinstance(group_data, GroupCreate):
raise ValidationError("group_data must be a GroupCreate object or dict")
# Build mutation
mutation = """
mutation ($name: String!, $redirectOnLogin: String, $permissions: [String]!, $pageRules: [PageRuleInput]!) {
groups {
create(
name: $name
redirectOnLogin: $redirectOnLogin
permissions: $permissions
pageRules: $pageRules
) {
responseResult {
succeeded
errorCode
slug
message
}
group {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
createdAt
updatedAt
}
}
}
}
"""
variables = {
"name": group_data.name,
"redirectOnLogin": group_data.redirect_on_login or "/",
"permissions": group_data.permissions,
"pageRules": group_data.page_rules,
}
response = self._post(
"/graphql", json_data={"query": mutation, "variables": variables}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("create", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to create group: {error_msg}")
# Extract and return created group
group_data = result.get("group")
if not group_data:
raise APIError("Group created but no data returned")
return Group(**self._normalize_group_data(group_data))
def update(self, group_id: int, group_data: Union[GroupUpdate, Dict]) -> Group:
"""Update an existing group.
Args:
group_id: The group ID
group_data: GroupUpdate object or dict with fields to update
Returns:
Updated Group object
Raises:
ValidationError: If group_id or group_data is invalid
APIError: If the API request fails
Example:
>>> from wikijs.models import GroupUpdate
>>> update_data = GroupUpdate(
... name="Senior Editors",
... permissions=["read:pages", "write:pages", "delete:pages"]
... )
>>> group = client.groups.update(1, update_data)
"""
# Validate group_id
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
# Validate and convert to dict
if isinstance(group_data, dict):
try:
group_data = GroupUpdate(**group_data)
except Exception as e:
raise ValidationError(f"Invalid group data: {e}")
elif not isinstance(group_data, GroupUpdate):
raise ValidationError("group_data must be a GroupUpdate object or dict")
# Build mutation with only non-None fields
mutation = """
mutation ($id: Int!, $name: String, $redirectOnLogin: String, $permissions: [String], $pageRules: [PageRuleInput]) {
groups {
update(
id: $id
name: $name
redirectOnLogin: $redirectOnLogin
permissions: $permissions
pageRules: $pageRules
) {
responseResult {
succeeded
errorCode
slug
message
}
group {
id
name
isSystem
redirectOnLogin
permissions
pageRules {
id
path
roles
match
deny
locales
}
createdAt
updatedAt
}
}
}
}
"""
variables = {"id": group_id}
# Add only non-None fields to variables
if group_data.name is not None:
variables["name"] = group_data.name
if group_data.redirect_on_login is not None:
variables["redirectOnLogin"] = group_data.redirect_on_login
if group_data.permissions is not None:
variables["permissions"] = group_data.permissions
if group_data.page_rules is not None:
variables["pageRules"] = group_data.page_rules
response = self._post(
"/graphql", json_data={"query": mutation, "variables": variables}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("update", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to update group: {error_msg}")
# Extract and return updated group
group_data_response = result.get("group")
if not group_data_response:
raise APIError("Group updated but no data returned")
return Group(**self._normalize_group_data(group_data_response))
def delete(self, group_id: int) -> bool:
"""Delete a group.
Args:
group_id: The group ID
Returns:
True if deletion was successful
Raises:
ValidationError: If group_id is invalid
APIError: If the API request fails
Example:
>>> success = client.groups.delete(5)
>>> if success:
... print("Group deleted")
"""
# Validate group_id
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
mutation = """
mutation ($id: Int!) {
groups {
delete(id: $id) {
responseResult {
succeeded
errorCode
slug
message
}
}
}
}
"""
response = self._post(
"/graphql", json_data={"query": mutation, "variables": {"id": group_id}}
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("delete", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to delete group: {error_msg}")
return True
def assign_user(self, group_id: int, user_id: int) -> bool:
"""Assign a user to a group.
Args:
group_id: The group ID
user_id: The user ID
Returns:
True if assignment was successful
Raises:
ValidationError: If group_id or user_id is invalid
APIError: If the API request fails
Example:
>>> success = client.groups.assign_user(group_id=1, user_id=5)
>>> if success:
... print("User assigned to group")
"""
# Validate IDs
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
if not isinstance(user_id, int) or user_id <= 0:
raise ValidationError("user_id must be a positive integer")
mutation = """
mutation ($groupId: Int!, $userId: Int!) {
groups {
assignUser(groupId: $groupId, userId: $userId) {
responseResult {
succeeded
errorCode
slug
message
}
}
}
}
"""
response = self._post(
"/graphql",
json_data={
"query": mutation,
"variables": {"groupId": group_id, "userId": user_id},
},
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("assignUser", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to assign user to group: {error_msg}")
return True
def unassign_user(self, group_id: int, user_id: int) -> bool:
"""Remove a user from a group.
Args:
group_id: The group ID
user_id: The user ID
Returns:
True if removal was successful
Raises:
ValidationError: If group_id or user_id is invalid
APIError: If the API request fails
Example:
>>> success = client.groups.unassign_user(group_id=1, user_id=5)
>>> if success:
... print("User removed from group")
"""
# Validate IDs
if not isinstance(group_id, int) or group_id <= 0:
raise ValidationError("group_id must be a positive integer")
if not isinstance(user_id, int) or user_id <= 0:
raise ValidationError("user_id must be a positive integer")
mutation = """
mutation ($groupId: Int!, $userId: Int!) {
groups {
unassignUser(groupId: $groupId, userId: $userId) {
responseResult {
succeeded
errorCode
slug
message
}
}
}
}
"""
response = self._post(
"/graphql",
json_data={
"query": mutation,
"variables": {"groupId": group_id, "userId": user_id},
},
)
# Check for GraphQL errors
if "errors" in response:
raise APIError(f"GraphQL errors: {response['errors']}")
# Check response result
result = response.get("data", {}).get("groups", {}).get("unassignUser", {})
response_result = result.get("responseResult", {})
if not response_result.get("succeeded"):
error_msg = response_result.get("message", "Unknown error")
raise APIError(f"Failed to remove user from group: {error_msg}")
return True
def _normalize_group_data(self, data: Dict) -> Dict:
"""Normalize group data from API response to Python naming convention.
Args:
data: Raw group data from API
Returns:
Normalized group data with snake_case field names
"""
normalized = {
"id": data.get("id"),
"name": data.get("name"),
"is_system": data.get("isSystem", False),
"redirect_on_login": data.get("redirectOnLogin"),
"permissions": data.get("permissions", []),
"page_rules": data.get("pageRules", []),
"users": data.get("users", []),
"created_at": data.get("createdAt"),
"updated_at": data.get("updatedAt"),
}
return normalized

View File

@@ -1,11 +1,29 @@
"""Data models for wikijs-python-sdk.""" """Data models for wikijs-python-sdk."""
from .base import BaseModel from .base import BaseModel
from .group import (
Group,
GroupAssignUser,
GroupCreate,
GroupPageRule,
GroupPermission,
GroupUnassignUser,
GroupUpdate,
GroupUser,
)
from .page import Page, PageCreate, PageUpdate from .page import Page, PageCreate, PageUpdate
from .user import User, UserCreate, UserGroup, UserUpdate from .user import User, UserCreate, UserGroup, UserUpdate
__all__ = [ __all__ = [
"BaseModel", "BaseModel",
"Group",
"GroupAssignUser",
"GroupCreate",
"GroupPageRule",
"GroupPermission",
"GroupUnassignUser",
"GroupUpdate",
"GroupUser",
"Page", "Page",
"PageCreate", "PageCreate",
"PageUpdate", "PageUpdate",

217
wikijs/models/group.py Normal file
View File

@@ -0,0 +1,217 @@
"""Data models for Wiki.js groups."""
from typing import List, Optional
from pydantic import Field, field_validator
from .base import BaseModel, TimestampedModel
class GroupPermission(BaseModel):
"""Group permission model."""
id: str = Field(..., description="Permission identifier")
name: Optional[str] = Field(None, description="Permission name")
class Config:
"""Pydantic configuration."""
populate_by_name = True
class GroupPageRule(BaseModel):
"""Group page access rule model."""
id: str = Field(..., description="Rule identifier")
path: str = Field(..., description="Page path pattern")
roles: List[str] = Field(default_factory=list, description="Allowed roles")
match: str = Field(default="START", description="Match type (START, EXACT, REGEX)")
deny: bool = Field(default=False, description="Whether this is a deny rule")
locales: List[str] = Field(default_factory=list, description="Allowed locales")
class Config:
"""Pydantic configuration."""
populate_by_name = True
class GroupUser(BaseModel):
"""User member of a group (minimal representation)."""
id: int = Field(..., description="User ID")
name: str = Field(..., description="User name")
email: str = Field(..., description="User email")
class Config:
"""Pydantic configuration."""
populate_by_name = True
class Group(TimestampedModel):
"""Wiki.js group model.
Represents a complete group with all fields.
Attributes:
id: Group ID
name: Group name
is_system: Whether this is a system group
redirect_on_login: Path to redirect to on login
permissions: List of group permissions
page_rules: List of page access rules
users: List of users in this group (only populated in get operations)
created_at: Creation timestamp
updated_at: Last update timestamp
"""
id: int = Field(..., description="Group ID")
name: str = Field(..., min_length=1, max_length=255, description="Group name")
is_system: bool = Field(
default=False, alias="isSystem", description="System group flag"
)
redirect_on_login: Optional[str] = Field(
None, alias="redirectOnLogin", description="Redirect path on login"
)
permissions: List[str] = Field(
default_factory=list, description="Permission identifiers"
)
page_rules: List[GroupPageRule] = Field(
default_factory=list, alias="pageRules", description="Page access rules"
)
users: List[GroupUser] = Field(
default_factory=list, description="Users in this group"
)
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Validate group name."""
if not v or not v.strip():
raise ValueError("Group name cannot be empty")
if len(v.strip()) < 1:
raise ValueError("Group name must be at least 1 character")
if len(v) > 255:
raise ValueError("Group name cannot exceed 255 characters")
return v.strip()
class Config:
"""Pydantic configuration."""
populate_by_name = True
class GroupCreate(BaseModel):
"""Model for creating a new group.
Attributes:
name: Group name (required)
redirect_on_login: Path to redirect to on login
permissions: List of permission identifiers
page_rules: List of page access rule configurations
"""
name: str = Field(..., min_length=1, max_length=255, description="Group name")
redirect_on_login: Optional[str] = Field(
None, alias="redirectOnLogin", description="Redirect path on login"
)
permissions: List[str] = Field(
default_factory=list, description="Permission identifiers"
)
page_rules: List[dict] = Field(
default_factory=list, alias="pageRules", description="Page access rules"
)
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Validate group name."""
if not v or not v.strip():
raise ValueError("Group name cannot be empty")
if len(v.strip()) < 1:
raise ValueError("Group name must be at least 1 character")
if len(v) > 255:
raise ValueError("Group name cannot exceed 255 characters")
return v.strip()
class Config:
"""Pydantic configuration."""
populate_by_name = True
class GroupUpdate(BaseModel):
"""Model for updating an existing group.
All fields are optional to support partial updates.
Attributes:
name: Updated group name
redirect_on_login: Updated redirect path
permissions: Updated permission list
page_rules: Updated page access rules
"""
name: Optional[str] = Field(
None, min_length=1, max_length=255, description="Group name"
)
redirect_on_login: Optional[str] = Field(
None, alias="redirectOnLogin", description="Redirect path on login"
)
permissions: Optional[List[str]] = Field(None, description="Permission identifiers")
page_rules: Optional[List[dict]] = Field(
None, alias="pageRules", description="Page access rules"
)
@field_validator("name")
@classmethod
def validate_name(cls, v: Optional[str]) -> Optional[str]:
"""Validate group name if provided."""
if v is None:
return v
if not v or not v.strip():
raise ValueError("Group name cannot be empty")
if len(v.strip()) < 1:
raise ValueError("Group name must be at least 1 character")
if len(v) > 255:
raise ValueError("Group name cannot exceed 255 characters")
return v.strip()
class Config:
"""Pydantic configuration."""
populate_by_name = True
class GroupAssignUser(BaseModel):
"""Model for assigning a user to a group.
Attributes:
group_id: Group ID
user_id: User ID
"""
group_id: int = Field(..., alias="groupId", description="Group ID")
user_id: int = Field(..., alias="userId", description="User ID")
class Config:
"""Pydantic configuration."""
populate_by_name = True
class GroupUnassignUser(BaseModel):
"""Model for removing a user from a group.
Attributes:
group_id: Group ID
user_id: User ID
"""
group_id: int = Field(..., alias="groupId", description="Group ID")
user_id: int = Field(..., alias="userId", description="User ID")
class Config:
"""Pydantic configuration."""
populate_by_name = True