Endpoints

Starlette contains the HTTPEndpoint and WebSocketEndpoint classes, which provide class-based view patterns for handling HTTP method scheduling and WebSocket sessions.

HTTP endpoint

HTTPEndpoint Class can be used as an ASGI application:

from starlette.responses import PlainTextResponse
from starlette.endpoints import HTTPEndpoint


class App(HTTPEndpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")

If you are using Starlette application instances to handle routing, you can dispatch to an HTTPEndpoint class. Make sure you dispatch to the class itself, not to instances of the class:

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.endpoints import HTTPEndpoint
from starlette.routing import Route


class Homepage(HTTPEndpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")


class User(HTTPEndpoint):
    async def get(self, request):
        username = request.path_params['username']
        return PlainTextResponse(f"Hello, {username}")

routes = [
    Route("/", Homepage),
    Route("/{username}", User)
]

app = Starlette(routes=routes)

For any request method that is not mapped to a corresponding handler, the HTTP endpoint class will respond with a "405 method not allowed" response.

WebSocket endpoint

The WebSocketEndpoint class is an ASGI application that provides a wrapper around the functionality of an WebSocket instance.

The ASGI connection scope is accessible via .scope on the endpoint instance and has an attribute encoding that can optionally be set in order to validate the expected WebSocket data in the on_receive method.

Code type is:

• 'json'
• 'bytes'
• 'text'

There are three overridable ways to handle specific ASGI WebSocket message types:

• async def on_connect(websocket, **kwargs)
• async def on_receive(websocket, data)

• async def on_disconnect(websocket, close_code)

  from starlette.endpoints import WebSocketEndpoint

  class App(WebSocketEndpoint): encoding = 'bytes'

  async def on_connect(self, websocket):
      await websocket.accept()
  
  async def on_receive(self, websocket, data):
      await websocket.send_bytes(b"Message: " + data)
  
  async def on_disconnect(self, websocket, close_code):
      pass
  

WebSocketEndpoint can also be used with Starlette application classes:

import uvicorn
from starlette.applications import Starlette
from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute


html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""

class Homepage(HTTPEndpoint):
    async def get(self, request):
        return HTMLResponse(html)

class Echo(WebSocketEndpoint):
    encoding = "text"

    async def on_receive(self, websocket, data):
        await websocket.send_text(f"Message text was: {data}")

routes = [
    Route("/", Homepage),
    WebSocketRoute("/ws", Echo)
]

app = Starlette(routes=routes)