From fc96472d555304f40a99229685dec029497e1d93 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 22 Oct 2025 20:27:22 +0000 Subject: [PATCH] Implement Groups API with complete CRUD operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- wikijs/aio/client.py | 5 +- wikijs/aio/endpoints/__init__.py | 2 + wikijs/aio/endpoints/groups.py | 558 +++++++++++++++++++++++++++++++ wikijs/client.py | 5 +- wikijs/endpoints/__init__.py | 4 +- wikijs/endpoints/groups.py | 549 ++++++++++++++++++++++++++++++ wikijs/models/__init__.py | 18 + wikijs/models/group.py | 217 ++++++++++++ 8 files changed, 1353 insertions(+), 5 deletions(-) create mode 100644 wikijs/aio/endpoints/groups.py create mode 100644 wikijs/endpoints/groups.py create mode 100644 wikijs/models/group.py diff --git a/wikijs/aio/client.py b/wikijs/aio/client.py index 5f3f8b6..d6b82d3 100644 --- a/wikijs/aio/client.py +++ b/wikijs/aio/client.py @@ -27,7 +27,7 @@ from ..utils import ( parse_wiki_response, ) from ..version import __version__ -from .endpoints import AsyncPagesEndpoint, AsyncUsersEndpoint +from .endpoints import AsyncGroupsEndpoint, AsyncPagesEndpoint, AsyncUsersEndpoint class AsyncWikiJSClient: @@ -104,8 +104,9 @@ class AsyncWikiJSClient: # Endpoint handlers (will be initialized when session is created) self.pages = AsyncPagesEndpoint(self) self.users = AsyncUsersEndpoint(self) + self.groups = AsyncGroupsEndpoint(self) # Future endpoints: - # self.groups = AsyncGroupsEndpoint(self) + # self.assets = AsyncAssetsEndpoint(self) def _get_session(self) -> aiohttp.ClientSession: """Get or create aiohttp session. diff --git a/wikijs/aio/endpoints/__init__.py b/wikijs/aio/endpoints/__init__.py index 69ac238..f59788a 100644 --- a/wikijs/aio/endpoints/__init__.py +++ b/wikijs/aio/endpoints/__init__.py @@ -1,11 +1,13 @@ """Async endpoint handlers for Wiki.js API.""" from .base import AsyncBaseEndpoint +from .groups import AsyncGroupsEndpoint from .pages import AsyncPagesEndpoint from .users import AsyncUsersEndpoint __all__ = [ "AsyncBaseEndpoint", + "AsyncGroupsEndpoint", "AsyncPagesEndpoint", "AsyncUsersEndpoint", ] diff --git a/wikijs/aio/endpoints/groups.py b/wikijs/aio/endpoints/groups.py new file mode 100644 index 0000000..c550187 --- /dev/null +++ b/wikijs/aio/endpoints/groups.py @@ -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 diff --git a/wikijs/client.py b/wikijs/client.py index 3f7e8e8..5f04d15 100644 --- a/wikijs/client.py +++ b/wikijs/client.py @@ -8,7 +8,7 @@ from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from .auth import APIKeyAuth, AuthHandler -from .endpoints import PagesEndpoint, UsersEndpoint +from .endpoints import GroupsEndpoint, PagesEndpoint, UsersEndpoint from .exceptions import ( APIError, AuthenticationError, @@ -91,8 +91,9 @@ class WikiJSClient: # Endpoint handlers self.pages = PagesEndpoint(self) self.users = UsersEndpoint(self) + self.groups = GroupsEndpoint(self) # Future endpoints: - # self.groups = GroupsEndpoint(self) + # self.assets = AssetsEndpoint(self) def _create_session(self) -> requests.Session: """Create configured HTTP session with retry strategy. diff --git a/wikijs/endpoints/__init__.py b/wikijs/endpoints/__init__.py index 4132af6..b102ba2 100644 --- a/wikijs/endpoints/__init__.py +++ b/wikijs/endpoints/__init__.py @@ -6,19 +6,21 @@ Wiki.js API endpoints. Implemented: - Pages API (CRUD operations) ✅ - Users API (user management) ✅ +- Groups API (group management) ✅ Future implementations: -- Groups API (group management) - Assets API (file management) - System API (system information) """ from .base import BaseEndpoint +from .groups import GroupsEndpoint from .pages import PagesEndpoint from .users import UsersEndpoint __all__ = [ "BaseEndpoint", + "GroupsEndpoint", "PagesEndpoint", "UsersEndpoint", ] diff --git a/wikijs/endpoints/groups.py b/wikijs/endpoints/groups.py new file mode 100644 index 0000000..30f111b --- /dev/null +++ b/wikijs/endpoints/groups.py @@ -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 diff --git a/wikijs/models/__init__.py b/wikijs/models/__init__.py index 7a535cf..e03c701 100644 --- a/wikijs/models/__init__.py +++ b/wikijs/models/__init__.py @@ -1,11 +1,29 @@ """Data models for wikijs-python-sdk.""" from .base import BaseModel +from .group import ( + Group, + GroupAssignUser, + GroupCreate, + GroupPageRule, + GroupPermission, + GroupUnassignUser, + GroupUpdate, + GroupUser, +) from .page import Page, PageCreate, PageUpdate from .user import User, UserCreate, UserGroup, UserUpdate __all__ = [ "BaseModel", + "Group", + "GroupAssignUser", + "GroupCreate", + "GroupPageRule", + "GroupPermission", + "GroupUnassignUser", + "GroupUpdate", + "GroupUser", "Page", "PageCreate", "PageUpdate", diff --git a/wikijs/models/group.py b/wikijs/models/group.py new file mode 100644 index 0000000..64b246f --- /dev/null +++ b/wikijs/models/group.py @@ -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