appwrite-python

Appwrite Python SDK

インストール

pip install appwrite

クライアントのセットアップ

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://<REGION>.cloud.appwrite.io/v1')
    .set_project(os.environ['APPWRITE_PROJECT_ID'])
    .set_key(os.environ['APPWRITE_API_KEY']))

コード例

ユーザー管理

users = Users(client)

# Create user
user = users.create(ID.unique(), '[email protected]', 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]')

データベース操作

注意: すべての新しいコードでは TablesDB を使用してください（非推奨の Databases クラスではありません）。Databases は、既存のコードベースがすでに依存している場合、またはユーザーが明示的に要求した場合にのみ使用してください。

ヒント: すべての SDK メソッド呼び出しでは、位置引数よりもキーワード引数（例: database_id='...'）を優先してください。位置引数のスタイルは、既存のコードベースがすでに使用している場合、またはユーザーが明示的に要求した場合にのみ使用してください。

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 型は非推奨です。すべての新しいカラムには明示的なカラム型を使用してください。

型	最大文字数	インデックス	ストレージ
varchar	16,383	完全インデックス（サイズが 768 以下の場合）	行内にインライン保存
text	16,383	プレフィックスのみ	オフページ
mediumtext	4,194,303	プレフィックスのみ	オフページ
longtext	1,073,741,823	プレフィックスのみ	オフページ

• varchar はインラインで保存され、64 KB の行サイズ制限にカウントされます。名前、スラッグ、識別子などの短くインデックスされたフィールドに適しています。
• text、mediumtext、longtext はオフページで保存されるため（行内には 20 バイトのポインタのみが存在します）、行サイズの予算を消費しません。これらの型には size は不要です。

# 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
    ]
)

クエリメソッド

# Filtering
Query.equal('field', 'value')             # == (or pass list for IN)
Query.not_equal('field', 'value')         # !=
Query.less_than('field', 100)             # <
Query.less_than_equal('field', 100)       # <=
Query.greater_than('field', 100)          # >
Query.greater_than_equal('field', 100)    # >=
Query.between('field', 1, 100)            # 1 <= field <= 100
Query.is_null('field')                    # is null
Query.is_not_null('field')                # is not null
Query.starts_with('field', 'prefix')      # starts with
Query.ends_with('field', 'suffix')        # ends with
Query.contains('field', 'sub')            # contains (string or array)
Query.search('field', 'keywords')         # full-text search (requires index)

# Sorting
Query.order_asc('field')
Query.order_desc('field')

# Pagination
Query.limit(25)                           # max rows (default 25, max 100)
Query.offset(0)                           # skip N rows
Query.cursor_after('[ROW_ID]')            # cursor pagination (preferred)
Query.cursor_before('[ROW_ID]')

# Selection & Logic
Query.select(['field1', 'field2'])        # return only specified fields
Query.or_queries([Query.equal('a', 1), Query.equal('b', 2)])   # OR
Query.and_queries([Query.greater_than('age', 18), Query.less_than('age', 65)])  # AND (default)
Query.relationship('comments', [Query.equal('approved', True), Query.limit(25)])  # related documents

# Aggregations (optional alias)
Query.count('*', 'total')
Query.sum('amount', 'revenue')
Query.avg('score', 'avg_score')
Query.min('price', 'min_price')
Query.max('price', 'max_price')

# Grouping
Query.group_by(['status', 'year'])
Query.having([Query.greater_than('total', 1)])
Query.distinct()

# Joins (table id, left key, right key, operator, alias)
Query.join('orders', '$id', 'customerId')
Query.left_join('orders', '$id', 'customerId', '=', 'ord')
Query.left_join('orders', 'ord', [
    Query.on('$id', 'customerId'),
    Query.equal('ord.status', 'paid'),
])

# Fluent builder. Pass build() into list methods.
queries = (
    Query.builder()
    .equal('status', 'published')
    .left_join('orders', '$id', 'customerId', '=', 'ord')
    .count('*', 'total')
    .group_by(['status'])
    .limit(25)
    .build()
)

tables_db.list_rows('[DATABASE_ID]', '[TABLE_ID]', queries)

ファイルストレージ

from appwrite.input_file import InputFile

storage = Storage(client)

