Points de terminaison

Starlette contient HTTPEndpoint et de WebSocketEndpoint Classes, qui fournissent un mode de vue basé sur les classes pour gérer la planification des méthodes HTTP et les sessions WebSocket.

Le point final HTTP

HTTPEndpoint Les classes peuvent être utilisées comme applications ASGI:

from starlette.responses import PlainTextResponse
de starlette.endpoints import HTTPendpoint


Application de classe (HTTPendpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")

Si vous utilisez une instance de l'application Starlette pour gérer le routage, vous pouvez planifier vers une classe HTTPEndpoint. assurez-vous que la planification se fait vers la classe elle-même et non vers une instance de la classe:

de starlette.applications import Starlette
from starlette.responses import PlainTextResponse
de starlette.endpoints import HTTPendpoint
de starlette.routing Route d'importation


Accueil de la classe (HTTPendpoint):
    async def get(self, request):
        return PlainTextResponse(f"Hello, world!")


Classe Utilisateur (HTTPendpoint):
    async def get(self, request):
        nom d'utilisateur = request.path_params ['nom d'utilisateur']
        return PlainTextResponse(f"Hello,{username}")

Routes = [
    Route("/", Homepage),
    Route("/{username}", utilisateur)
]

app = Starlette(routes=routes)

Pour toute méthode de requête qui n'est pas mappée au gestionnaire approprié, la classe de point final HTTP répondra par une réponse "405 Method Not Allowed".

WebSocket Endpoint

La classe WebSocketEndpoint est une application ASGI qui fournit un wrapper autour des fonctionnalités de l'instance WebSocket.

La plage de connexion ASGI est accessible via .scope sur l'instance de point de terminaison et possède une propriété encoding qui peut être définie de manière facultative afin de valider les données WebSocket attendues dans la méthode on_receive.

Les types de codage sont:

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

Il existe trois façons réinscriptibles de traiter un type de message ASGI WebSocket particulier:

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

• async def on_disconnect(websocket, close_code)

  de starlette.endpoints import WebSocketEndpoint

  Classe App(WebSocketEndpoint): encoding ="octets"

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

WebSocketEndpoint peut également être utilisé avec la classe d'application Starlette:

Importation d'uvicorn
de starlette.applications import Starlette
de starlette.endpoints import WebSocketEndpoint, HTTPendpoint
de starlette.responses à import HTMLResponse
de starlette.routing Route d'importation, WebSocketRoute


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

Accueil de la classe (HTTPendpoint):
    async def get(self, request):
        return HTMLResponse(html)

Classe Echo(WebSocketEndpoint):
    encoding = "texte"

    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)