FastHTML plus Jupyter plus websockets

websockets_example.ipynb

{
  "cells": [
    {
      "metadata": {},
      "id": "dcb7a6c1",
      "cell_type": "markdown",
      "source": "# Websockets in Jupyter"
    },
    {
      "metadata": {},
      "id": "1e4f42f0",
      "cell_type": "markdown",
      "source": "Use at least FastHTML 0.6.5. \n\nThis code is heavily inspired by this [code](https://docs.fastht.ml/tutorials/by_example.html#websockets) written by Jeremy Howard."
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "ccc69a8c",
      "cell_type": "code",
      "source": "# import necessary pieces\nfrom fasthtml.common import *\nfrom asyncio import sleep\n\n# Might be merged into fasthtml.common\nfrom fasthtml.jupyter import *",
      "execution_count": 1,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "257f7974",
      "cell_type": "code",
      "source": "def mk_inp(): return Input(id='msg')",
      "execution_count": 2,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "d18faba3",
      "cell_type": "code",
      "source": "# FastJupy() is a wrapper around the FastHTML() class\n# ws_hdr brings in the websocket headers\napp = FastJupy(ws_hdr=True)\nrt = app.route",
      "execution_count": 3,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "604e2c0b",
      "cell_type": "code",
      "source": "@rt('/')\nasync def get(request):\n    cts = Div(\n        Div(id='notifications'),\n        Form(mk_inp(), id='form', ws_send=True),\n        hx_ext='ws', ws_connect='/ws')\n    return Titled('Websocket Test', cts)",
      "execution_count": 9,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "4c54f144",
      "cell_type": "code",
      "source": "async def on_connect(send):\n    print('Connected!')\n    await send(Div('Hello, you have connected', id=\"notifications\"))\n\nasync def on_disconnect(ws):\n    print('Disconnected!')",
      "execution_count": 5,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "0affff9a",
      "cell_type": "code",
      "source": "@app.ws('/ws', conn=on_connect, disconn=on_disconnect)\nasync def ws(msg:str, send):\n    await send(Div('Hello ' + msg, id=\"notifications\"))\n    await sleep(2)\n    return Div('Goodbye ' + msg, id=\"notifications\"), mk_inp()",
      "execution_count": 6,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "99142b38",
      "cell_type": "code",
      "source": "port = 8000\nserver = JupyUvi(app, port=port)",
      "execution_count": 7,
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": "Connected!\n"
        }
      ]
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "4e8621e5",
      "cell_type": "code",
      "source": "HTMX()",
      "execution_count": 8,
      "outputs": [
        {
          "data": {
            "text/html": "<iframe src=\"http://localhost:8000\" style=\"width: 100%; border: none;\" onload=\"{\n        let frame = this;\n        window.addEventListener('message', function(e) {\n            if (e.data.height) frame.style.height = (e.data.height+1) + 'px';\n        }, false);\n    }\" allow=\"accelerometer;\nautoplay;\ncamera;\nclipboard-read;\nclipboard-write;\ndisplay-capture;\nencrypted-media;\nfullscreen;\ngamepad;\ngeolocation;\ngyroscope;\nhid;\nidentity-credentials-get;\nidle-detection;\nmagnetometer;\nmicrophone;\nmidi;\npayment;\npicture-in-picture;\npublickey-credentials-get;\nscreen-wake-lock;\nserial;\nusb;\nweb-share;\nxr-spatial-tracking\n\"></iframe> ",
            "text/plain": "<IPython.core.display.HTML object>"
          },
          "execution_count": 8,
          "metadata": {},
          "output_type": "execute_result"
        }
      ]
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "bda7bea5",
      "cell_type": "code",
      "source": "# Run me if you want to gracefully stop the HTTP server\n# without restarting Jupyter\nserver.stop()",
      "execution_count": null,
      "outputs": []
    },
    {
      "metadata": {
        "trusted": true
      },
      "id": "02ce0e9c",
      "cell_type": "code",
      "source": "",
      "execution_count": null,
      "outputs": []
    }
  ],
  "metadata": {
    "_draft": {
      "nbviewer_url": "https://gist.github.com/pydanny/1eee38ceec17a915a67f1c382d3387ea"
    },
    "gist": {
      "id": "1eee38ceec17a915a67f1c382d3387ea",
      "data": {
        "description": "FastHTML plus Jupyter plus websockets",
        "public": true
      }
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3 (ipykernel)",
      "language": "python"
    },
    "language_info": {
      "name": "python",
      "version": "3.10.6",
      "mimetype": "text/x-python",
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "pygments_lexer": "ipython3",
      "nbconvert_exporter": "python",
      "file_extension": ".py"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}