Build a simple web app with HTMX and WebSockets
import logging
from pathlib import Path

SQLITE_DB_DIR = Path.cwd() / "db_files"
SQLITE_DB_DIR.mkdir(exist_ok=True)
SQLITE_DB_URL = "sqlite:///" + str(SQLITE_DB_DIR / "db.sqlite")
TEMPLATES_PATH = Path.cwd() / Path("src") / "templates"
LOG_LEVEL = logging.INFO

src/config.py
dan-osull/python_htmx_websockets_demo
from tortoise import fields
from tortoise.models import Model


class Link(Model):
    id = fields.UUIDField(pk=True)
    added = fields.DatetimeField(auto_now_add=True)
    url = fields.TextField()
    title = fields.TextField(default=None, null=True)
    user_emoji = fields.TextField(default="🐍")

    @property
    def div_id(self) -> str:
        return "node_" + "".join(char for char in str(self.id) if char.isalnum())

    @property
    def added_date(self) -> str:
        return self.added.strftime("%-d %b %Y")

    @property
    def added_time(self) -> str:
        return self.added.strftime("%H:%M")

    @property
    def display_name(self) -> str:
        name = str(self.url)
        if self.title is not None:
            name = self.title
        if len(name) > 100:
            name = name[:100] + "..."
        return name

src/db.py
dan-osull/python_htmx_websockets_demo
from pathlib import Path
from random import randint


class Emoji(str):
    """A type to represent an Emoji character"""

    ...


# Emojis copied from https://c.r74n.com/emoji
EMOJI_PATH = Path.cwd() / Path("src") / "emoji.txt"
with open(EMOJI_PATH, "r") as file:
    ALL_EMOJIS = file.read()


def get_random_emoji() -> Emoji:
    index = randint(0, len(ALL_EMOJIS) - 2)
    return Emoji(ALL_EMOJIS[index])

src/emoji.py
dan-osull/python_htmx_websockets_demo
from fastapi import Request
from fastapi.templating import Jinja2Templates

from .config import TEMPLATES_PATH
from .db import Link

templates = Jinja2Templates(directory=TEMPLATES_PATH)


def render_link_html(link: Link) -> str:
    template_response = templates.TemplateResponse(
        "render_link.jinja2",
        {
            "request": Request(scope={"type": "http"}),
            "link": link,
        },
    )
    return template_response.body.decode()


def render_all_links_html(all_links: list[Link]) -> str:
    link_html = '<div id="main">'
    for link in all_links:
        link_html += render_link_html(link=link)
    return link_html + "</div>"


def render_append_html(link: Link) -> str:
    return (
        '<div id="main" hx-swap-oob="beforeend">'
        + render_link_html(link=link)
        + "</div>"
    )


def render_delete_html(link: Link) -> str:
    return f'<div id="{link.div_id}" hx-swap-oob="delete"></div>'

src/html.py
dan-osull/python_htmx_websockets_demo
from asyncio import get_running_loop
from concurrent.futures import ProcessPoolExecutor

import httpx
from bs4 import BeautifulSoup, SoupStrainer
from httpx import HTTPError

from .db import Link
from .html import render_link_html
from .logging import log
from .websockets import ConnectionManager

process_pool = ProcessPoolExecutor(max_workers=1)


async def _get_new_link_url_and_title(link: Link) -> Link:
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(str(link.url), follow_redirects=True)
        except HTTPError:
            log.info(f"HTTPError: {link.url}")
            return link
    link.url = str(response.url)  # type: ignore

    loop = get_running_loop()
    link_title_awaitable = loop.run_in_executor(
        process_pool, _parse_content_for_link_title, link, response.text
    )
    link = await link_title_awaitable

    return link


def _parse_content_for_link_title(link: Link, content: str) -> Link:
    """CPU bound, so runs in a separate process to avoid locking event loop"""
    strainer = SoupStrainer("title")
    soup = BeautifulSoup(content, features="lxml", parse_only=strainer)
    title_tag = soup.find("title")
    if title_tag is not None:
        link.title = title_tag.text  # type: ignore
    return link


async def save_and_broadcast_new_link_url_and_title(
    link: Link, connection_manager: ConnectionManager
) -> None:
    """
    Navigates to the user provided URL, checks for an updated address or title,
    saves to DB and broadcasts to clients (if necessary)
    """
    updated_link = await _get_new_link_url_and_title(link=link)
    if (link.url != updated_link.url) or (link.title != updated_link.title):
        await updated_link.save()
        response = render_link_html(link=updated_link)
        log.debug(response)
        await connection_manager.broadcast(response)

