Appwrite Python

Appwrite Python SDK

Installation

pip install appwrite

Setting Up the Client

from appwrite.client import Client
from appwrite.id import ID
from appwrite.query import Query
from appwrite.services.users import Users
from appwrite.services.tables_db import TablesDB
from appwrite.services.storage import Storage
from appwrite.services.functions import Functions
from appwrite.enums.o_auth_provider import OAuthProvider

import os

client = (Client()
    .set_endpoint('https://.cloud.appwrite.io/v1')
    .set_project(os.environ['APPWRITE_PROJECT_ID'])
    .set_key(os.environ['APPWRITE_API_KEY']))

Code Examples

User Management

users = Users(client)

# Create user
user = users.create(ID.unique(), 'user@example.com', None, 'password123', 'User Name')

# List users
result = users.list([Query.limit(25)])

# Get user
fetched = users.get('[USER_ID]')

# Delete user
users.delete('[USER_ID]')

Database Operations

> Note: Use TablesDB (not the deprecated Databases class) for all new code. Only use Databases if the existing codebase already relies on it or the user explicitly requests it. > > Tip: Prefer keyword arguments (e.g., database_id='...') over positional arguments for all SDK method calls. Only use positional style if the existing codebase already uses it or the user explicitly requests it.

tables_db = TablesDB(client)

# Create database
db = tables_db.create(ID.unique(), 'My Database')

# Create row
doc = tables_db.create_row('[DATABASE_ID]', '[TABLE_ID]', ID.unique(), {
    'title': 'Hello World'
})

# Query rows
results = tables_db.list_rows('[DATABASE_ID]', '[TABLE_ID]', [
    Query.equal('title', 'Hello World'),
    Query.limit(10)
])

# Get row
row = tables_db.get_row('[DATABASE_ID]', '[TABLE_ID]', '[ROW_ID]')

# Update row
tables_db.update_row('[DATABASE_ID]', '[TABLE_ID]', '[ROW_ID]', {
    'title': 'Updated'
})

# Delete row
tables_db.delete_row('[DATABASE_ID]', '[TABLE_ID]', '[ROW_ID]')

String Column Types

> Note: The legacy string type is deprecated. Use explicit column types for all new columns.

| Type | Max characters | Indexing | Storage | |------|---------------|----------|---------| | varchar | 16,383 | Full index (if size ≤ 768) | Inline in row | | text | 16,383 | Prefix only | Off-page | | mediumtext | 4,194,303 | Prefix only | Off-page | | longtext | 1,073,741,823 | Prefix only | Off-page |

• varchar is stored inline and counts towards the 64 KB row size limit. Prefer for short, indexed fields like names, slugs, or identifiers.
• text, mediumtext, and longtext are stored off-page (only a 20-byte pointer lives in the row), so they don't consume the row size budget. size is not required for these types.

# Create table with explicit string column types
tables_db.create_table(
    database_id='[DATABASE_ID]',
    table_id=ID.unique(),
    name='articles',
    columns=[
        {'key': 'title',    'type': 'varchar',    'size': 255, 'required': True},   # inline, fully indexable
        {'key': 'summary',  'type': 'text',                    'required': False},  # off-page, prefix index only
        {'key': 'body',     'type': 'mediumtext',              'required': False},  # up to ~4 M chars
        {'key': 'raw_data', 'type': 'longtext',                'required': False},  # up to ~1 B chars
    ]
)

Query Methods

# Filtering
Query.equal('field', 'value')             # == (or pass list for IN)
Query.not_equal('field', 'value')         # !=
Query.less_than('field', 100)             # 
Query.greater_than_equal('field', 100)    # >=
Query.between('field', 1, 100)            # 1  **Role-based access:** Use `Role.team('[TEAM_ID]')` for all team members or `Role.team('[TEAM_ID]', 'editor')` for a specific team role when setting permissions.

### Serverless Functions

