Skip to main content
Markdown

FastAPI and Starlette

wordplus_realtime.fastapi gives FastAPI the browser's routes as a router, for the user your app finds.

pip install 'wordplus-realtime[fastapi]'
npm install @wordplus/realtime

Your server​

from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from wordplus_realtime import Realtime
import wordplus_realtime.fastapi as wr

realtime = Realtime.from_environment(
    profile=lambda user: {"name": name_of(user)},
    worker="/static/realtime-worker.js",
    rooms="/static/realtime-rooms.min.js",
)
realtime.channel("private-orders-", lambda user, channel: user is not None and owner_of(channel[15:]) == user)

async def current_user(request: Request):
    session = await session_of(request)
    return session.user_id if session else None

app = FastAPI()
app.include_router(wr.router(realtime, current_user), prefix="/api/realtime")   # POST /auth, /calls, /keys
app.mount("/static", StaticFiles(directory="static"), name="static")            # npx wordplus-realtime copy static/
  • current_user(request) gives the request's user's id, or None. It may be async.
  • The owners (channel(), calls(), kind(), profile) are plain functions. The router runs them on a worker thread, so a blocking database query in one doesn't hold up the event loop.
  • The routes take JSON POSTs only. A page on another site can't send JSON without asking first, and it is never allowed to. A session cookie therefore needs no CSRF token here.

The page​

Render the settings into the page:

@app.get("/orders/{pk}")
async def order_page(request: Request, pk: int):
    config = realtime.config(await current_user(request))
    return templates.TemplateResponse(request, "order.html", {"realtime": config, "pk": pk})
<!-- order.html (Jinja) -->
<script>window.wordplusRealtimeConfig = {{ realtime|tojson }};</script>
import { connect } from '@wordplus/realtime';

const rt = await connect( window.wordplusRealtimeConfig );
rt.subscribe( `private-orders-${ pk }` ).on( 'updated', ( data ) => refresh( data ) );

A single-page app can skip that, and connect with connect( { site: 's1500000007', worker: '/static/realtime-worker.js' } ), which asks /api/realtime/keys first. An app on another address, such as one in Capacitor, uses endpoint: 'https://api.example.com/api/realtime', and your server answers the app's address with CORS.

Publishing​

@app.post("/orders/{pk}/pay")
async def pay(pk: int):
    order = await mark_paid(pk)
    await realtime.apublish(f"private-orders-{pk}", "updated", {"status": order.status})
    return order

apublish() runs the request on a worker thread. It raises RealtimeError when WordPlus Cloud refuses, with its code and status.