Python

1# One page — 20 records by default, 100 maximum.
2page = authix.get_user_data(user_id=user_id, category="notes", limit=50)
3
4for note in page:          # a Page iterates its records
5    print(note["title"])
6
7page.has_more              # bool
8page.next_cursor           # pass back as cursor=... for the next page
9
10# Or let the SDK follow the cursor for you.
11for note in authix.iter_user_data(user_id=user_id, category="notes"):
12    process(note)
13
14# Bound the traversal when the size is unknown.
15for row in authix.iter_all_users_data(q="invoice", max_pages=10):
16    process(row)

1# app/main.py
2from fastapi import Depends, FastAPI, HTTPException, Request
3from pydantic import BaseModel
4
5from neuctra_authix import (
6    AuthenticationError,
7    AuthixAPIError,
8    InsufficientScopeError,
9    NotFoundError,
10    ValidationError,
11)
12
13from .authix import authix
14
15app = FastAPI()
16
17
18class Credentials(BaseModel):
19    email: str
20    password: str
21
22
23class SignupBody(Credentials):
24    name: str
25
26
27@app.post("/auth/signup", status_code=201)
28def signup(body: SignupBody):
29    result = authix.signup_user(
30        name=body.name, email=body.email, password=body.password
31    )
32
33    # An unverified account can sign in but cannot store records, so send the
34    # code now rather than letting the first write fail with a 403.
35    authix.request_email_verification_otp(
36        user_id=result["user"]["id"], email=body.email
37    )
38
39    return {"user_id": result["user"]["id"]}
40
41
42@app.post("/auth/login")
43def login(body: Credentials):
44    return authix.login_user(email=body.email, password=body.password)
45
46
47def current_user(request: Request) -> str:
48    """Your own session check. Whatever it is, the id must come from here."""
49    user_id = request.session.get("user_id")
50    if not user_id:
51        raise HTTPException(status_code=401, detail="Not signed in")
52    return user_id
53
54
55@app.get("/notes")
56def list_notes(cursor: str | None = None, user_id: str = Depends(current_user)):
57    page = authix.get_user_data(
58        user_id=user_id,          # <- trusted, never from the request body
59        category="notes",
60        limit=20,
61        cursor=cursor,
62    )
63
64    return {
65        "data": page.data,
66        "has_more": page.has_more,
67        "next_cursor": page.next_cursor,
68    }

1# app/errors.py
2from fastapi import Request
3from fastapi.responses import JSONResponse
4
5from neuctra_authix import (
6    AuthixAPIError,
7    InsufficientScopeError,
8    VersionConflictError,
9)
10
11
12def install(app):
13    @app.exception_handler(AuthixAPIError)
14    def handle(request: Request, exc: AuthixAPIError):
15        # A scope error means this server used the wrong key. That is our bug,
16        # so it must not surface as a 403 the caller might try to "fix".
17        if isinstance(exc, InsufficientScopeError):
18            app.logger.error("wrong Neuctra Authix key for %s: %s", request.url.path, exc.hint)
19            return JSONResponse({"message": "Server misconfiguration."}, 500)
20
21        body = {"message": exc.message, "code": exc.code}
22
23        if isinstance(exc, VersionConflictError):
24            body["current_version"] = exc.current_version
25
26        # Passing the status through keeps a 409 a 409 — the caller can retry
27        # a conflict, and can do nothing at all with a 500.
28        return JSONResponse(body, exc.status)

1# app.py  (Flask)
2from flask import Flask, g, jsonify, request, session
3
4from neuctra_authix import AuthixAPIError
5
6from .authix import authix
7
8app = Flask(__name__)
9
10
11@app.before_request
12def load_user():
13    g.user_id = session.get("user_id")
14
15
16@app.errorhandler(AuthixAPIError)
17def on_authix_error(exc):
18    return jsonify(message=exc.message, code=exc.code), exc.status
19
20
21@app.get("/notes")
22def list_notes():
23    if not g.user_id:
24        return jsonify(message="Not signed in"), 401
25
26    page = authix.get_user_data(
27        user_id=g.user_id,
28        category="notes",
29        limit=20,
30        cursor=request.args.get("cursor"),
31    )
32
33    return jsonify(
34        data=page.data, has_more=page.has_more, next_cursor=page.next_cursor
35    )

1from neuctra_authix import VersionConflictError
2
3record = authix.get_single_user_data(user_id=user_id, data_id=data_id)
4
5try:
6    authix.update_user_data(
7        user_id=user_id,
8        data_id=data_id,
9        data={"status": "shipped"},
10        version=record["data"]["version"],
11    )
12except VersionConflictError as exc:
13    # Another writer got there first. Nothing was overwritten.
14    print("changed since read; now at version", exc.current_version)

1# Every operation commits, or none of them do.
2authix.batch(
3    user_id=user_id,
4    operations=[
5        {"type": "create", "dataCategory": "orders", "total": 40},
6        {"type": "update", "id": stock_id, "version": 3, "remaining": 9},
7        {"type": "delete", "id": draft_id},
8    ],
9)

1# Exact field match — a containment query, index-backed in Postgres.
2paid = authix.search_in_user_data(user_id=user_id, keys={"status": "paid"})
3
4# Substring match across the whole record.
5invoices = authix.search_in_all_app_users_data(q="invoice", limit=100)
6
7# Across users — secret key only.
8admins = authix.search_in_all_app_users(keys={"role": "admin", "isActive": True})

1# Short-lived processes — a CLI, a cron job, a test — should close the
2# connection pool deterministically.
3with Authix(app_id=APP_ID, secret_key=SECRET_KEY) as authix:
4    for user in authix.iter_all_users():
5        print(user["email"])

app/
├── authix.py          # one client for the process
├── main.py            # FastAPI app + routes
├── errors.py          # AuthixAPIError -> HTTP responses
└── routes/
    ├── auth.py        # signup, verify, login
    ├── notes.py       # per-user data, scoped by YOU
    └── admin.py       # cross-user reads