Skip to content
Jiayang Cloud Docs

Python

The jiayang package verifies the identity token the edge sends your app and gives you the caller. It works with any framework, and ships integrations for Flask, FastAPI, Django, Streamlit and Gradio.

Terminal window
pip install jiayang

It needs Python 3.10 or newer, and depends only on PyJWT (with its crypto extra). Each framework integration needs its framework, and an extra says so at install time:

Terminal window
pip install "jiayang[flask]"      # or [fastapi], [django], [streamlit], [gradio]

Any framework

Section titled “Any framework”

require_user(headers) takes the request’s headers and returns the caller, or raises Unauthorized.

from jiayang import Unauthorized, require_user


@app.get("/")
def index():
    try:
        user = require_user(request.headers)
    except Unauthorized:
        return "unauthorized", 401
    return f"hello {user.email or user.sub}"

The headers can be any mapping with .get: Flask’s and Starlette’s request.headers, Django’s request.META, or a plain dict. The header name is matched in any case.

It returns a User:

@dataclass(frozen=True)
class User:
    kind: str              # "user", or "service" for a bypass token
    sub: str               # "usr_…", or "service:<token id>"
    email: str | None      # None for a bypass token
    role: str              # "viewer", "editor" or "owner"
    workspace_id: str

See the SDK overview for what each field means and what the SDK checks.

Roles

Section titled “Roles”
from jiayang import Forbidden, has_role, require_role


require_role(user, "editor")   # raises Forbidden unless the caller is an editor or an owner


if has_role(user, "owner"):
    ...

Forbidden has status = 403 and needed, the role that was asked for.

Flask

Section titled “Flask”
from jiayang.flask import current_user, get_user, login_required


@app.get("/")
@login_required
def index():
    return f"hello {current_user.email}"


@app.post("/orders")
@login_required(role="editor")
def place_order():
    ...

@login_required answers 401 unauthorized without a valid caller, and 403 forbidden: this needs editor when the caller has less than role. Neither reaches your view.

current_user is the caller @login_required verified. Using it in a view without the decorator raises RuntimeError. get_user() returns the caller or None, for a page that renders either way.

FastAPI

Section titled “FastAPI”
from typing import Annotated


from fastapi import Depends
from jiayang import User
from jiayang.fastapi import CurrentUser, optional_user, requires


@app.get("/")
async def index(user: CurrentUser):
    return {"hello": user.email}


@app.post("/orders")
async def place_order(user: Annotated[User, Depends(requires("editor"))]):
    ...


@app.get("/welcome")
async def welcome(user: Annotated[User | None, Depends(optional_user)]):
    ...

The dependencies raise HTTPException, so FastAPI answers 401 ("detail": "unauthorized") or 403 ("detail": "forbidden: this needs editor") and your handler never runs. optional_user gives None instead of refusing.

They verify without blocking the event loop. The request is verified once, however many dependencies ask for the caller.

Django

Section titled “Django”

Add the middleware:

MIDDLEWARE = [
    "jiayang.django.JiayangMiddleware",
    # ...
]

Then guard the views that need a caller:

from django.http import HttpResponse
from jiayang.django import login_required, role_required


def index(request):
    user = request.jiayang_user            # None when there is no valid caller
    return HttpResponse(f"hello {user.email if user else 'stranger'}")


@login_required
def account(request):
    return HttpResponse(f"hello {request.jiayang_user.email}")


@role_required("editor")
def place_order(request):
    ...

The middleware sets request.jiayang_user to the caller or None, and refuses nothing by itself, so one app can have public and private pages. login_required answers 401 unauthorized. role_required answers 401 without a caller and 403 forbidden: this needs editor with too little role.

Streamlit

Section titled “Streamlit”
import streamlit as st
from jiayang.streamlit import get_user


user = get_user()
if user is None:
    st.error("sign in to use this")
    st.stop()
st.write(f"hello {user.email}")

require_user() raises Unauthorized instead of returning None.

The caller is verified once per session and remembered. Streamlit only sees the headers of the request that opened the session, and the token in them lasts 60 seconds. Checking again on a rerun would refuse someone who never left. On the platform, the edge closes the connection within a minute of someone’s access being taken away, and the session ends with it.

Gradio

Section titled “Gradio”
import os


import gradio as gr
from jiayang.gradio import require_user


def answer(question, request: gr.Request):
    user = require_user(request)
    return f"{user.email} asked: {question}"


gr.Interface(answer, "textbox", "textbox").launch(
    server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860"))
)

On the platform the app has to listen on 0.0.0.0 at the port in PORT. Gradio’s defaults, 127.0.0.1 and 7860, can’t be reached there.

Gradio passes the request to any handler that asks for one by type. It passes None when the handler is reached some other way, such as the API or a cached example. require_user refuses that: it’s a caller your app knows nothing about. get_user(request) returns None instead of raising.

async

Section titled “async”

