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
|
from pydantic import BaseModel
import sqlite3
from pywebpush import webpush
class PushSubscriptionData(BaseModel):
endpoint: str
expirationTime: float
keys: object
conn = sqlite3.connect("db.db", check_same_thread=False)
cursor = conn.cursor()
f = open("private_key.pem", "r")
PRIVATE_VAPID_KEY = f.read()
f.close()
cursor.execute("""
CREATE TABLE IF NOT EXISTS subscriptions (
endpoint TEXT PRIMARY KEY,
expiration_time REAL,
keys_p256dh TEXT NOT NULL,
keys_auth TEXT NOT NULL
);""")
# --- subscribe
def subscribe(data: PushSubscriptionData):
if data.expirationTime:
cursor.execute("""
INSERT INTO subscriptions (endpoint, expiration_time, keys_p256dh, keys_auth)
VALUES (?, ?, ?, ?)""", (data.endpoint, data.expirationTime, data.keys["p256dh"], data.keys["auth"]))
else:
cursor.execute("""
INSERT INTO subscriptions (endpoint, keys_p256dh, keys_auth)
VALUES (?, ?, ?, ?)""", (data.endpoint, data.keys["p256dh"], data.keys["auth"]))
conn.commit()
return data.endpoint
# --- send notification
def send_notification(endpoint: str):
cursor.execute("SELECT * FROM subscriptions WHERE endpoint = ?", (endpoint,))
row = cursor.fetchall()[0]
subscription_info = {
"endpoint": endpoint,
"keys": {
"p256dh": row[2],
"auth": row[3],
},
}
try:
webpush(
subscription_info=subscription_info,
data="Hello, world!",
vapid_private_key=PRIVATE_VAPID_KEY,
vapid_claims={
"sub": "mailto:dev@altafcreator.com",
},
)
except Exception as exception:
print(exception)
|
