103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
import os
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor
|
|
|
|
|
|
def get_db_connection():
|
|
"""Get database connection"""
|
|
return psycopg2.connect(
|
|
host=os.getenv("DB_HOST", "localhost"),
|
|
port=int(os.getenv("DB_PORT", "5432")),
|
|
database=os.getenv("DB_NAME", "recipes_db"),
|
|
user=os.getenv("DB_USER", "recipes_user"),
|
|
password=os.getenv("DB_PASSWORD", "recipes_password"),
|
|
)
|
|
|
|
|
|
def create_user(username: str, email: str, password_hash: str, first_name: str = None, last_name: str = None, display_name: str = None, is_admin: bool = False):
|
|
"""Create a new user"""
|
|
conn = get_db_connection()
|
|
cur = conn.cursor(cursor_factory=RealDictCursor)
|
|
try:
|
|
# Use display_name if provided, otherwise use username
|
|
final_display_name = display_name if display_name else username
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO users (username, email, password_hash, first_name, last_name, display_name, is_admin)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id, username, email, first_name, last_name, display_name, is_admin, created_at
|
|
""",
|
|
(username, email, password_hash, first_name, last_name, final_display_name, is_admin)
|
|
)
|
|
user = cur.fetchone()
|
|
conn.commit()
|
|
return dict(user)
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|
|
|
|
|
|
def get_user_by_username(username: str):
|
|
"""Get user by username"""
|
|
conn = get_db_connection()
|
|
cur = conn.cursor(cursor_factory=RealDictCursor)
|
|
try:
|
|
cur.execute(
|
|
"SELECT id, username, email, password_hash, first_name, last_name, display_name, is_admin, created_at FROM users WHERE username = %s",
|
|
(username,)
|
|
)
|
|
user = cur.fetchone()
|
|
return dict(user) if user else None
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|
|
|
|
|
|
def get_user_by_email(email: str):
|
|
"""Get user by email"""
|
|
conn = get_db_connection()
|
|
cur = conn.cursor(cursor_factory=RealDictCursor)
|
|
try:
|
|
cur.execute(
|
|
"SELECT id, username, email, password_hash, first_name, last_name, display_name, is_admin, created_at FROM users WHERE email = %s",
|
|
(email,)
|
|
)
|
|
user = cur.fetchone()
|
|
return dict(user) if user else None
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|
|
|
|
|
|
def get_user_by_id(user_id: int):
|
|
"""Get user by ID"""
|
|
conn = get_db_connection()
|
|
cur = conn.cursor(cursor_factory=RealDictCursor)
|
|
try:
|
|
cur.execute(
|
|
"SELECT id, username, email, password_hash, first_name, last_name, display_name, is_admin, created_at FROM users WHERE id = %s",
|
|
(user_id,)
|
|
)
|
|
user = cur.fetchone()
|
|
return dict(user) if user else None
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|
|
|
|
|
|
def get_user_by_display_name(display_name: str):
|
|
"""Get user by display name"""
|
|
conn = get_db_connection()
|
|
cur = conn.cursor(cursor_factory=RealDictCursor)
|
|
try:
|
|
cur.execute(
|
|
"SELECT id, username, email, display_name, is_admin, created_at FROM users WHERE display_name = %s",
|
|
(display_name,)
|
|
)
|
|
user = cur.fetchone()
|
|
return dict(user) if user else None
|
|
finally:
|
|
cur.close()
|
|
conn.close()
|