""" Users Handler - manages Appwrite user operations Phase I.2: 12 tools - User CRUD: 5 (list, get, create, update, delete) - User Properties: 4 (update_email, update_phone, update_status, update_labels) - Sessions: 3 (list_sessions, delete_sessions, delete_session) """ import json from typing import Any from plugins.appwrite.client import AppwriteClient def get_tool_specifications() -> list[dict[str, Any]]: """Return tool specifications for ToolGenerator (12 tools)""" return [ # ===================== # USER CRUD (5) # ===================== { "name": "list_users", "method_name": "list_users", "description": "List all users in the project with optional filtering and search.", "schema": { "type": "object", "properties": { "queries": { "anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}], "description": "Query strings (e.g., 'limit(25)', 'offset(0)')", }, "search": { "anyOf": [{"type": "string"}, {"type": "null"}], "description": "Search term to filter users by name or email", }, }, "required": [], }, "scope": "read", }, { "name": "get_user", "method_name": "get_user", "description": "Get user details by ID.", "schema": { "type": "object", "properties": {"user_id": {"type": "string", "description": "User ID"}}, "required": ["user_id"], }, "scope": "read", }, { "name": "create_user", "method_name": "create_user", "description": "Create a new user. Use 'unique()' for auto-generated user ID.", "schema": { "type": "object", "properties": { "user_id": { "type": "string", "description": "Unique user ID. Use 'unique()' for auto-generation", }, "email": { "anyOf": [{"type": "string", "format": "email"}, {"type": "null"}], "description": "User email address", }, "phone": { "anyOf": [{"type": "string"}, {"type": "null"}], "description": "User phone number (E.164 format)", }, "password": { "anyOf": [{"type": "string", "minLength": 8}, {"type": "null"}], "description": "User password (min 8 characters)", }, "name": { "anyOf": [{"type": "string"}, {"type": "null"}], "description": "User display name", }, }, "required": ["user_id"], }, "scope": "write", }, { "name": "update_user_name", "method_name": "update_user_name", "description": "Update user display name.", "schema": { "type": "object", "properties": { "user_id": {"type": "string", "description": "User ID"}, "name": {"type": "string", "description": "New display name"}, }, "required": ["user_id", "name"], }, "scope": "write", }, { "name": "delete_user", "method_name": "delete_user", "description": "Delete a user. This action is irreversible.", "schema": { "type": "object", "properties": {"user_id": {"type": "string", "description": "User ID to delete"}}, "required": ["user_id"], }, "scope": "admin", }, # ===================== # USER PROPERTIES (4) # ===================== { "name": "update_user_email", "method_name": "update_user_email", "description": "Update user email address.", "schema": { "type": "object", "properties": { "user_id": {"type": "string", "description": "User ID"}, "email": { "type": "string", "format": "email", "description": "New email address", }, }, "required": ["user_id", "email"], }, "scope": "write", }, { "name": "update_user_phone", "method_name": "update_user_phone", "description": "Update user phone number.", "schema": { "type": "object", "properties": { "user_id": {"type": "string", "description": "User ID"}, "number": { "type": "string", "description": "New phone number (E.164 format, e.g., +14155552671)", }, }, "required": ["user_id", "number"], }, "scope": "write", }, { "name": "update_user_status", "method_name": "update_user_status", "description": "Enable or disable a user account.", "schema": { "type": "object", "properties": { "user_id": {"type": "string", "description": "User ID"}, "status": { "type": "boolean", "description": "True to enable, False to disable", }, }, "required": ["user_id", "status"], }, "scope": "admin", }, { "name": "update_user_labels", "method_name": "update_user_labels", "description": "Set user labels for permission management. Labels can be used in permission strings like 'label:admin'.", "schema": { "type": "object", "properties": { "user_id": {"type": "string", "description": "User ID"}, "labels": { "type": "array", "items": {"type": "string"}, "description": "Array of label strings (replaces existing labels)", }, }, "required": ["user_id", "labels"], }, "scope": "admin", }, # ===================== # SESSIONS (3) # ===================== { "name": "list_user_sessions", "method_name": "list_user_sessions", "description": "List all active sessions for a user.", "schema": { "type": "object", "properties": {"user_id": {"type": "string", "description": "User ID"}}, "required": ["user_id"], }, "scope": "read", }, { "name": "delete_user_sessions", "method_name": "delete_user_sessions", "description": "Delete all sessions for a user (force logout everywhere).", "schema": { "type": "object", "properties": {"user_id": {"type": "string", "description": "User ID"}}, "required": ["user_id"], }, "scope": "admin", }, { "name": "delete_user_session", "method_name": "delete_user_session", "description": "Delete a specific user session.", "schema": { "type": "object", "properties": { "user_id": {"type": "string", "description": "User ID"}, "session_id": {"type": "string", "description": "Session ID to delete"}, }, "required": ["user_id", "session_id"], }, "scope": "admin", }, ] # ===================== # HANDLER FUNCTIONS # ===================== async def list_users( client: AppwriteClient, queries: list[str] | None = None, search: str | None = None ) -> str: """List all users.""" try: result = await client.list_users(queries=queries, search=search) users = result.get("users", []) response = {"success": True, "total": result.get("total", len(users)), "users": users} return json.dumps(response, indent=2, ensure_ascii=False) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_user(client: AppwriteClient, user_id: str) -> str: """Get user by ID.""" try: result = await client.get_user(user_id) return json.dumps({"success": True, "user": result}, indent=2, ensure_ascii=False) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def create_user( client: AppwriteClient, user_id: str, email: str | None = None, phone: str | None = None, password: str | None = None, name: str | None = None, ) -> str: """Create a new user.""" try: result = await client.create_user( user_id=user_id, email=email, phone=phone, password=password, name=name ) return json.dumps( {"success": True, "message": "User created successfully", "user": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_user_name(client: AppwriteClient, user_id: str, name: str) -> str: """Update user name.""" try: result = await client.update_user_name(user_id=user_id, name=name) return json.dumps( {"success": True, "message": "User name updated successfully", "user": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def delete_user(client: AppwriteClient, user_id: str) -> str: """Delete user.""" try: await client.delete_user(user_id) return json.dumps( {"success": True, "message": f"User '{user_id}' deleted successfully"}, indent=2 ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_user_email(client: AppwriteClient, user_id: str, email: str) -> str: """Update user email.""" try: result = await client.update_user_email(user_id=user_id, email=email) return json.dumps( {"success": True, "message": "User email updated successfully", "user": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_user_phone(client: AppwriteClient, user_id: str, number: str) -> str: """Update user phone.""" try: result = await client.update_user_phone(user_id=user_id, number=number) return json.dumps( {"success": True, "message": "User phone updated successfully", "user": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_user_status(client: AppwriteClient, user_id: str, status: bool) -> str: """Update user status (enable/disable).""" try: result = await client.update_user_status(user_id=user_id, status=status) action = "enabled" if status else "disabled" return json.dumps( {"success": True, "message": f"User {action} successfully", "user": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_user_labels(client: AppwriteClient, user_id: str, labels: list[str]) -> str: """Update user labels.""" try: result = await client.update_user_labels(user_id=user_id, labels=labels) return json.dumps( {"success": True, "message": f"User labels updated: {labels}", "user": result}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def list_user_sessions(client: AppwriteClient, user_id: str) -> str: """List user sessions.""" try: result = await client.list_user_sessions(user_id) sessions = result.get("sessions", []) response = { "success": True, "user_id": user_id, "total": result.get("total", len(sessions)), "sessions": sessions, } return json.dumps(response, indent=2, ensure_ascii=False) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def delete_user_sessions(client: AppwriteClient, user_id: str) -> str: """Delete all user sessions.""" try: await client.delete_user_sessions(user_id) return json.dumps( {"success": True, "message": f"All sessions for user '{user_id}' deleted"}, indent=2 ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def delete_user_session(client: AppwriteClient, user_id: str, session_id: str) -> str: """Delete a specific session.""" try: await client.delete_user_session(user_id, session_id) return json.dumps({"success": True, "message": f"Session '{session_id}' deleted"}, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2)