--- name: appwrite-python description: Appwrite Python SDK skill. Use when building server-side Python applications with Appwrite, including Django, Flask, and FastAPI integrations. Covers user management, database/table CRUD, file storage, and functions via API keys. --- # Appwrite Python SDK ## Installation ```bash pip install appwrite ``` ## Setting Up the Client ```python from appwrite.client import Client from appwrite.id import ID from appwrite.query import Query from appwrite.services.users import Users from appwrite.services.tablesdb 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 ```python 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. ```python 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]') ``` ### Query Methods ```python # 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) ``` ### File Storage ```python 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 Factory Methods ```python 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 ``` ### Teams ```python 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='user@example.com') # 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-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) ```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 ``` ### 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) ```python 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 ```python @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 ```python @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 ```python # 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 ```python 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 document/file level or inherited from the collection/bucket settings. Permissions are arrays of strings built with the `Permission` and `Role` helpers. ```python from appwrite.permission import Permission from appwrite.role import Role ``` ### Database Row with Permissions ```python 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 ```python 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 document/file-level permissions when you need per-resource access control. If all documents in a collection share the same rules, configure permissions at the collection/bucket level and leave document 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

appwrite-ruby

Scan your own AI components