patx/micropie
remove un-used send_websocket_response method
Commit 85f5d5e · patx · 2025-12-03T20:27:24-05:00
Comments
No comments yet.
Diff
diff --git a/examples/pastebin/app.py b/examples/pastebin/app.py
index 2127e41..ce7f750 100644
--- a/examples/pastebin/app.py
+++ b/examples/pastebin/app.py
@@ -1,33 +1,32 @@
-from uuid import uuid4
-import asyncio
from micropie import App
-from markupsafe import escape
-from pickledb import AsyncPickleDB
+import dataset
-db = AsyncPickleDB('pastes.json')
+db = dataset.connect('sqlite:///pastes.db')
+pastes = db['pastes']
class Root(App):
async def index(self, paste_content=None):
- if self.request.method == "POST":
- #paste_content = self.request.body_params.get('paste_content', [''])[0]
- pid = str(uuid4())
- await db.aset(pid, escape(paste_content))
- await db.asave()
- return self._redirect(f'/paste/{pid}')
+ if self.request.method == 'POST':
+ new_id = pastes.insert({'content': paste_content})
+ return self._redirect(f'/paste/{new_id}')
return await self._render_template('index.html')
async def paste(self, paste_id, delete=None):
if delete == 'delete':
- await db.aremove(paste_id)
- await db.asave()
+ pastes.delete(id=paste_id)
return self._redirect('/')
- paste_content = await db.aget(paste_id)
+
+ paste = pastes.find_one(id=paste_id)
+ if not paste:
+ paste = {'content':404}
+
return await self._render_template(
'paste.html',
paste_id=paste_id,
- paste_content=paste_content
+ paste_content=paste['content'],
)
+
app = Root()
diff --git a/examples/socketio/webtrc/app.py b/examples/socketio/webtrc/app.py
index 7f6cdf7..fafaa2d 100644
--- a/examples/socketio/webtrc/app.py
+++ b/examples/socketio/webtrc/app.py
@@ -19,7 +19,7 @@ class MyApp(App):
async def stream(self, username: str):
# Check if the username is already actively streaming
if username in active_users:
- return {'error': f'Username {username} is already actively streaming'}, 403
+ return 403, {'error': f'Username {username} is already actively streaming'}
# Mark the username active, render the streamer template
active_users.add(username)
return await self._render_template("stream.html", username=username)
diff --git a/micropie.py b/micropie.py
index 1b6b151..3fd49ce 100644
--- a/micropie.py
+++ b/micropie.py
@@ -80,8 +80,16 @@ class InMemorySessionBackend(SessionBackend):
return {}
async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
- self.sessions[session_id] = data
- self.last_access[session_id] = time.time()
+ """
+ If `data` is empty, treat this as a logout and remove the session.
+ Otherwise, upsert the session and update last_access.
+ """
+ if not data:
+ self.sessions.pop(session_id, None)
+ self.last_access.pop(session_id, None)
+ else:
+ self.sessions[session_id] = data
+ self.last_access[session_id] = time.time()
# -----------------------------
@@ -991,39 +999,6 @@ class App:
"more_body": False
})
- async def _send_websocket_response(
- self,
- send: Callable[[Dict[str, Any]], Awaitable[None]],
- status_code: int,
- body: bytes,
- extra_headers: List[Tuple[str, str]]
- ) -> None:
- """
- Send an HTTP response for WebSocket-related headers (e.g., cookies).
-
- Args:
- send: The ASGI send callable.
- status_code: The HTTP status code.
- body: The response body.
- extra_headers: List of header tuples.
- """
- sanitized_headers: List[Tuple[str, str]] = []
- for k, v in extra_headers:
- if "\n" in k or "\r" in k or "\n" in v or "\r" in v:
- print(f"Header injection attempt detected: {k}: {v}")
- continue
- sanitized_headers.append((k, v))
- await send({
- "type": "http.response.start",
- "status": status_code,
- "headers": [(k.encode("latin-1"), v.encode("latin-1")) for k, v in sanitized_headers],
- })
- await send({
- "type": "http.response.body",
- "body": body,
- "more_body": False
- })
-
async def _send_websocket_close(
self,
send: Callable[[Dict[str, Any]], Awaitable[None]],