Was ist der beste Weg, um WebSockets zu implementieren, um Datenbankänderungen zu verfolgen? ⇐ Python

[Anonymous]

Ich verbinde mich mit einer MySQL -Datenbank mit einem Fastapi -Backend und benötige Live -Updates zwischen der Rückseite und Frontend, wenn das Inventar in der Datenbank geändert wird. Das Hauptziel besteht darin, zu vermeiden, dass Bestände an Gegenständen angezeigt werden, die aktiv verkauft werden (wechseln von 1 auf 0) während der Suche nach der Suche. Ich kann keine Bibliothek finden, die hilft, diese Art von Funktionalität zu organisieren, und es sieht so aus, als würde ich sie selbst implementieren. Gibt es eine beste Methode, um dies zu tun? Auf den ersten Blick sieht es so aus, als würde ich in meinem WebSocket -Rückruf wahrscheinlich eine Menge Objektsortierungen/-bearbeitung durchführen. Leider sind die Funktionen, die für das Abrufen und Rücksenden der Daten verantwortlich sind, nicht alle identische Informationen. /> Anfrage als normal verarbeitet, aber vor der Rückkehr -Anweisung sortiert
durch verbundene Websockets, findet einen mit passender Route und fügt die Rückgabedaten zu einem Wörterbuch, das den übereinstimmenden Socket enthält, hinzu. Andere. < /li>
Frontend -Updates basierend auf Daten, die von Socket empfangen werden. Hat jemand mit Erfahrung eine gute Lösung/Empfehlung, um diese Funktion zu machen?

Code: Select all

    def __init__(self):
self.activate_connections: list[WebSocket] = []

async def connect(self, websocket: WebSocket, route: str):
await websocket.accept()
self.activate_connections.append({
'socket': websocket,
'route': route
})

async def disconnect(self, websocket: WebSocket, route: str):
for socket in self.activate_connections:
if (socket['route'] == route):
self.activate_connections.remove(socket)

async def addReturnedData(self, data: dict, route: str):
for connection in self.activate_connections:
print(connection)
if connection['route'] == route:
connection['data'] = data

async def broadcast(self, newData: dict):
print(newData)
jsonData = json.dumps(newData)
print(jsonData)
for connection in self.activate_connections:
await connection['socket'].send_json(jsonData)