This repository has been archived by the owner on Apr 16, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
156 lines (113 loc) · 4.88 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
from db_functions import dbh_request
from urllib import response
from fastapi import FastAPI, Request, Header, Form, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing import List
import asyncio
import uvicorn
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<h2>Your ID: <span id="ws-id"></span></h2>
<ul id='messages'>
</ul>
<script>
var client_id = Date.now()
document.querySelector("#ws-id").textContent = client_id;
var ws = new WebSocket(`ws://${location.host}}/ws/${client_id}`);
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 ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
print("Количество подключенных устройств:", len(self.active_connections))
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
print("Количество подключенных устройств:", len(self.active_connections))
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.get("/notifications")
async def get():
return HTMLResponse(html)
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await websocket.send_text('ping')
await asyncio.sleep(5)
except WebSocketDisconnect:
manager.disconnect(websocket)
#------------------API---------------#
@app.get("/api/notifications") #ТЕСТЫ ПРОЙДЕНЫ
async def get_notifications():
return dbh_request('select * from notifications')
@app.get("/api/notifications/{id}") #ТЕСТЫ ПРОЙДЕНЫ
async def get_notifications(id:int = Header()):
return dbh_request(f"select * from notifications where id = '{id}'")
@app.post("/api/notifications") #ТЕСТЫ ПРОЙДЕНЫ
async def post_notifications(type: str = Form(), title: str = Form(), content: str = Form()):
response = dbh_request(f'insert into notifications(type, title, content) values ("{type}", "{title}", "{content}");')
match response['status']:
case 'OK':
return 'Данные успешно вставлены, ', response['data']
case _:
return 'Ошибка, ', response['data']
@app.put("/api/notifications/{id}") #ТЕСТЫ ПРОЙДЕНЫ
async def put_notifications(id: int = Header(), type: str = Form(), title: str = Form(), content: str = Form()):
response = dbh_request(f'update notifications set type = "{type}", title = "{title}", content = "{content}" where id = "{id}"')
match response['status']:
case 'OK':
return 'Данные успешно изменены, ', response['data']
case _:
return 'Ошибка, ', response['data']
@app.delete("/api/notifications/{id}") #ТЕСТЫ ПРОЙДЕНЫ
async def delete_notifications(id: int):
response = dbh_request(f'delete from notifications where id = {id}')
match response['status']:
case 'OK':
return 'Данные успешно удалены, ', response['data']
case _:
return 'Ошибка, ', response['data']
@app.post("/api/notifications/{id}/send")
async def post_notifications_send(id: int, time:int = Form(default=0)):
response = dbh_request(f'select * from notifications where id = {id}')
await asyncio.sleep(time)
await manager.broadcast(str(response['data'][0]))
match response['status']:
case 'OK':
return 'Уведомление всем пользователям отправлено'
case _:
return 'Ошибка, ', response['data']
if __name__ == "__main__":
uvicorn.run(app, host = 'https://notifs-api.herokuapp.com', port = 8000)
#--ЗАПУСК--#
# uvicorn app:app --ws-ping-interval=2 --ws-ping-timeout=5 --reload