Create OpenUrl packet type

src/textual_web/app_session.py

@@ -231,15 +231,15 @@ class AppSession(Session):
                     type_bytes = await readexactly(1)
                     size_bytes = await readexactly(4)
                     size = from_bytes(size_bytes, "big")
-                    data = await readexactly(size)
+                    payload = await readexactly(size)
                     if type_bytes == DATA:
-                        await on_data(data)
+                        await on_data(payload)
                     elif type_bytes == META:
-                        meta_data = json.loads(data)
+                        meta_data = json.loads(payload)
                         if meta_data.get("type") in {"exit", "blur", "focus"}:
                             await self.send_meta({"type": meta_data["type"]})
                         else:
-                            await on_meta(json.loads(data))
+                            await on_meta(json.loads(payload))
 
         except IncompleteReadError:
             # Incomplete read means that the stream was closed

src/textual_web/ganglion_client.py

@@ -22,6 +22,7 @@ from .packets import (
     PACKET_MAP,
     Handlers,
     NotifyTerminalSize,
+    OpenUrl,
     Packet,
     RoutePing,
     RoutePong,
@@ -327,7 +328,7 @@ class GanglionClient(Handlers):
             log.exception(str(error))
 
     async def post_connect(self) -> None:
-        """Called immediately after connecting to server."""
+        """Called immediately after connecting to the Ganglion server."""
         # Inform the server about our apps
         try:
             apps = [
@@ -347,7 +348,7 @@ class GanglionClient(Handlers):
             self._connected_event.set()
 
     async def send(self, packet: Packet) -> bool:
-        """Send a packet.
+        """Send a packet to the Ganglion server through the websocket.
 
         Args:
             packet: Packet to send.
@@ -436,3 +437,6 @@ class GanglionClient(Handlers):
         )
         if session_process is not None:
             await session_process.send_meta({"type": "blur"})
+
+    async def on_open_url(self, packet: OpenUrl) -> None:
+        return await super().on_open_url(packet)

src/textual_web/packets.py

@@ -1,7 +1,7 @@
 """
 This file is auto-generated from packets.yml and packets.py.template
 
-Time: Tue Nov 28 13:57:53 2023
+Time: Tue Jul 30 10:27:41 2024
 Version: 1
 
 To regenerate run `make packets.py` (in src directory)
@@ -10,6 +10,7 @@ To regenerate run `make packets.py` (in src directory)
 
 
 """
+
 from __future__ import annotations
 
 from enum import IntEnum
@@ -73,6 +74,9 @@ class PacketType(IntEnum):
     # App was blurred.
     BLUR = 13  # See Blur()
 
+    # Open a URL in the browser.
+    OPEN_URL = 14  # See OpenUrl()
+
 
 class Packet(tuple):
     """Base class for a packet.
@@ -877,6 +881,65 @@ class Blur(Packet):
         return self[1]
 
 
+# PacketType.OPEN_URL (14)
+class OpenUrl(Packet):
+    """Open a URL in the browser.
+
+    Args:
+        url (str): URL to open.
+        new_tab (bool): Open in new tab.
+
+    """
+
+    sender: ClassVar[str] = "both"
+    """Permitted sender, should be "client", "server", or "both"."""
+    handler_name: ClassVar[str] = "on_open_url"
+    """Name of the method used to handle this packet."""
+    type: ClassVar[PacketType] = PacketType.OPEN_URL
+    """The packet type enumeration."""
+
+    _attributes: ClassVar[list[tuple[str, Type]]] = [
+        ("url", str),
+        ("new_tab", bool),
+    ]
+    _attribute_count = 2
+    _get_handler = attrgetter("on_open_url")
+
+    def __new__(cls, url: str, new_tab: bool) -> "OpenUrl":
+        return tuple.__new__(cls, (PacketType.OPEN_URL, url, new_tab))
+
+    @classmethod
+    def build(cls, url: str, new_tab: bool) -> "OpenUrl":
+        """Build and validate a packet from its attributes."""
+        if not isinstance(url, str):
+            raise TypeError(
+                f'packets.OpenUrl Type of "url" incorrect; expected str, found {type(url)}'
+            )
+        if not isinstance(new_tab, bool):
+            raise TypeError(
+                f'packets.OpenUrl Type of "new_tab" incorrect; expected bool, found {type(new_tab)}'
+            )
+        return tuple.__new__(cls, (PacketType.OPEN_URL, url, new_tab))
+
+    def __repr__(self) -> str:
+        _type, url, new_tab = self
+        return f"OpenUrl({abbreviate_repr(url)}, {abbreviate_repr(new_tab)})"
+
+    def __rich_repr__(self) -> rich.repr.Result:
+        yield "url", self.url
+        yield "new_tab", self.new_tab
+
+    @property
+    def url(self) -> str:
+        """URL to open."""
+        return self[1]
+
+    @property
+    def new_tab(self) -> bool:
+        """Open in new tab."""
+        return self[2]
+
+
 # A mapping of the packet id on to the packet class
 PACKET_MAP: dict[int, type[Packet]] = {
     1: Ping,
@@ -892,6 +955,7 @@ PACKET_MAP: dict[int, type[Packet]] = {
     11: NotifyTerminalSize,
     12: Focus,
     13: Blur,
+    14: OpenUrl,
 }
 
 # A mapping of the packet name on to the packet class
@@ -909,6 +973,7 @@ PACKET_NAME_MAP: dict[str, type[Packet]] = {
     "notifyterminalsize": NotifyTerminalSize,
     "focus": Focus,
     "blur": Blur,
+    "openurl": OpenUrl,
 }
 
 
@@ -977,6 +1042,10 @@ class Handlers:
         """App was blurred."""
         await self.on_default(packet)
 
+    async def on_open_url(self, packet: OpenUrl) -> None:
+        """Open a URL in the browser."""
+        await self.on_default(packet)
+
     async def on_default(self, packet: Packet) -> None:
         """Called when a packet is not handled."""