c73e39f6e4
- Fix Python formatting (sync no longer strips blank lines from .py files) - Remove test_community_build.py (tests private sync module) - Fix ruff warnings in test files - Add CODE_OF_CONDUCT.md - Add .github/PULL_REQUEST_TEMPLATE.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1613 lines
54 KiB
Python
1613 lines
54 KiB
Python
"""
|
|
Appwrite REST API Client (Self-Hosted)
|
|
|
|
Handles all HTTP communication with Appwrite Self-Hosted APIs.
|
|
Uses X-Appwrite-Project and X-Appwrite-Key headers for authentication.
|
|
|
|
APIs:
|
|
- Databases (/databases/) - Database, Collections, Documents
|
|
- Users (/users/) - User management
|
|
- Teams (/teams/) - Team management
|
|
- Storage (/storage/) - Buckets and Files
|
|
- Functions (/functions/) - Serverless Functions
|
|
- Messaging (/messaging/) - Email, SMS, Push
|
|
- Health (/health/) - Service health checks
|
|
- Avatars (/avatars/) - Avatar utilities
|
|
"""
|
|
|
|
import base64
|
|
import logging
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
|
|
class AppwriteClient:
|
|
"""
|
|
Appwrite Self-Hosted API client.
|
|
|
|
All requests include X-Appwrite-Project and X-Appwrite-Key headers.
|
|
API key provides admin access and bypasses rate limits.
|
|
"""
|
|
|
|
def __init__(self, base_url: str, project_id: str, api_key: str):
|
|
"""
|
|
Initialize Appwrite API client.
|
|
|
|
Args:
|
|
base_url: Appwrite instance URL (e.g., https://appwrite.example.com/v1)
|
|
project_id: Appwrite project ID
|
|
api_key: API key with required scopes
|
|
"""
|
|
self.base_url = base_url.rstrip("/")
|
|
self.project_id = project_id
|
|
self.api_key = api_key
|
|
|
|
# Initialize logger
|
|
self.logger = logging.getLogger(f"AppwriteClient.{base_url}")
|
|
|
|
def _get_headers(self, additional_headers: dict | None = None) -> dict[str, str]:
|
|
"""
|
|
Get request headers with API key authentication.
|
|
|
|
Args:
|
|
additional_headers: Additional headers to include
|
|
|
|
Returns:
|
|
Dict: Headers with authentication
|
|
"""
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"X-Appwrite-Project": self.project_id,
|
|
"X-Appwrite-Key": self.api_key,
|
|
}
|
|
|
|
if additional_headers:
|
|
headers.update(additional_headers)
|
|
|
|
return headers
|
|
|
|
async def request(
|
|
self,
|
|
method: str,
|
|
endpoint: str,
|
|
params: dict | list | None = None,
|
|
json_data: dict | list | None = None,
|
|
data: bytes | None = None,
|
|
headers_override: dict | None = None,
|
|
) -> Any:
|
|
"""
|
|
Make authenticated request to Appwrite API.
|
|
|
|
Args:
|
|
method: HTTP method
|
|
endpoint: API endpoint (with leading /)
|
|
params: Query parameters (dict or list of tuples for repeated keys)
|
|
json_data: JSON body data
|
|
data: Raw binary data (for file uploads)
|
|
headers_override: Override/add headers
|
|
|
|
Returns:
|
|
API response
|
|
|
|
Raises:
|
|
Exception: On API errors
|
|
"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
headers = self._get_headers(headers_override)
|
|
|
|
# Remove Content-Type for binary data (multipart)
|
|
if data is not None:
|
|
headers.pop("Content-Type", None)
|
|
|
|
# Process params: convert dict to list of tuples, handle lists and booleans
|
|
if params:
|
|
if isinstance(params, dict):
|
|
clean_params = []
|
|
for k, v in params.items():
|
|
if v is None:
|
|
continue
|
|
# Handle list values - create multiple tuples with same key
|
|
if isinstance(v, list):
|
|
for item in v:
|
|
clean_params.append((k, item))
|
|
# Convert booleans to lowercase strings for URL params
|
|
elif isinstance(v, bool):
|
|
clean_params.append((k, "true" if v else "false"))
|
|
else:
|
|
clean_params.append((k, v))
|
|
params = clean_params
|
|
# If already a list, just filter None values
|
|
elif isinstance(params, list):
|
|
params = [(k, v) for k, v in params if v is not None]
|
|
|
|
# Process JSON body: filter None values and coerce string types back to native
|
|
if isinstance(json_data, dict):
|
|
json_data = self._coerce_json_types(json_data)
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
kwargs = {
|
|
"method": method,
|
|
"url": url,
|
|
"headers": headers,
|
|
}
|
|
|
|
if params:
|
|
kwargs["params"] = params
|
|
if json_data is not None:
|
|
kwargs["json"] = json_data
|
|
if data:
|
|
kwargs["data"] = data
|
|
|
|
async with session.request(**kwargs) as response:
|
|
self.logger.debug(f"Response status: {response.status}")
|
|
|
|
# Handle 204 No Content
|
|
if response.status == 204:
|
|
return {"success": True}
|
|
|
|
# Handle binary responses (file downloads)
|
|
content_type = response.headers.get("Content-Type", "")
|
|
if "application/json" not in content_type and response.status < 400:
|
|
# Return binary data as base64
|
|
binary_data = await response.read()
|
|
return {
|
|
"data": base64.b64encode(binary_data).decode(),
|
|
"content_type": content_type,
|
|
"size": len(binary_data),
|
|
}
|
|
|
|
# Parse JSON response
|
|
try:
|
|
response_data = await response.json()
|
|
except Exception:
|
|
response_text = await response.text()
|
|
if response.status >= 400:
|
|
raise Exception(
|
|
f"Appwrite API error (status {response.status}): {response_text}"
|
|
)
|
|
return {"success": True, "message": response_text}
|
|
|
|
# Check for errors
|
|
if response.status >= 400:
|
|
error_msg = self._extract_error_message(response_data)
|
|
raise Exception(f"Appwrite API error (status {response.status}): {error_msg}")
|
|
|
|
return response_data
|
|
|
|
def _extract_error_message(self, response_data: Any) -> str:
|
|
"""Extract error message from Appwrite error response."""
|
|
if isinstance(response_data, dict):
|
|
if "message" in response_data:
|
|
return response_data["message"]
|
|
if "error" in response_data:
|
|
return response_data["error"]
|
|
if "type" in response_data:
|
|
return (
|
|
f"{response_data.get('type')}: {response_data.get('message', 'Unknown error')}"
|
|
)
|
|
return str(response_data)
|
|
|
|
def _coerce_json_types(self, data: dict) -> dict:
|
|
"""
|
|
Coerce string representations back to native JSON types.
|
|
|
|
MCP clients may send "true"/"false" strings instead of boolean,
|
|
or "123" strings instead of integers. This method converts them
|
|
back to native types for proper Appwrite API compatibility.
|
|
|
|
Args:
|
|
data: Dictionary with potential string-encoded values
|
|
|
|
Returns:
|
|
Dictionary with properly typed values
|
|
"""
|
|
result = {}
|
|
for k, v in data.items():
|
|
if v is None:
|
|
continue # Filter out None values
|
|
|
|
# Handle nested dictionaries
|
|
if isinstance(v, dict):
|
|
result[k] = self._coerce_json_types(v)
|
|
# Handle lists
|
|
elif isinstance(v, list):
|
|
result[k] = [
|
|
(
|
|
self._coerce_json_types(item)
|
|
if isinstance(item, dict)
|
|
else self._coerce_value(item)
|
|
)
|
|
for item in v
|
|
]
|
|
else:
|
|
result[k] = self._coerce_value(v)
|
|
|
|
return result
|
|
|
|
def _coerce_value(self, value: Any) -> Any:
|
|
"""
|
|
Coerce a single value to its native type.
|
|
|
|
Converts:
|
|
- "true"/"false" strings to boolean
|
|
- Numeric strings to int/float
|
|
- Preserves other values as-is
|
|
"""
|
|
if isinstance(value, str):
|
|
# Boolean conversion
|
|
if value.lower() == "true":
|
|
return True
|
|
if value.lower() == "false":
|
|
return False
|
|
|
|
# Integer conversion (only for pure numeric strings)
|
|
if value.lstrip("-").isdigit():
|
|
try:
|
|
return int(value)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Float conversion
|
|
try:
|
|
if "." in value and value.replace(".", "", 1).lstrip("-").isdigit():
|
|
return float(value)
|
|
except ValueError:
|
|
pass
|
|
|
|
return value
|
|
|
|
# =====================
|
|
# DATABASES
|
|
# =====================
|
|
|
|
async def list_databases(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List all databases."""
|
|
params = {}
|
|
if queries:
|
|
params["queries[]"] = queries
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/databases", params=params)
|
|
|
|
async def get_database(self, database_id: str) -> dict[str, Any]:
|
|
"""Get database by ID."""
|
|
return await self.request("GET", f"/databases/{database_id}")
|
|
|
|
async def create_database(
|
|
self, database_id: str, name: str, enabled: bool = True
|
|
) -> dict[str, Any]:
|
|
"""Create a new database."""
|
|
data = {"databaseId": database_id, "name": name, "enabled": enabled}
|
|
return await self.request("POST", "/databases", json_data=data)
|
|
|
|
async def update_database(
|
|
self, database_id: str, name: str, enabled: bool | None = None
|
|
) -> dict[str, Any]:
|
|
"""Update database."""
|
|
data = {"name": name}
|
|
if enabled is not None:
|
|
data["enabled"] = enabled
|
|
return await self.request("PUT", f"/databases/{database_id}", json_data=data)
|
|
|
|
async def delete_database(self, database_id: str) -> dict[str, Any]:
|
|
"""Delete database."""
|
|
return await self.request("DELETE", f"/databases/{database_id}")
|
|
|
|
# =====================
|
|
# COLLECTIONS
|
|
# =====================
|
|
|
|
async def list_collections(
|
|
self, database_id: str, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List collections in a database."""
|
|
params = {}
|
|
if queries:
|
|
params["queries[]"] = queries
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", f"/databases/{database_id}/collections", params=params)
|
|
|
|
async def get_collection(self, database_id: str, collection_id: str) -> dict[str, Any]:
|
|
"""Get collection by ID."""
|
|
return await self.request("GET", f"/databases/{database_id}/collections/{collection_id}")
|
|
|
|
async def create_collection(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
name: str,
|
|
permissions: list[str] | None = None,
|
|
document_security: bool = True,
|
|
enabled: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Create a new collection."""
|
|
data = {
|
|
"collectionId": collection_id,
|
|
"name": name,
|
|
"documentSecurity": document_security,
|
|
"enabled": enabled,
|
|
}
|
|
if permissions:
|
|
data["permissions"] = permissions
|
|
|
|
return await self.request("POST", f"/databases/{database_id}/collections", json_data=data)
|
|
|
|
async def update_collection(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
name: str,
|
|
permissions: list[str] | None = None,
|
|
document_security: bool | None = None,
|
|
enabled: bool | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Update collection."""
|
|
data = {"name": name}
|
|
if permissions is not None:
|
|
data["permissions"] = permissions
|
|
if document_security is not None:
|
|
data["documentSecurity"] = document_security
|
|
if enabled is not None:
|
|
data["enabled"] = enabled
|
|
|
|
return await self.request(
|
|
"PUT", f"/databases/{database_id}/collections/{collection_id}", json_data=data
|
|
)
|
|
|
|
async def delete_collection(self, database_id: str, collection_id: str) -> dict[str, Any]:
|
|
"""Delete collection."""
|
|
return await self.request("DELETE", f"/databases/{database_id}/collections/{collection_id}")
|
|
|
|
# =====================
|
|
# ATTRIBUTES
|
|
# =====================
|
|
|
|
async def list_attributes(
|
|
self, database_id: str, collection_id: str, queries: list[str] | None = None
|
|
) -> dict[str, Any]:
|
|
"""List attributes of a collection."""
|
|
params = {}
|
|
if queries:
|
|
params["queries[]"] = queries
|
|
|
|
return await self.request(
|
|
"GET", f"/databases/{database_id}/collections/{collection_id}/attributes", params=params
|
|
)
|
|
|
|
async def create_string_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
size: int,
|
|
required: bool = False,
|
|
default: str | None = None,
|
|
array: bool = False,
|
|
encrypt: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a string attribute."""
|
|
data = {"key": key, "size": size, "required": required, "array": array, "encrypt": encrypt}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/string",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_integer_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
min: int | None = None,
|
|
max: int | None = None,
|
|
default: int | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create an integer attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if min is not None:
|
|
data["min"] = min
|
|
if max is not None:
|
|
data["max"] = max
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/integer",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_float_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
min: float | None = None,
|
|
max: float | None = None,
|
|
default: float | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a float attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if min is not None:
|
|
data["min"] = min
|
|
if max is not None:
|
|
data["max"] = max
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/float",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_boolean_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
default: bool | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a boolean attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/boolean",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_datetime_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
default: str | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a datetime attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/datetime",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_email_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
default: str | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create an email attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/email",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_url_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
default: str | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a URL attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/url",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_ip_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
required: bool = False,
|
|
default: str | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create an IP address attribute."""
|
|
data = {"key": key, "required": required, "array": array}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/ip",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_enum_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
elements: list[str],
|
|
required: bool = False,
|
|
default: str | None = None,
|
|
array: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create an enum attribute."""
|
|
data = {"key": key, "elements": elements, "required": required, "array": array}
|
|
if default is not None:
|
|
data["default"] = default
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/enum",
|
|
json_data=data,
|
|
)
|
|
|
|
async def create_relationship_attribute(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
related_collection_id: str,
|
|
type: str,
|
|
two_way: bool = False,
|
|
key: str | None = None,
|
|
two_way_key: str | None = None,
|
|
on_delete: str = "setNull",
|
|
) -> dict[str, Any]:
|
|
"""Create a relationship attribute."""
|
|
data = {
|
|
"relatedCollectionId": related_collection_id,
|
|
"type": type, # oneToOne, oneToMany, manyToOne, manyToMany
|
|
"twoWay": two_way,
|
|
"onDelete": on_delete, # setNull, cascade, restrict
|
|
}
|
|
if key:
|
|
data["key"] = key
|
|
if two_way_key:
|
|
data["twoWayKey"] = two_way_key
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/attributes/relationship",
|
|
json_data=data,
|
|
)
|
|
|
|
async def get_attribute(self, database_id: str, collection_id: str, key: str) -> dict[str, Any]:
|
|
"""Get attribute by key."""
|
|
return await self.request(
|
|
"GET", f"/databases/{database_id}/collections/{collection_id}/attributes/{key}"
|
|
)
|
|
|
|
async def delete_attribute(
|
|
self, database_id: str, collection_id: str, key: str
|
|
) -> dict[str, Any]:
|
|
"""Delete attribute."""
|
|
return await self.request(
|
|
"DELETE", f"/databases/{database_id}/collections/{collection_id}/attributes/{key}"
|
|
)
|
|
|
|
# =====================
|
|
# INDEXES
|
|
# =====================
|
|
|
|
async def list_indexes(
|
|
self, database_id: str, collection_id: str, queries: list[str] | None = None
|
|
) -> dict[str, Any]:
|
|
"""List indexes of a collection."""
|
|
params = {}
|
|
if queries:
|
|
params["queries[]"] = queries
|
|
|
|
return await self.request(
|
|
"GET", f"/databases/{database_id}/collections/{collection_id}/indexes", params=params
|
|
)
|
|
|
|
async def create_index(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
key: str,
|
|
type: str,
|
|
attributes: list[str],
|
|
orders: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create an index."""
|
|
data = {"key": key, "type": type, "attributes": attributes} # key, unique, fulltext
|
|
if orders:
|
|
data["orders"] = orders
|
|
|
|
return await self.request(
|
|
"POST", f"/databases/{database_id}/collections/{collection_id}/indexes", json_data=data
|
|
)
|
|
|
|
async def get_index(self, database_id: str, collection_id: str, key: str) -> dict[str, Any]:
|
|
"""Get index by key."""
|
|
return await self.request(
|
|
"GET", f"/databases/{database_id}/collections/{collection_id}/indexes/{key}"
|
|
)
|
|
|
|
async def delete_index(self, database_id: str, collection_id: str, key: str) -> dict[str, Any]:
|
|
"""Delete index."""
|
|
return await self.request(
|
|
"DELETE", f"/databases/{database_id}/collections/{collection_id}/indexes/{key}"
|
|
)
|
|
|
|
# =====================
|
|
# DOCUMENTS
|
|
# =====================
|
|
|
|
async def list_documents(
|
|
self, database_id: str, collection_id: str, queries: list[str] | None = None
|
|
) -> dict[str, Any]:
|
|
"""List documents in a collection."""
|
|
params = {}
|
|
if queries:
|
|
# Send each query as separate queries[] param - Appwrite REST API format
|
|
params["queries[]"] = queries
|
|
|
|
return await self.request(
|
|
"GET", f"/databases/{database_id}/collections/{collection_id}/documents", params=params
|
|
)
|
|
|
|
async def get_document(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
document_id: str,
|
|
queries: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Get document by ID."""
|
|
params = {}
|
|
if queries:
|
|
params["queries[]"] = queries
|
|
|
|
return await self.request(
|
|
"GET",
|
|
f"/databases/{database_id}/collections/{collection_id}/documents/{document_id}",
|
|
params=params,
|
|
)
|
|
|
|
async def create_document(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
document_id: str,
|
|
data: dict[str, Any],
|
|
permissions: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new document."""
|
|
body = {"documentId": document_id, "data": data}
|
|
if permissions:
|
|
body["permissions"] = permissions
|
|
|
|
return await self.request(
|
|
"POST",
|
|
f"/databases/{database_id}/collections/{collection_id}/documents",
|
|
json_data=body,
|
|
)
|
|
|
|
async def update_document(
|
|
self,
|
|
database_id: str,
|
|
collection_id: str,
|
|
document_id: str,
|
|
data: dict[str, Any] | None = None,
|
|
permissions: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Update document."""
|
|
body = {}
|
|
if data:
|
|
body["data"] = data
|
|
if permissions is not None:
|
|
body["permissions"] = permissions
|
|
|
|
return await self.request(
|
|
"PATCH",
|
|
f"/databases/{database_id}/collections/{collection_id}/documents/{document_id}",
|
|
json_data=body,
|
|
)
|
|
|
|
async def delete_document(
|
|
self, database_id: str, collection_id: str, document_id: str
|
|
) -> dict[str, Any]:
|
|
"""Delete document."""
|
|
return await self.request(
|
|
"DELETE",
|
|
f"/databases/{database_id}/collections/{collection_id}/documents/{document_id}",
|
|
)
|
|
|
|
# =====================
|
|
# USERS
|
|
# =====================
|
|
|
|
async def list_users(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List users."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/users", params=params)
|
|
|
|
async def get_user(self, user_id: str) -> dict[str, Any]:
|
|
"""Get user by ID."""
|
|
return await self.request("GET", f"/users/{user_id}")
|
|
|
|
async def create_user(
|
|
self,
|
|
user_id: str,
|
|
email: str | None = None,
|
|
phone: str | None = None,
|
|
password: str | None = None,
|
|
name: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new user."""
|
|
data = {"userId": user_id}
|
|
if email:
|
|
data["email"] = email
|
|
if phone:
|
|
data["phone"] = phone
|
|
if password:
|
|
data["password"] = password
|
|
if name:
|
|
data["name"] = name
|
|
|
|
return await self.request("POST", "/users", json_data=data)
|
|
|
|
async def update_user_name(self, user_id: str, name: str) -> dict[str, Any]:
|
|
"""Update user name."""
|
|
return await self.request("PATCH", f"/users/{user_id}/name", json_data={"name": name})
|
|
|
|
async def update_user_email(self, user_id: str, email: str) -> dict[str, Any]:
|
|
"""Update user email."""
|
|
return await self.request("PATCH", f"/users/{user_id}/email", json_data={"email": email})
|
|
|
|
async def update_user_phone(self, user_id: str, number: str) -> dict[str, Any]:
|
|
"""Update user phone."""
|
|
return await self.request("PATCH", f"/users/{user_id}/phone", json_data={"number": number})
|
|
|
|
async def update_user_password(self, user_id: str, password: str) -> dict[str, Any]:
|
|
"""Update user password."""
|
|
return await self.request(
|
|
"PATCH", f"/users/{user_id}/password", json_data={"password": password}
|
|
)
|
|
|
|
async def update_user_status(self, user_id: str, status: bool) -> dict[str, Any]:
|
|
"""Update user status (enable/disable)."""
|
|
return await self.request("PATCH", f"/users/{user_id}/status", json_data={"status": status})
|
|
|
|
async def update_user_labels(self, user_id: str, labels: list[str]) -> dict[str, Any]:
|
|
"""Update user labels."""
|
|
return await self.request("PUT", f"/users/{user_id}/labels", json_data={"labels": labels})
|
|
|
|
async def update_user_prefs(self, user_id: str, prefs: dict[str, Any]) -> dict[str, Any]:
|
|
"""Update user preferences."""
|
|
return await self.request("PATCH", f"/users/{user_id}/prefs", json_data={"prefs": prefs})
|
|
|
|
async def delete_user(self, user_id: str) -> dict[str, Any]:
|
|
"""Delete user."""
|
|
return await self.request("DELETE", f"/users/{user_id}")
|
|
|
|
async def list_user_sessions(self, user_id: str) -> dict[str, Any]:
|
|
"""List user sessions."""
|
|
return await self.request("GET", f"/users/{user_id}/sessions")
|
|
|
|
async def delete_user_sessions(self, user_id: str) -> dict[str, Any]:
|
|
"""Delete all user sessions."""
|
|
return await self.request("DELETE", f"/users/{user_id}/sessions")
|
|
|
|
async def delete_user_session(self, user_id: str, session_id: str) -> dict[str, Any]:
|
|
"""Delete a specific user session."""
|
|
return await self.request("DELETE", f"/users/{user_id}/sessions/{session_id}")
|
|
|
|
# =====================
|
|
# TEAMS
|
|
# =====================
|
|
|
|
async def list_teams(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List teams."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/teams", params=params)
|
|
|
|
async def get_team(self, team_id: str) -> dict[str, Any]:
|
|
"""Get team by ID."""
|
|
return await self.request("GET", f"/teams/{team_id}")
|
|
|
|
async def create_team(
|
|
self, team_id: str, name: str, roles: list[str] | None = None
|
|
) -> dict[str, Any]:
|
|
"""Create a new team."""
|
|
data = {"teamId": team_id, "name": name}
|
|
if roles:
|
|
data["roles"] = roles
|
|
|
|
return await self.request("POST", "/teams", json_data=data)
|
|
|
|
async def update_team(self, team_id: str, name: str) -> dict[str, Any]:
|
|
"""Update team name."""
|
|
return await self.request("PUT", f"/teams/{team_id}", json_data={"name": name})
|
|
|
|
async def delete_team(self, team_id: str) -> dict[str, Any]:
|
|
"""Delete team."""
|
|
return await self.request("DELETE", f"/teams/{team_id}")
|
|
|
|
async def list_team_memberships(
|
|
self, team_id: str, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List team memberships."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", f"/teams/{team_id}/memberships", params=params)
|
|
|
|
async def create_team_membership(
|
|
self,
|
|
team_id: str,
|
|
roles: list[str],
|
|
email: str | None = None,
|
|
user_id: str | None = None,
|
|
phone: str | None = None,
|
|
url: str | None = None,
|
|
name: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create team membership (invite)."""
|
|
data = {"roles": roles}
|
|
if email:
|
|
data["email"] = email
|
|
if user_id:
|
|
data["userId"] = user_id
|
|
if phone:
|
|
data["phone"] = phone
|
|
if url:
|
|
data["url"] = url
|
|
if name:
|
|
data["name"] = name
|
|
|
|
return await self.request("POST", f"/teams/{team_id}/memberships", json_data=data)
|
|
|
|
async def update_membership(
|
|
self, team_id: str, membership_id: str, roles: list[str]
|
|
) -> dict[str, Any]:
|
|
"""Update membership roles."""
|
|
return await self.request(
|
|
"PATCH", f"/teams/{team_id}/memberships/{membership_id}", json_data={"roles": roles}
|
|
)
|
|
|
|
async def delete_membership(self, team_id: str, membership_id: str) -> dict[str, Any]:
|
|
"""Delete team membership."""
|
|
return await self.request("DELETE", f"/teams/{team_id}/memberships/{membership_id}")
|
|
|
|
# =====================
|
|
# STORAGE
|
|
# =====================
|
|
|
|
async def list_buckets(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List storage buckets."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/storage/buckets", params=params)
|
|
|
|
async def get_bucket(self, bucket_id: str) -> dict[str, Any]:
|
|
"""Get bucket by ID."""
|
|
return await self.request("GET", f"/storage/buckets/{bucket_id}")
|
|
|
|
async def create_bucket(
|
|
self,
|
|
bucket_id: str,
|
|
name: str,
|
|
permissions: list[str] | None = None,
|
|
file_security: bool = True,
|
|
enabled: bool = True,
|
|
maximum_file_size: int | None = None,
|
|
allowed_file_extensions: list[str] | None = None,
|
|
compression: str = "none",
|
|
encryption: bool = True,
|
|
antivirus: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Create a new bucket."""
|
|
data = {
|
|
"bucketId": bucket_id,
|
|
"name": name,
|
|
"fileSecurity": file_security,
|
|
"enabled": enabled,
|
|
"compression": compression,
|
|
"encryption": encryption,
|
|
"antivirus": antivirus,
|
|
}
|
|
if permissions:
|
|
data["permissions"] = permissions
|
|
if maximum_file_size:
|
|
data["maximumFileSize"] = maximum_file_size
|
|
if allowed_file_extensions:
|
|
data["allowedFileExtensions"] = allowed_file_extensions
|
|
|
|
return await self.request("POST", "/storage/buckets", json_data=data)
|
|
|
|
async def update_bucket(
|
|
self,
|
|
bucket_id: str,
|
|
name: str,
|
|
permissions: list[str] | None = None,
|
|
file_security: bool | None = None,
|
|
enabled: bool | None = None,
|
|
maximum_file_size: int | None = None,
|
|
allowed_file_extensions: list[str] | None = None,
|
|
compression: str | None = None,
|
|
encryption: bool | None = None,
|
|
antivirus: bool | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Update bucket."""
|
|
data = {"name": name}
|
|
if permissions is not None:
|
|
data["permissions"] = permissions
|
|
if file_security is not None:
|
|
data["fileSecurity"] = file_security
|
|
if enabled is not None:
|
|
data["enabled"] = enabled
|
|
if maximum_file_size is not None:
|
|
data["maximumFileSize"] = maximum_file_size
|
|
if allowed_file_extensions is not None:
|
|
data["allowedFileExtensions"] = allowed_file_extensions
|
|
if compression is not None:
|
|
data["compression"] = compression
|
|
if encryption is not None:
|
|
data["encryption"] = encryption
|
|
if antivirus is not None:
|
|
data["antivirus"] = antivirus
|
|
|
|
return await self.request("PUT", f"/storage/buckets/{bucket_id}", json_data=data)
|
|
|
|
async def delete_bucket(self, bucket_id: str) -> dict[str, Any]:
|
|
"""Delete bucket."""
|
|
return await self.request("DELETE", f"/storage/buckets/{bucket_id}")
|
|
|
|
async def list_files(
|
|
self, bucket_id: str, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List files in a bucket."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", f"/storage/buckets/{bucket_id}/files", params=params)
|
|
|
|
async def get_file(self, bucket_id: str, file_id: str) -> dict[str, Any]:
|
|
"""Get file metadata."""
|
|
return await self.request("GET", f"/storage/buckets/{bucket_id}/files/{file_id}")
|
|
|
|
async def delete_file(self, bucket_id: str, file_id: str) -> dict[str, Any]:
|
|
"""Delete file."""
|
|
return await self.request("DELETE", f"/storage/buckets/{bucket_id}/files/{file_id}")
|
|
|
|
async def get_file_download(self, bucket_id: str, file_id: str) -> dict[str, Any]:
|
|
"""Download file (returns base64 content)."""
|
|
return await self.request("GET", f"/storage/buckets/{bucket_id}/files/{file_id}/download")
|
|
|
|
async def get_file_preview(
|
|
self,
|
|
bucket_id: str,
|
|
file_id: str,
|
|
width: int | None = None,
|
|
height: int | None = None,
|
|
gravity: str | None = None,
|
|
quality: int | None = None,
|
|
border_width: int | None = None,
|
|
border_color: str | None = None,
|
|
border_radius: int | None = None,
|
|
opacity: float | None = None,
|
|
rotation: int | None = None,
|
|
background: str | None = None,
|
|
output: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Get file preview (image transformation)."""
|
|
params = {}
|
|
if width:
|
|
params["width"] = width
|
|
if height:
|
|
params["height"] = height
|
|
if gravity:
|
|
params["gravity"] = gravity
|
|
if quality:
|
|
params["quality"] = quality
|
|
if border_width:
|
|
params["borderWidth"] = border_width
|
|
if border_color:
|
|
params["borderColor"] = border_color
|
|
if border_radius:
|
|
params["borderRadius"] = border_radius
|
|
if opacity:
|
|
params["opacity"] = opacity
|
|
if rotation:
|
|
params["rotation"] = rotation
|
|
if background:
|
|
params["background"] = background
|
|
if output:
|
|
params["output"] = output
|
|
|
|
return await self.request(
|
|
"GET", f"/storage/buckets/{bucket_id}/files/{file_id}/preview", params=params
|
|
)
|
|
|
|
async def get_file_view(self, bucket_id: str, file_id: str) -> dict[str, Any]:
|
|
"""Get file for viewing in browser."""
|
|
return await self.request("GET", f"/storage/buckets/{bucket_id}/files/{file_id}/view")
|
|
|
|
# =====================
|
|
# FUNCTIONS
|
|
# =====================
|
|
|
|
async def list_functions(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List functions."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/functions", params=params)
|
|
|
|
async def get_function(self, function_id: str) -> dict[str, Any]:
|
|
"""Get function by ID."""
|
|
return await self.request("GET", f"/functions/{function_id}")
|
|
|
|
async def create_function(
|
|
self,
|
|
function_id: str,
|
|
name: str,
|
|
runtime: str,
|
|
execute: list[str] | None = None,
|
|
events: list[str] | None = None,
|
|
schedule: str | None = None,
|
|
timeout: int = 15,
|
|
enabled: bool = True,
|
|
logging: bool = True,
|
|
entrypoint: str | None = None,
|
|
commands: str | None = None,
|
|
scopes: list[str] | None = None,
|
|
installation_id: str | None = None,
|
|
provider_repository_id: str | None = None,
|
|
provider_branch: str | None = None,
|
|
provider_silent_mode: bool = False,
|
|
provider_root_directory: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new function."""
|
|
data = {
|
|
"functionId": function_id,
|
|
"name": name,
|
|
"runtime": runtime,
|
|
"timeout": timeout,
|
|
"enabled": enabled,
|
|
"logging": logging,
|
|
"providerSilentMode": provider_silent_mode,
|
|
}
|
|
if execute:
|
|
data["execute"] = execute
|
|
if events:
|
|
data["events"] = events
|
|
if schedule:
|
|
data["schedule"] = schedule
|
|
if entrypoint:
|
|
data["entrypoint"] = entrypoint
|
|
if commands:
|
|
data["commands"] = commands
|
|
if scopes:
|
|
data["scopes"] = scopes
|
|
if installation_id:
|
|
data["installationId"] = installation_id
|
|
if provider_repository_id:
|
|
data["providerRepositoryId"] = provider_repository_id
|
|
if provider_branch:
|
|
data["providerBranch"] = provider_branch
|
|
if provider_root_directory:
|
|
data["providerRootDirectory"] = provider_root_directory
|
|
|
|
return await self.request("POST", "/functions", json_data=data)
|
|
|
|
async def update_function(
|
|
self,
|
|
function_id: str,
|
|
name: str,
|
|
runtime: str | None = None,
|
|
execute: list[str] | None = None,
|
|
events: list[str] | None = None,
|
|
schedule: str | None = None,
|
|
timeout: int | None = None,
|
|
enabled: bool | None = None,
|
|
logging: bool | None = None,
|
|
entrypoint: str | None = None,
|
|
commands: str | None = None,
|
|
scopes: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Update function."""
|
|
data = {"name": name}
|
|
if runtime:
|
|
data["runtime"] = runtime
|
|
if execute is not None:
|
|
data["execute"] = execute
|
|
if events is not None:
|
|
data["events"] = events
|
|
if schedule is not None:
|
|
data["schedule"] = schedule
|
|
if timeout is not None:
|
|
data["timeout"] = timeout
|
|
if enabled is not None:
|
|
data["enabled"] = enabled
|
|
if logging is not None:
|
|
data["logging"] = logging
|
|
if entrypoint:
|
|
data["entrypoint"] = entrypoint
|
|
if commands:
|
|
data["commands"] = commands
|
|
if scopes is not None:
|
|
data["scopes"] = scopes
|
|
|
|
return await self.request("PUT", f"/functions/{function_id}", json_data=data)
|
|
|
|
async def delete_function(self, function_id: str) -> dict[str, Any]:
|
|
"""Delete function."""
|
|
return await self.request("DELETE", f"/functions/{function_id}")
|
|
|
|
async def list_deployments(
|
|
self, function_id: str, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List function deployments."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", f"/functions/{function_id}/deployments", params=params)
|
|
|
|
async def get_deployment(self, function_id: str, deployment_id: str) -> dict[str, Any]:
|
|
"""Get deployment by ID."""
|
|
return await self.request("GET", f"/functions/{function_id}/deployments/{deployment_id}")
|
|
|
|
async def delete_deployment(self, function_id: str, deployment_id: str) -> dict[str, Any]:
|
|
"""Delete deployment."""
|
|
return await self.request("DELETE", f"/functions/{function_id}/deployments/{deployment_id}")
|
|
|
|
async def update_deployment(self, function_id: str, deployment_id: str) -> dict[str, Any]:
|
|
"""Activate deployment (set as active)."""
|
|
return await self.request("PATCH", f"/functions/{function_id}/deployments/{deployment_id}")
|
|
|
|
async def list_executions(
|
|
self, function_id: str, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List function executions."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", f"/functions/{function_id}/executions", params=params)
|
|
|
|
async def get_execution(self, function_id: str, execution_id: str) -> dict[str, Any]:
|
|
"""Get execution by ID."""
|
|
return await self.request("GET", f"/functions/{function_id}/executions/{execution_id}")
|
|
|
|
async def create_execution(
|
|
self,
|
|
function_id: str,
|
|
body: str | None = None,
|
|
async_execution: bool = False,
|
|
path: str | None = None,
|
|
method: str = "POST",
|
|
headers: dict[str, str] | None = None,
|
|
scheduled_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Execute a function."""
|
|
data = {"async": async_execution, "method": method}
|
|
if body:
|
|
data["body"] = body
|
|
if path:
|
|
data["path"] = path
|
|
if headers:
|
|
data["headers"] = headers
|
|
if scheduled_at:
|
|
data["scheduledAt"] = scheduled_at
|
|
|
|
return await self.request("POST", f"/functions/{function_id}/executions", json_data=data)
|
|
|
|
async def delete_execution(self, function_id: str, execution_id: str) -> dict[str, Any]:
|
|
"""Delete execution."""
|
|
return await self.request("DELETE", f"/functions/{function_id}/executions/{execution_id}")
|
|
|
|
# =====================
|
|
# HEALTH
|
|
# =====================
|
|
|
|
async def health(self) -> dict[str, Any]:
|
|
"""Get general health status."""
|
|
return await self.request("GET", "/health")
|
|
|
|
async def health_db(self) -> dict[str, Any]:
|
|
"""Get database health status."""
|
|
return await self.request("GET", "/health/db")
|
|
|
|
async def health_cache(self) -> dict[str, Any]:
|
|
"""Get cache health status."""
|
|
return await self.request("GET", "/health/cache")
|
|
|
|
async def health_pubsub(self) -> dict[str, Any]:
|
|
"""Get pub/sub health status."""
|
|
return await self.request("GET", "/health/pubsub")
|
|
|
|
async def health_queue(self) -> dict[str, Any]:
|
|
"""Get queue health status."""
|
|
return await self.request("GET", "/health/queue")
|
|
|
|
async def health_storage_local(self) -> dict[str, Any]:
|
|
"""Get local storage health status."""
|
|
return await self.request("GET", "/health/storage/local")
|
|
|
|
async def health_time(self) -> dict[str, Any]:
|
|
"""Get time sync health status."""
|
|
return await self.request("GET", "/health/time")
|
|
|
|
async def health_certificate(self, domain: str | None = None) -> dict[str, Any]:
|
|
"""Get SSL certificate health status."""
|
|
params = {}
|
|
if domain:
|
|
params["domain"] = domain
|
|
return await self.request("GET", "/health/certificate", params=params)
|
|
|
|
# =====================
|
|
# AVATARS
|
|
# =====================
|
|
|
|
async def get_avatar_initials(
|
|
self,
|
|
name: str | None = None,
|
|
width: int = 100,
|
|
height: int = 100,
|
|
background: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Get avatar image from initials."""
|
|
params = {"width": width, "height": height}
|
|
if name:
|
|
params["name"] = name
|
|
if background:
|
|
params["background"] = background
|
|
|
|
return await self.request("GET", "/avatars/initials", params=params)
|
|
|
|
async def get_avatar_image(
|
|
self, url: str, width: int = 400, height: int = 400
|
|
) -> dict[str, Any]:
|
|
"""Get avatar from URL."""
|
|
params = {"url": url, "width": width, "height": height}
|
|
return await self.request("GET", "/avatars/image", params=params)
|
|
|
|
async def get_favicon(self, url: str) -> dict[str, Any]:
|
|
"""Get website favicon."""
|
|
return await self.request("GET", "/avatars/favicon", params={"url": url})
|
|
|
|
async def get_qr_code(
|
|
self, text: str, size: int = 400, margin: int = 1, download: bool = False
|
|
) -> dict[str, Any]:
|
|
"""Generate QR code."""
|
|
params = {"text": text, "size": size, "margin": margin, "download": download}
|
|
return await self.request("GET", "/avatars/qr", params=params)
|
|
|
|
# =====================
|
|
# MESSAGING (Server SDK)
|
|
# =====================
|
|
|
|
async def list_providers(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List messaging providers."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/messaging/providers", params=params)
|
|
|
|
async def list_topics(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List messaging topics."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/messaging/topics", params=params)
|
|
|
|
async def get_topic(self, topic_id: str) -> dict[str, Any]:
|
|
"""Get topic by ID."""
|
|
return await self.request("GET", f"/messaging/topics/{topic_id}")
|
|
|
|
async def create_topic(
|
|
self, topic_id: str, name: str, subscribe: list[str] | None = None
|
|
) -> dict[str, Any]:
|
|
"""Create a messaging topic."""
|
|
data = {"topicId": topic_id, "name": name}
|
|
if subscribe:
|
|
data["subscribe"] = subscribe
|
|
|
|
return await self.request("POST", "/messaging/topics", json_data=data)
|
|
|
|
async def update_topic(
|
|
self, topic_id: str, name: str | None = None, subscribe: list[str] | None = None
|
|
) -> dict[str, Any]:
|
|
"""Update topic."""
|
|
data = {}
|
|
if name:
|
|
data["name"] = name
|
|
if subscribe is not None:
|
|
data["subscribe"] = subscribe
|
|
|
|
return await self.request("PATCH", f"/messaging/topics/{topic_id}", json_data=data)
|
|
|
|
async def delete_topic(self, topic_id: str) -> dict[str, Any]:
|
|
"""Delete topic."""
|
|
return await self.request("DELETE", f"/messaging/topics/{topic_id}")
|
|
|
|
async def create_subscriber(
|
|
self, topic_id: str, subscriber_id: str, target_id: str
|
|
) -> dict[str, Any]:
|
|
"""Add subscriber to topic."""
|
|
data = {"subscriberId": subscriber_id, "targetId": target_id}
|
|
return await self.request(
|
|
"POST", f"/messaging/topics/{topic_id}/subscribers", json_data=data
|
|
)
|
|
|
|
async def delete_subscriber(self, topic_id: str, subscriber_id: str) -> dict[str, Any]:
|
|
"""Remove subscriber from topic."""
|
|
return await self.request(
|
|
"DELETE", f"/messaging/topics/{topic_id}/subscribers/{subscriber_id}"
|
|
)
|
|
|
|
async def list_messages(
|
|
self, queries: list[str] | None = None, search: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""List messages."""
|
|
params = {}
|
|
if queries:
|
|
for i, q in enumerate(queries):
|
|
params[f"queries[{i}]"] = q
|
|
if search:
|
|
params["search"] = search
|
|
|
|
return await self.request("GET", "/messaging/messages", params=params)
|
|
|
|
async def get_message(self, message_id: str) -> dict[str, Any]:
|
|
"""Get message by ID."""
|
|
return await self.request("GET", f"/messaging/messages/{message_id}")
|
|
|
|
async def create_email(
|
|
self,
|
|
message_id: str,
|
|
subject: str,
|
|
content: str,
|
|
topics: list[str] | None = None,
|
|
users: list[str] | None = None,
|
|
targets: list[str] | None = None,
|
|
cc: list[str] | None = None,
|
|
bcc: list[str] | None = None,
|
|
attachments: list[str] | None = None,
|
|
draft: bool = False,
|
|
html: bool = True,
|
|
scheduled_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create/send email message."""
|
|
data = {
|
|
"messageId": message_id,
|
|
"subject": subject,
|
|
"content": content,
|
|
"draft": draft,
|
|
"html": html,
|
|
}
|
|
if topics:
|
|
data["topics"] = topics
|
|
if users:
|
|
data["users"] = users
|
|
if targets:
|
|
data["targets"] = targets
|
|
if cc:
|
|
data["cc"] = cc
|
|
if bcc:
|
|
data["bcc"] = bcc
|
|
if attachments:
|
|
data["attachments"] = attachments
|
|
if scheduled_at:
|
|
data["scheduledAt"] = scheduled_at
|
|
|
|
return await self.request("POST", "/messaging/messages/email", json_data=data)
|
|
|
|
async def create_sms(
|
|
self,
|
|
message_id: str,
|
|
content: str,
|
|
topics: list[str] | None = None,
|
|
users: list[str] | None = None,
|
|
targets: list[str] | None = None,
|
|
draft: bool = False,
|
|
scheduled_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create/send SMS message."""
|
|
data = {"messageId": message_id, "content": content, "draft": draft}
|
|
if topics:
|
|
data["topics"] = topics
|
|
if users:
|
|
data["users"] = users
|
|
if targets:
|
|
data["targets"] = targets
|
|
if scheduled_at:
|
|
data["scheduledAt"] = scheduled_at
|
|
|
|
return await self.request("POST", "/messaging/messages/sms", json_data=data)
|
|
|
|
async def create_push(
|
|
self,
|
|
message_id: str,
|
|
title: str,
|
|
body: str,
|
|
topics: list[str] | None = None,
|
|
users: list[str] | None = None,
|
|
targets: list[str] | None = None,
|
|
data_payload: dict[str, Any] | None = None,
|
|
action: str | None = None,
|
|
image: str | None = None,
|
|
icon: str | None = None,
|
|
sound: str | None = None,
|
|
color: str | None = None,
|
|
tag: str | None = None,
|
|
badge: int | None = None,
|
|
draft: bool = False,
|
|
scheduled_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create/send push notification."""
|
|
data = {"messageId": message_id, "title": title, "body": body, "draft": draft}
|
|
if topics:
|
|
data["topics"] = topics
|
|
if users:
|
|
data["users"] = users
|
|
if targets:
|
|
data["targets"] = targets
|
|
if data_payload:
|
|
data["data"] = data_payload
|
|
if action:
|
|
data["action"] = action
|
|
if image:
|
|
data["image"] = image
|
|
if icon:
|
|
data["icon"] = icon
|
|
if sound:
|
|
data["sound"] = sound
|
|
if color:
|
|
data["color"] = color
|
|
if tag:
|
|
data["tag"] = tag
|
|
if badge is not None:
|
|
data["badge"] = badge
|
|
if scheduled_at:
|
|
data["scheduledAt"] = scheduled_at
|
|
|
|
return await self.request("POST", "/messaging/messages/push", json_data=data)
|
|
|
|
async def delete_message(self, message_id: str) -> dict[str, Any]:
|
|
"""Delete message."""
|
|
return await self.request("DELETE", f"/messaging/messages/{message_id}")
|
|
|
|
# =====================
|
|
# COMPREHENSIVE HEALTH CHECK
|
|
# =====================
|
|
|
|
async def health_check(self) -> dict[str, Any]:
|
|
"""
|
|
Check Appwrite instance health.
|
|
|
|
Returns comprehensive health status of all services.
|
|
"""
|
|
results = {"healthy": True, "services": {}}
|
|
|
|
# Check general health
|
|
try:
|
|
health = await self.health()
|
|
results["services"]["general"] = health.get("status", "ok")
|
|
except Exception as e:
|
|
results["services"]["general"] = f"error: {str(e)}"
|
|
results["healthy"] = False
|
|
|
|
# Check database
|
|
try:
|
|
db_health = await self.health_db()
|
|
results["services"]["database"] = db_health.get("status", "ok")
|
|
except Exception as e:
|
|
results["services"]["database"] = f"error: {str(e)}"
|
|
results["healthy"] = False
|
|
|
|
# Check cache
|
|
try:
|
|
cache_health = await self.health_cache()
|
|
results["services"]["cache"] = cache_health.get("status", "ok")
|
|
except Exception as e:
|
|
results["services"]["cache"] = f"error: {str(e)}"
|
|
# Cache error might not be critical
|
|
if "error" not in str(e).lower():
|
|
results["services"]["cache"] = "warning"
|
|
|
|
# Check storage
|
|
try:
|
|
storage_health = await self.health_storage_local()
|
|
results["services"]["storage"] = storage_health.get("status", "ok")
|
|
except Exception as e:
|
|
results["services"]["storage"] = f"error: {str(e)}"
|
|
results["healthy"] = False
|
|
|
|
return results
|