```python
functions = Functions(client)

# Execute function
execution = functions.create_execution('[FUNCTION_ID]', body='{"key": "value"}')

# List executions
executions = functions.list_executions('[FUNCTION_ID]')

Writing a Function Handler (Python runtime)

# src/main.py — Appwrite Function entry point
def main(context):
    # context.req — request object
    #   .body        — raw request body (string)
    #   .body_json   — parsed JSON body (dict, or None if not JSON)
    #   .headers     — request headers (dict)
    #   .method      — HTTP method (GET, POST, etc.)
    #   .path        — URL path
    #   .query       — parsed query parameters (dict)
    #   .query_string — raw query string

    context.log('Processing: ' + context.req.method + ' ' + context.req.path)

    if context.req.method == 'GET':
        return context.res.json({'message': 'Hello from Appwrite Function!'})

    data = context.req.body_json or {}
    if 'name' not in data:
        context.error('Missing name field')
        return context.res.json({'error': 'Name is required'}, 400)

    # Response methods
    return context.res.json({'success': True})                    # JSON response
    # return context.res.text('Hello')                           # plain text
    # return context.res.empty()                                 # 204 No Content
    # return context.res.redirect('https://example.com')         # 302 Redirect
    # return context.res.send('data', 200, {'X-Custom': '1'})   # custom response

Server-Side Rendering (SSR) Authentication

SSR apps (Flask, Django, FastAPI, etc.) use the server SDK to handle auth. You need two clients:

• Admin client — uses an API key, creates sessions, bypasses rate limits (reusable singleton)
• Session client — uses a session cookie, acts on behalf of a user (create per-request, never share)

from appwrite.client import Client
from appwrite.services.account import Account
from flask import request, jsonify, make_response, redirect

# Admin client (reusable)
admin_client = (Client()
    .set_endpoint('https://.cloud.appwrite.io/v1')
    .set_project('[PROJECT_ID]')
    .set_key(os.environ['APPWRITE_API_KEY']))

# Session client (create per-request)
session_client = (Client()
    .set_endpoint('https://.cloud.appwrite.io/v1')
    .set_project('[PROJECT_ID]'))

session = request.cookies.get('a_session_[PROJECT_ID]')
if session:
    session_client.set_session(session)

Email/Password Login

@app.post('/login')
def login():
    account = Account(admin_client)
    session = account.create_email_password_session(
        request.json['email'], request.json['password']
    )

    # Cookie name must be a_session_
    resp = make_response(jsonify({'success': True}))
    resp.set_cookie('a_session_[PROJECT_ID]', session['secret'],
                    httponly=True, secure=True, samesite='Strict',
                    expires=session['expire'], path='/')
    return resp

Authenticated Requests

@app.get('/user')
def get_user():
    session = request.cookies.get('a_session_[PROJECT_ID]')
    if not session:
        return jsonify({'error': 'Unauthorized'}), 401

    session_client = (Client()
        .set_endpoint('https://.cloud.appwrite.io/v1')
        .set_project('[PROJECT_ID]')
        .set_session(session))

    account = Account(session_client)
    return jsonify(account.get())

OAuth2 SSR Flow

# Step 1: Redirect to OAuth provider
@app.get('/oauth')
def oauth():
    account = Account(admin_client)
    redirect_url = account.create_o_auth2_token(
        OAuthProvider.Github,
        'https://example.com/oauth/success',
        'https://example.com/oauth/failure',
    )
    return redirect(redirect_url)

# Step 2: Handle callback — exchange token for session
@app.get('/oauth/success')
def oauth_success():
    account = Account(admin_client)
    session = account.create_session(request.args['userId'], request.args['secret'])

    resp = make_response(jsonify({'success': True}))
    resp.set_cookie('a_session_[PROJECT_ID]', session['secret'],
                    httponly=True, secure=True, samesite='Strict',
                    expires=session['expire'], path='/')
    return resp

> Cookie security: Always use httponly, secure, and samesite='Strict' to prevent XSS. The cookie name must be a_session_.

> Forwarding user agent: Call session_client.set_forwarded_user_agent(request.headers.get('user-agent')) to record the end-user's browser info for debugging and security.

Error Handling

from appwrite.exception import AppwriteException

try:
    row = tables_db.get_row('[DATABASE_ID]', '[TABLE_ID]', '[ROW_ID]')
except AppwriteException as e:
    print(e.message)    # human-readable error message
    print(e.code)       # HTTP status code (int)
    print(e.type)       # Appwrite error type string (e.g. 'document_not_found')
    print(e.response)   # full response body (dict)

Common error codes:

| Code | Meaning | |------|---------| | 401 | Unauthorized — missing or invalid session/API key | | 403 | Forbidden — insufficient permissions for this action | | 404 | Not found — resource does not exist | | 409 | Conflict — duplicate ID or unique constraint violation | | 429 | Rate limited — too many requests, retry after backoff |

Permissions & Roles (Critical)

Appwrite uses permission strings to control access to resources. Each permission pairs an action (read, update, delete, create, or write which grants create + update + delete) with a role target. By default, no user has access unless permissions are explicitly set at the row/file level or inherited from the table/bucket settings. Permissions are arrays of strings built with the Permission and Role helpers.

from appwrite.permission import Permission
from appwrite.role import Role

Database Row with Permissions

doc = tables_db.create_row('[DATABASE_ID]', '[TABLE_ID]', ID.unique(), {
    'title': 'Hello World'
}, [
    Permission.read(Role.user('[USER_ID]')),     # specific user can read
    Permission.update(Role.user('[USER_ID]')),   # specific user can update
    Permission.read(Role.team('[TEAM_ID]')),     # all team members can read
    Permission.read(Role.any()),                 # anyone (including guests) can read
])

File Upload with Permissions

file = storage.create_file('[BUCKET_ID]', ID.unique(), InputFile.from_path('/path/to/file.png'), [
    Permission.read(Role.any()),
    Permission.update(Role.user('[USER_ID]')),
    Permission.delete(Role.user('[USER_ID]')),
])

> When to set permissions: Set row/file-level permissions when you need per-resource access control. If all rows in a table share the same rules, configure permissions at the table/bucket level and leave row permissions empty.

> Common mistakes: > - Forgetting permissions — the resource becomes inaccessible to all users (including the creator) > - Role.any() with write/update/delete — allows any user, including unauthenticated guests, to modify or remove the resource > - Permission.read(Role.any()) on sensitive data — makes the resource publicly readable

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: appwrite
• Source: appwrite/agent-skills
• License: BSD-3-Clause

Install and usage instructions live in the source repository linked above.