require_user fetches the platform’s keys when it has none cached, about once every five minutes, and that fetch can block for up to three seconds. In an async app, use jiayang.aio. It does the fetch on a thread, one at a time, and everything else stays on the loop.

from fastapi import HTTPException, Request
from jiayang import Unauthorized
from jiayang.aio import require_user


@app.get("/")
async def index(request: Request):
    try:
        user = await require_user(request.headers)
    except Unauthorized as err:
        raise HTTPException(401, str(err))
    return {"hello": user.email}

Left uncaught, Unauthorized is a 500. In FastAPI you rarely need this: the CurrentUser dependency uses jiayang.aio and answers 401 for you.

Webhooks

Section titled “Webhooks”

A delivery on a verified public path arrives with a token of kind webhook, not a person’s. require_webhook checks it and says which provider signed the delivery. require_user refuses a webhook’s token, and require_webhook refuses a person’s.

from jiayang import Unauthorized, require_webhook


@app.post("/hooks/stripe")
def stripe_hook():
    try:
        hook = require_webhook(request.headers, provider="stripe")
    except Unauthorized:
        return "unauthorized", 401
    event = request.get_json()
    # hook.delivery is Stripe's event id. Skip one you've handled.
    return "", 204

provider is required: stripe, github, slack, shopify, standard_webhooks or hmac_sha256, or a list of them, like provider=["github", "stripe"]. A delivery from any other provider is refused, and so is every delivery when the list is empty. It returns a Webhook:

@dataclass(frozen=True)
class Webhook:
    provider: str             # "stripe", "github", "slack", "shopify", "standard_webhooks" or "hmac_sha256"
    pattern: str              # the public path it came in on, like "/hooks/stripe"
    delivery: str | None      # the provider's signed id, None where it signs none
    signed_at: int | None     # unix seconds, None where the provider signs no time
    workspace_id: str

The body arrives byte for byte as the provider sent it, so parse it as usual. In an async app, jiayang.aio.require_webhook is the same check as a coroutine. verify_webhook(token, provider="stripe") checks a webhook’s token you already have.

Each framework checks per route, so a webhook’s route and a person’s sit side by side:

# Flask
from jiayang.flask import get_webhook, webhook_required


@app.post("/hooks/stripe")
@webhook_required(provider="stripe")
def stripe_hook():
    hook = get_webhook()
    ...
# FastAPI
from typing import Annotated


from fastapi import Depends, Request
from jiayang import Webhook
from jiayang.fastapi import webhook


@app.post("/hooks/stripe")
async def stripe_hook(hook: Annotated[Webhook, Depends(webhook("stripe"))], request: Request):
    event = await request.json()
# Django
from jiayang.django import webhook_required


@webhook_required(provider="stripe")
def stripe_hook(request):
    hook = request.jiayang_webhook
    event = json.loads(request.body)
    ...

Each answers anything but a verified delivery from the providers it names with 401 before the view runs, a signed-in person included. Flask’s get_webhook() raises if @webhook_required didn’t run on that view.

A provider has no CSRF token to send. webhook_required exempts its Django view from the CSRF check, and with Flask-WTF’s CSRFProtect, add @csrf.exempt to the route yourself.

Configuration

Section titled “Configuration”

The SDK reads JIAYANG_APP_ID, JIAYANG_IDENTITY_ISSUER and JIAYANG_JWKS_URL from os.environ on each call. The platform sets them. To pass them yourself, use a Config:

from jiayang import Config, require_user


config = Config(app_id="…", issuer="https://edge.jiayang.cloud", jwks_url="https://edge.jiayang.cloud/.well-known/jwks.json")
user = require_user(request.headers, config)

verify_identity(token) checks a token you already have, the same way.

Errors

Section titled “Errors”

A refusal raises Unauthorized (status = 401) or Forbidden (status = 403). The message says why, for your logs:

Message Why
no identity token The request has no X-Jiayang-Identity header
invalid identity token The token failed a check
unknown signing key No key in the key set matches the token’s kid
couldn't fetch the identity keys The key set couldn’t be fetched, and there’s no fresh copy
JIAYANG_APP_ID, JIAYANG_IDENTITY_ISSUER and JIAYANG_JWKS_URL must be set The variables aren’t in the environment
no request to read an identity from Gradio passed None for the request
this needs editor A Forbidden from require_role(user, "editor")
not a webhook this route takes From require_webhook: a provider the route doesn’t name signed the delivery
invalid webhook token From require_webhook: the token’s webhook claims failed a check
no provider named From require_webhook: the call named no provider, so it refuses everything

A person’s token given to require_webhook, or a webhook’s given to require_user, fails as invalid identity token: its audience is the other kind’s.

Run it locally

Section titled “Run it locally”

jiayang dev starts your app behind the platform’s front door on your machine, with real tokens. It knows how to start Flask, FastAPI, Django, Streamlit and Gradio apps. See Local development.

Terminal window
jiayang dev