85 lines
3.5 KiB
Python
85 lines
3.5 KiB
Python
# app/services/appwrite_client.py
|
|
from __future__ import annotations
|
|
import os
|
|
from typing import Optional, Dict, Any, Mapping, Union
|
|
|
|
from flask import current_app, has_request_context, request, session
|
|
from appwrite.client import Client
|
|
from appwrite.services.account import Account
|
|
from appwrite.id import ID
|
|
|
|
|
|
|
|
class AppwriteAccountClient:
|
|
def __init__(self, cookies: Optional[Union[str, Mapping[str, str]]] = None, use_admin: bool = False) -> None:
|
|
endpoint = current_app.config.get("APPWRITE_ENDPOINT") or os.getenv("APPWRITE_ENDPOINT")
|
|
project_id = current_app.config.get("APPWRITE_PROJECT_ID") or os.getenv("APPWRITE_PROJECT_ID")
|
|
api_key = current_app.config.get("APPWRITE_API_KEY") or os.getenv("APPWRITE_API_KEY")
|
|
if not endpoint or not project_id:
|
|
raise RuntimeError("APPWRITE_ENDPOINT and APPWRITE_PROJECT_ID must be configured")
|
|
|
|
self.endpoint = endpoint
|
|
self.project_id = project_id
|
|
|
|
self.client = Client()
|
|
self.client.set_endpoint(self.endpoint)
|
|
self.client.set_project(self.project_id)
|
|
|
|
# If we need admin privileges (to get session.secret), set the API key
|
|
if use_admin:
|
|
if not api_key:
|
|
raise RuntimeError("APPWRITE_API_KEY is required when use_admin=True")
|
|
self.client.set_key(api_key)
|
|
|
|
# Bind session if available (explicit → browser cookie → Flask session)
|
|
bound = False
|
|
if cookies:
|
|
bound = self._bind_session_from(cookies)
|
|
if not bound and has_request_context():
|
|
bound = self._bind_session_from(request.cookies)
|
|
if not bound and has_request_context():
|
|
secret = session.get("appwrite_cookies")
|
|
if secret:
|
|
self.client.set_session(secret)
|
|
bound = True
|
|
|
|
self.account = Account(self.client)
|
|
|
|
@staticmethod
|
|
def session_cookie_key(project_id: str) -> str:
|
|
return f"a_session_{project_id}"
|
|
|
|
def _bind_session_from(self, cookies: Union[str, Mapping[str, str]]) -> bool:
|
|
if isinstance(cookies, str):
|
|
self.client.set_session(cookies); return True
|
|
key = f"a_session_{self.project_id}"
|
|
if key in cookies and cookies.get(key):
|
|
self.client.set_session(cookies[key]); return True
|
|
for v in cookies.values():
|
|
if v: self.client.set_session(v); return True
|
|
return False
|
|
|
|
# --- Auth & account helpers ---
|
|
def create_user(self, email: str, password: str, name: Optional[str] = None, user_id: Optional[str] = None) -> Dict[str, Any]:
|
|
return dict(self.account.create(user_id=user_id or ID.unique(), email=email, password=password, name=name))
|
|
|
|
def create_email_password_session(self, email: str, password: str) -> Dict[str, Any]:
|
|
return dict(self.account.create_email_password_session(email=email, password=password))
|
|
|
|
def create_jwt(self) -> Dict[str, Any]:
|
|
return dict(self.account.create_jwt())
|
|
|
|
def get_account(self) -> Dict[str, Any]:
|
|
return dict(self.account.get())
|
|
|
|
def logout_current(self) -> bool:
|
|
self.account.delete_session("current")
|
|
return True
|
|
|
|
# --- Email verification ---
|
|
def send_verification(self, callback_url: str) -> Dict[str, Any]:
|
|
return dict(self.account.create_verification(url=callback_url))
|
|
|
|
def complete_verification(self, user_id: str, secret: str) -> Dict[str, Any]:
|
|
return dict(self.account.update_verification(user_id=user_id, secret=secret))
|