160 lines
4.2 KiB
Python
160 lines
4.2 KiB
Python
from db import get_connection
|
|
|
|
|
|
def get_user_by_username(username: str):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, username, password_hash, is_active, created_at
|
|
FROM admin_users
|
|
WHERE username = %s
|
|
LIMIT 1
|
|
""",
|
|
(username,),
|
|
)
|
|
return cur.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_user_by_id(user_id: int):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, username, password_hash, is_active, created_at
|
|
FROM admin_users
|
|
WHERE id = %s
|
|
LIMIT 1
|
|
""",
|
|
(user_id,),
|
|
)
|
|
return cur.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def create_auth_session_record(
|
|
user_id: int,
|
|
session_token: str,
|
|
expires_at,
|
|
ip_address: str | None = None,
|
|
user_agent: str | None = None,
|
|
):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO auth_sessions (
|
|
user_id,
|
|
session_token,
|
|
last_activity_at,
|
|
expires_at,
|
|
ip_address,
|
|
user_agent
|
|
)
|
|
VALUES (%s, %s, NOW(), %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(user_id, session_token, expires_at, ip_address, user_agent),
|
|
)
|
|
row = cur.fetchone()
|
|
conn.commit()
|
|
return row[0] if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_auth_session_by_token(session_token: str):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
s.id,
|
|
s.user_id,
|
|
s.session_token,
|
|
s.created_at,
|
|
s.last_activity_at,
|
|
s.expires_at,
|
|
s.revoked_at,
|
|
s.ip_address,
|
|
s.user_agent,
|
|
u.username,
|
|
u.is_active,
|
|
u.role
|
|
FROM auth_sessions s
|
|
JOIN admin_users u ON u.id = s.user_id
|
|
WHERE s.session_token = %s
|
|
LIMIT 1
|
|
""",
|
|
(session_token,),
|
|
)
|
|
return cur.fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def touch_auth_session_if_needed(session_token: str, expires_at, throttle_seconds: int = 60):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE auth_sessions
|
|
SET last_activity_at = NOW(),
|
|
expires_at = %s
|
|
WHERE session_token = %s
|
|
AND revoked_at IS NULL
|
|
AND last_activity_at < NOW() - (%s || ' seconds')::interval
|
|
""",
|
|
(expires_at, session_token, str(throttle_seconds)),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def revoke_auth_session(session_token: str):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE auth_sessions
|
|
SET revoked_at = NOW()
|
|
WHERE session_token = %s
|
|
AND revoked_at IS NULL
|
|
""",
|
|
(session_token,),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def revoke_all_user_sessions(user_id: int):
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE auth_sessions
|
|
SET revoked_at = NOW()
|
|
WHERE user_id = %s
|
|
AND revoked_at IS NULL
|
|
""",
|
|
(user_id,),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount
|
|
finally:
|
|
conn.close()
|