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
|
from pydantic import BaseModel
import sqlite3
from pywebpush import webpush
from dotenv import load_dotenv
from os import getenv
class PushSubscriptionData(BaseModel):
endpoint: str
keys: object
conn = sqlite3.connect("db.db", check_same_thread=False)
cursor = conn.cursor()
load_dotenv()
PRIVATE_VAPID_KEY = getenv("PRIVATE_VAPID_KEY")
cursor.execute("""
CREATE TABLE IF NOT EXISTS subscriptions (
endpoint TEXT PRIMARY KEY,
keys_p256dh TEXT NOT NULL,
keys_auth TEXT NOT NULL
);""")
# --- subscribe
def subscribe(data: PushSubscriptionData):
cursor.execute("SELECT * FROM subscriptions WHERE endpoint = ?", (data.endpoint,))
result = cursor.fetchall()
if len(result) > 0:
return data.endpoint
cursor.execute("""
INSERT INTO subscriptions (endpoint, keys_p256dh, keys_auth)
VALUES (?, ?, ?)""", (data.endpoint, data.keys["p256dh"], data.keys["auth"]))
conn.commit()
cursor.execute("SELECT * FROM subscriptions");
result = cursor.fetchall()
for row in result:
print(row)
return data.endpoint
# --- send notification
# ---- not used yet
def send_notification(endpoint: str):
cursor.execute("SELECT * FROM subscriptions WHERE endpoint = ?", (endpoint,))
row = cursor.fetchall()[0]
print(row)
subscription_info = {
"endpoint": endpoint,
"keys": {
"p256dh": row[1],
"auth": row[2],
},
}
try:
webpush(
subscription_info=subscription_info,
data={
"title": "Hello, world!",
"body": "Hello, Victoria Hall!"
},
vapid_private_key=PRIVATE_VAPID_KEY,
vapid_claims={
"sub": "mailto:dev@altafcreator.com",
},
)
except Exception as exception:
print(exception)
|