src/http_client.py
dan-osull/python_htmx_websockets_demo
import logging

from .config import LOG_LEVEL

logging.basicConfig(level=LOG_LEVEL)
log = logging.getLogger(__name__)

src/logging.py
dan-osull/python_htmx_websockets_demo
import json
from asyncio import create_task
from enum import StrEnum, auto
from typing import Any, Awaitable, Callable
from uuid import UUID

import tortoise.connection
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from pydantic import AnyHttpUrl, BaseModel, ValidationError
from tortoise import Tortoise

from .config import SQLITE_DB_URL, TEMPLATES_PATH
from .db import Link
from .emoji import Emoji, get_random_emoji
from .html import render_all_links_html, render_append_html, render_delete_html
from .http_client import save_and_broadcast_new_link_url_and_title
from .logging import log
from .websockets import ConnectionManager

RequestDict = dict[str, Any]
"""JSON payload from Websocket client"""


class FormCreate(BaseModel):
    url: AnyHttpUrl


async def create_link(
    request: RequestDict, connection_manager: ConnectionManager, user_emoji: Emoji
) -> None:
    form = FormCreate(**request)
    link = Link(url=form.url, user_emoji=user_emoji)
    await link.save()

    response = render_append_html(link=link)
    log.debug(response)
    await connection_manager.broadcast(response)

    create_task(
        save_and_broadcast_new_link_url_and_title(
            link=link, connection_manager=connection_manager
        )
    )


class FormDelete(BaseModel):
    id: UUID


async def delete_link(
    request: RequestDict, connection_manager: ConnectionManager, _user_emoji: Emoji
) -> None:
    form = FormDelete(**request)
    link = await Link.get(id=form.id)
    await link.delete()

    response = render_delete_html(link)
    log.debug(response)
    await connection_manager.broadcast(response)


class ApiAction(StrEnum):
    CREATE = auto()
    DELETE = auto()


# Register functions for API actions
ApiActionFunc = Callable[[RequestDict, ConnectionManager, Emoji], Awaitable[None]]
API_ACTION_FUNC_TABLE: dict[ApiAction, ApiActionFunc] = {
    ApiAction.CREATE: create_link,
    ApiAction.DELETE: delete_link,
}

app = FastAPI()
connection_manager = ConnectionManager()


@app.on_event("startup")
async def init_orm() -> None:
    """Based on `tortoise.contrib.fastapi.register_tortoise`"""
    await Tortoise.init(db_url=SQLITE_DB_URL, modules={"models": ["src.db"]})
    await Tortoise.generate_schemas()


@app.on_event("shutdown")
async def close_orm() -> None:
    await tortoise.connection.connections.close_all()


INDEX_PATH = TEMPLATES_PATH / "index.html"
with open(INDEX_PATH, "r") as file:
    INDEX_HTML = file.read()


@app.get("/")
async def index_html() -> HTMLResponse:
    return HTMLResponse(INDEX_HTML)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    # Register new connection in manager so it gets broadcast messages
    await connection_manager.connect(websocket)

    try:
        # Send HTML for existing Links to this new client
        links = await Link.all()
        await websocket.send_text(render_all_links_html(links))

        user_emoji = get_random_emoji()
        while True:
            # Wait for a message from the client
            raw_request: str = await websocket.receive_text()
            log.debug(raw_request)
            request: RequestDict = json.loads(raw_request)
            action = ApiAction(request["action"])
            function = API_ACTION_FUNC_TABLE[action]

            try:
                # Run the function for the requested action
                await function(request, connection_manager, user_emoji)
            except ValidationError as exception:
                # Pydantic error - triggered e.g. if Link URL is invalid
                log.warning(exception)

    except WebSocketDisconnect as exception:
        connection_manager.disconnect(websocket)
        if exception.code not in [1001, 1006]:
            # Codes seen in normal operation:
            # 1001 Going away
            # 1006 Closed abnormally
            # https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1
            raise exception

src/main.py
dan-osull/python_htmx_websockets_demo
from fastapi import WebSocket


class ConnectionManager:
    """
    Keeps track of current WebSockets clients so that we can broadcast() to them.

    From FastAPI tutorial: https://fastapi.tiangolo.com/advanced/websockets/
    """

    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

src/websockets.py
dan-osull/python_htmx_websockets_demo