blob: 2a3bb719adbe47e61cfb0729048d35eccd004db1 (
plain)
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
|
import psycopg
import beanstalkc
import json
def connect_to_beanstalkd():
return beanstalkc.Connection("localhost", 11300, parse_yaml=False, encoding="utf-8")
def connect_to_db():
return psycopg.connect("dbname=testdb user=postgres")
def handle_job(db_conn: psycopg.Connection, job: beanstalkc.Job):
body = json.loads(job.body)
user_id = body["userId"]
message = body["message"]
url = body["url"]
# TODO: Fetch push creds from DB by userId
# TODO: Send event to push server
# Super ugly hack for demo
import os
os.system(f"""
curl https://notifications.linus.onl/api/send-notification/40a466025b3aac7 \
--request POST \
--header "Content-Type: application/json" \
--data '{
"title": "{message}",
"url": "{url}"
}'
""")
print(f"PRETEND SENDIGN MESSAGE {user_id=} {message=} {url=}")
print(f"PRETEND SENDIGN MESSAGE {user_id=} {message=} {url=}")
print(f"PRETEND SENDIGN MESSAGE {user_id=} {message=} {url=}")
print(f"PRETEND SENDIGN MESSAGE {user_id=} {message=} {url=}")
print(f"PRETEND SENDIGN MESSAGE {user_id=} {message=} {url=}")
def main():
beanstalkc_conn = connect_to_beanstalkd()
db_conn = connect_to_db()
beanstalkc_conn.watch("notification")
while True:
job = beanstalkc_conn.reserve()
if not job: continue
handle_job(db_conn, job)
job.delete()
if __name__ == "__main__":
main()
|
