-
Notifications
You must be signed in to change notification settings - Fork 7
/
test_server_postgres.py
88 lines (72 loc) · 2.84 KB
/
test_server_postgres.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
from tangerine import Tangerine, Ctx, Router
from tangerine_auth import Yuzu, KeyLime
import json
import jwt
import psycopg2
from middleware_extension import cors_middleware
app = Tangerine()
keychain = KeyLime({
"SECRET_KEY": "ILOVECATS",
})
app.use(cors_middleware)
def get_user_by_email(email):
conn = psycopg2.connect("postgresql://postgres:C4melz!!@localhost:5432/local_development")
cur = conn.cursor()
cur.execute("SELECT * FROM tangerine.users WHERE email = %s", (email,))
user = cur.fetchone()
cur.close()
conn.close()
if user:
return {'_id': user[0], 'email': user[1], 'password': user[2]}
else:
return None
def create_user(user_data):
conn = psycopg2.connect("postgresql://postgres:C4melz!!@localhost:5432/local_development")
cur = conn.cursor()
cur.execute("INSERT INTO tangerine.users (email, password) VALUES (%s, %s) RETURNING id", (user_data['email'], user_data['password']))
user_id = cur.fetchone()[0]
conn.commit()
cur.close()
conn.close()
return {'_id': user_id, 'email': user_data['email'], 'password': user_data['password']}
auth = Yuzu(keychain, get_user_by_email, create_user)
# serve static files to any request not starting with /api
app.static('^/(?!api).*$', './public')
def signup(ctx: Ctx) -> None:
user_data = ctx.request.body
created_user = auth.sign_up(user_data)
if created_user:
ctx.body = json.dumps(created_user)
ctx.send(201, content_type='application/json')
else:
ctx.send(500, content_type='application/json')
def login(ctx: Ctx) -> None:
user_data = ctx.request.body
email = user_data['email']
password = user_data['password']
user_id, token = auth.login(email, password)
if token:
ctx.body = json.dumps({"message": "Logged in successfully", "token": token})
ctx.set_res_header("Set-Cookie", f"auth_token={token}; HttpOnly; Path=/")
ctx.send(200, content_type='application/json')
else:
ctx.body = json.dumps({"message": "Invalid credentials"})
ctx.send(401, content_type='application/json')
def logout(ctx: Ctx) -> None:
auth.logout()
ctx.body = json.dumps({"message": "Logged out successfully"})
ctx.send(200, content_type='application/json')
@Router.auth_required
def get_protected_content(ctx: Ctx) -> None:
ctx.body = json.dumps({"message": "This is protected content. Only authenticated users can see this. I hope you feel special 🍊🍊🍊."})
ctx.send(200, content_type='application/json')
# ==================== API ROUTES ====================
api_router = Router(prefix='/api')
api_router.post('/logout', logout)
api_router.post('/login', login)
api_router.post('/signup', signup)
api_router.get('/protected', get_protected_content)
app.use(auth.jwt_middleware)
app.use_router(api_router)
app.start()
app.use(cors_middleware)