# Upload file
file = storage.create_file('[BUCKET_ID]', ID.unique(), InputFile.from_path('/path/to/file.png'))

# List files
files = storage.list_files('[BUCKET_ID]')

# Delete file
storage.delete_file('[BUCKET_ID]', '[FILE_ID]')

InputFile ファクトリメソッド

from appwrite.input_file import InputFile

InputFile.from_path('/path/to/file.png')             # from filesystem path
InputFile.from_bytes(byte_data, 'file.png')          # from bytes
InputFile.from_string('Hello world', 'hello.txt')    # from string content

チーム

from appwrite.services.teams import Teams

teams = Teams(client)

# Create team
team = teams.create(ID.unique(), 'Engineering')

# List teams
team_list = teams.list()

# Create membership (invite user by email)
membership = teams.create_membership('[TEAM_ID]', roles=['editor'], email='[email protected]')

# List memberships
members = teams.list_memberships('[TEAM_ID]')

# Update membership roles
teams.update_membership('[TEAM_ID]', '[MEMBERSHIP_ID]', roles=['admin'])

# Delete team
teams.delete('[TEAM_ID]')

ロールベースのアクセス: 権限を設定する際は、すべてのチームメンバーには Role.team('[TEAM_ID]') を、特定のチームロールには Role.team('[TEAM_ID]', 'editor') を使用してください。

サーバーレス関数

functions = Functions(client)

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

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

関数ハンドラーの記述（Python ランタイム）

# 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

サーバーサイドレンダリング（SSR）認証

SSR アプリ（Flask、Django、FastAPI など）は、認証を処理するために サーバー SDK を使用します。2 つのクライアントが必要です:

• 管理クライアント — API キーを使用し、セッションを作成し、レート制限をバイパスします（再利用可能なシングルトン）
• セッションクライアント — セッション Cookie を使用し、ユーザーの代理として動作します（リクエストごとに作成し、決して共有しないでください）

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://<REGION>.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://<REGION>.cloud.appwrite.io/v1')
    .set_project('[PROJECT_ID]'))

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

メール/パスワードログイン

@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_<PROJECT_ID>
    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

認証済みリクエスト

@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://<REGION>.cloud.appwrite.io/v1')
        .set_project('[PROJECT_ID]')
        .set_session(session))

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

OAuth2 SSR フロー

# 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 のセキュリティ: XSS を防ぐために、常に httponly、secure、samesite='Strict' を使用してください。Cookie 名は a_session_<PROJECT_ID> である必要があります。

ユーザーエージェントの転送: デバッグとセキュリティのためにエンドユーザーのブラウザ情報を記録するには、session_client.set_forwarded_user_agent(request.headers.get('user-agent')) を呼び出してください。

エラーハンドリング

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)

一般的なエラーコード:

コード	意味
401	認証エラー — セッション/API キーが欠落しているか無効です
403	アクセス禁止 — このアクションに対する権限が不十分です
404	見つかりません — リソースが存在しません
409	競合 — ID の重複または一意制約違反です
429	レート制限 — リクエストが多すぎます。バックオフ後に再試行してください

権限とロール（重要）

Appwrite は権限文字列を使用してリソースへのアクセスを制御します。各権限は、アクション（read、update、delete、create、または create + update + delete を付与する write）とロールターゲットを組み合わせたものです。デフォルトでは、行/ファイルレベルで権限が明示的に設定されているか、テーブル/バケット設定から継承されていない限り、どのユーザーもアクセスできません。権限は Permission および Role ヘルパーで構築された文字列の配列です。

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

権限付きデータベース行

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 = 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]')),
])

パーミッションを設定すべきタイミング: リソースごとのアクセス制御が必要な場合は、行レベル/ファイルレベルのパーミッションを設定します。テーブル内のすべての行が同じルールを共有する場合は、テーブル/バケットレベルでパーミッションを設定し、行のパーミッションは空のままにします。

よくある間違い:

• パーミッションの設定忘れ — すべてのユーザー（作成者を含む）がリソースにアクセスできなくなります
• write/update/delete を指定した Role.any() — 未認証のゲストを含む任意のユーザーがリソースを変更または削除できるようになります
• 機密データに対する Permission.read(Role.any()) — リソースが誰でも読み取り可能になります