py-sse
minimal python sse server for datastar — built on rsgi/granian + apsw/sqlite
sse.py → datastar event formatters (patch_elements, patch_signals, …)
app.py → rsgi app: routing, signals, cookies, static, @app.stream
db.py → sqlite helpers + Changes (update_hook → asyncio)
mserver.py → background server for notebooks (serve_background/stop_background)
ngrok.py → public tunnel + .env loader
pip install py-sse
uv add py-sse # preferred
hello
from py_sse import create_app, serve
app = create_app()
@app.get('/')
async def index(req): return '<h1>hello</h1>'
if __name__ == '__main__': serve(app)
live feed
Backend is the source of truth. Each write wakes every /feed stream; the
handler re-renders. No polling, no optimistic updates, no client state.
from py_sse import *
db = create_db('chat.db')
migrate(db, "CREATE TABLE IF NOT EXISTS msgs (id INTEGER PRIMARY KEY, txt TEXT)")
def startup(loop):
global changes
changes = Changes(db, loop)
app = create_app(on_init=startup, on_del=lambda loop: changes.close())
@app.post('/say')
async def say(req):
s = await signals(req)
write(db, lambda c: c.execute("INSERT INTO msgs(txt) VALUES(?)", (s['text'],)))
@app.stream('/feed', on=lambda: changes)
def feed(req):
rows = query(db, "SELECT txt FROM msgs ORDER BY id DESC LIMIT 50")
return ''.join(f'<p>{t}</p>' for (t,) in rows)
handler returns
str → 200 html
dict → 200 json
None → 204
(url, int) → redirect (3xx) or text body (4xx/5xx)
async gen → sse stream (use @app.stream for the common case)
api
create_app(on_init=, on_del=) → app handle: .get .post .put .patch .delete .mount .before .stream
@app.stream(path, on=) → sse route; re-renders each time `on` ticks
signals(req) → datastar signals (query on GET, json body otherwise)
body(req) / body_stream(req) → full body / chunked body
set_cookie(req, name, value) → queue a Set-Cookie
create_signer(secret) → .sign(value) / .unsign(signed) — hmac-sha256
static(app, prefix, dir) → zero-copy file serving, http range support
create_db(path) → wal sqlite connection
migrate(conn, sql) → apply schema in one txn
query(conn, sql, …, limit=) → rows as tuples
write(conn, fn, …) → run fn(conn, …) in a txn, return its result
Changes(db, loop) → fan db writes out to streams; .wait() / .close()
patch_elements(html, …) → datastar-patch-elements event
patch_signals(signals, …) → datastar-patch-signals event
remove_signals(*names) → null those signals
execute_script(js, …) → datastar-execute-script event
redirect(url) → navigate via patched script
serve(app) → foreground granian server (blocks)
serve_background(app) → background thread → ServerState
stop_background(state) → clean shutdown, 3s join
dev_alive(port) → is the port accepting connections
start_tunnel(port) → ngrok tunnel → TunnelState (needs NGROK_AUTHTOKEN)
stop_tunnel(tunnel) → close it
load_env(path='.env') → read KEY=VALUE; missing file is a noop
datastar
patch elements & signals → the verb; morph is the mechanism
fat morph → send large dom chunks, up to <html>
signals → user interaction only, used sparingly
cqrs → long-lived read stream + short-lived writes
brotli on sse → ratios up to 200:1
MIT · built from marimo notebooks with marimo-dev
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
py_sse-0.12.2.tar.gz
(13.0 kB
view details)
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
py_sse-0.12.2-py3-none-any.whl
(15.4 kB
view details)
File details
Details for the file py_sse-0.12.2.tar.gz.
File metadata
- Download URL: py_sse-0.12.2.tar.gz
- Upload date:
- Size: 13.0 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
uv/0.11.6 {"installer":{"name":"uv","version":"0.11.6","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Arch Linux ARM","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
e76418326ae242c50800c55cfa1f7b854e30911165f125d0bae3c7dcf62de78a
|
|
| MD5 |
8206cf70079ac9e3ed0ac6d9e8e44699
|
|
| BLAKE2b-256 |
768401fc121d3296810486ba13710bbf273d95264f53ed113638f8932a71b504
|
File details
Details for the file py_sse-0.12.2-py3-none-any.whl.
File metadata
- Download URL: py_sse-0.12.2-py3-none-any.whl
- Upload date:
- Size: 15.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
uv/0.11.6 {"installer":{"name":"uv","version":"0.11.6","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Arch Linux ARM","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
bd4495f865b692ea96c09fa4e213370534b850e8f1aa1b22b93663ce5cbb141a
|
|
| MD5 |
a3b81a02677e5bfcbdc7f5f7b2dd4660
|
|
| BLAKE2b-256 |
4717eae8e79f2a1a94dfc2696f7f93b5ca873e8f286fcbf1e85f754ae0c679e4
|
