my-recipes/backend/grocery_db_utils.py

import os
from typing import List, Optional, Dict, Any
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_grocery_list(owner_id: int, name: str, items: List[str] = None) -> Dict[str, Any]:
    """Create a new grocery list"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        items = items or []
        cur.execute(
            """
            INSERT INTO grocery_lists (owner_id, name, items)
            VALUES (%s, %s, %s)
            RETURNING id, name, items, owner_id, created_at, updated_at
            """,
            (owner_id, name, items)
        )
        grocery_list = cur.fetchone()
        conn.commit()
        return dict(grocery_list)
    finally:
        cur.close()
        conn.close()


def get_user_grocery_lists(user_id: int) -> List[Dict[str, Any]]:
    """Get all grocery lists owned by or shared with a user"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            SELECT DISTINCT gl.id, gl.name, gl.items, gl.owner_id, gl.is_pinned, gl.created_at, gl.updated_at,
                   u.display_name as owner_display_name,
                   CASE WHEN gl.owner_id = %s THEN TRUE ELSE gls.can_edit END as can_edit,
                   CASE WHEN gl.owner_id = %s THEN TRUE ELSE FALSE END as is_owner
            FROM grocery_lists gl
            LEFT JOIN users u ON gl.owner_id = u.id
            LEFT JOIN grocery_list_shares gls ON gl.id = gls.list_id AND gls.shared_with_user_id = %s
            WHERE gl.owner_id = %s OR gls.shared_with_user_id = %s
            ORDER BY gl.updated_at DESC
            """,
            (user_id, user_id, user_id, user_id, user_id)
        )
        lists = cur.fetchall()
        return [dict(row) for row in lists]
    finally:
        cur.close()
        conn.close()


def get_grocery_list_by_id(list_id: int, user_id: int) -> Optional[Dict[str, Any]]:
    """Get a specific grocery list if user has access"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            SELECT gl.id, gl.name, gl.items, gl.owner_id, gl.is_pinned, gl.created_at, gl.updated_at,
                   u.display_name as owner_display_name,
                   CASE WHEN gl.owner_id = %s THEN TRUE ELSE gls.can_edit END as can_edit,
                   CASE WHEN gl.owner_id = %s THEN TRUE ELSE FALSE END as is_owner
            FROM grocery_lists gl
            LEFT JOIN users u ON gl.owner_id = u.id
            LEFT JOIN grocery_list_shares gls ON gl.id = gls.list_id AND gls.shared_with_user_id = %s
            WHERE gl.id = %s AND (gl.owner_id = %s OR gls.shared_with_user_id = %s)
            """,
            (user_id, user_id, user_id, list_id, user_id, user_id)
        )
        grocery_list = cur.fetchone()
        return dict(grocery_list) if grocery_list else None
    finally:
        cur.close()
        conn.close()


def update_grocery_list(list_id: int, name: str = None, items: List[str] = None) -> Optional[Dict[str, Any]]:
    """Update a grocery list"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        updates = []
        params = []
        
        if name is not None:
            updates.append("name = %s")
            params.append(name)
        
        if items is not None:
            updates.append("items = %s")
            params.append(items)
        
        if not updates:
            return None
        
        updates.append("updated_at = CURRENT_TIMESTAMP")
        params.append(list_id)
        
        query = f"UPDATE grocery_lists SET {', '.join(updates)} WHERE id = %s RETURNING id, name, items, owner_id, created_at, updated_at"
        
        cur.execute(query, params)
        grocery_list = cur.fetchone()
        conn.commit()
        return dict(grocery_list) if grocery_list else None
    finally:
        cur.close()
        conn.close()


def delete_grocery_list(list_id: int) -> bool:
    """Delete a grocery list"""
    conn = get_db_connection()
    cur = conn.cursor()
    try:
        cur.execute("DELETE FROM grocery_lists WHERE id = %s", (list_id,))
        deleted = cur.rowcount > 0
        conn.commit()
        return deleted
    finally:
        cur.close()
        conn.close()


def share_grocery_list(list_id: int, shared_with_user_id: int, can_edit: bool = False) -> Dict[str, Any]:
    """Share a grocery list with another user or update existing share permissions"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            INSERT INTO grocery_list_shares (list_id, shared_with_user_id, can_edit)
            VALUES (%s, %s, %s)
            ON CONFLICT (list_id, shared_with_user_id) 
            DO UPDATE SET can_edit = EXCLUDED.can_edit, shared_at = CURRENT_TIMESTAMP
            RETURNING id, list_id, shared_with_user_id, can_edit, shared_at
            """,
            (list_id, shared_with_user_id, can_edit)
        )
        share = cur.fetchone()
        conn.commit()
        return dict(share)
    finally:
        cur.close()
        conn.close()


def update_share_permission(list_id: int, shared_with_user_id: int, can_edit: bool) -> Optional[Dict[str, Any]]:
    """Update edit permission for an existing share"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            UPDATE grocery_list_shares
            SET can_edit = %s
            WHERE list_id = %s AND shared_with_user_id = %s
            RETURNING id, list_id, shared_with_user_id, can_edit, shared_at
            """,
            (can_edit, list_id, shared_with_user_id)
        )
        share = cur.fetchone()
        conn.commit()
        return dict(share) if share else None
    finally:
        cur.close()
        conn.close()


def unshare_grocery_list(list_id: int, user_id: int) -> bool:
    """Remove sharing access for a user"""
    conn = get_db_connection()
    cur = conn.cursor()
    try:
        cur.execute(
            "DELETE FROM grocery_list_shares WHERE list_id = %s AND shared_with_user_id = %s",
            (list_id, user_id)
        )
        deleted = cur.rowcount > 0
        conn.commit()
        return deleted
    finally:
        cur.close()
        conn.close()


def get_grocery_list_shares(list_id: int) -> List[Dict[str, Any]]:
    """Get all users a grocery list is shared with"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            SELECT gls.id, gls.list_id, gls.shared_with_user_id, gls.can_edit, gls.shared_at,
                   u.username, u.display_name, u.email
            FROM grocery_list_shares gls
            JOIN users u ON gls.shared_with_user_id = u.id
            WHERE gls.list_id = %s
            ORDER BY gls.shared_at DESC
            """,
            (list_id,)
        )
        shares = cur.fetchall()
        return [dict(row) for row in shares]
    finally:
        cur.close()
        conn.close()


def search_users(query: str, limit: int = 10) -> List[Dict[str, Any]]:
    """Search users by username or display_name for autocomplete"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        cur.execute(
            """
            SELECT id, username, display_name, email
            FROM users
            WHERE username ILIKE %s OR display_name ILIKE %s
            ORDER BY username
            LIMIT %s
            """,
            (f"%{query}%", f"%{query}%", limit)
        )
        users = cur.fetchall()
        return [dict(row) for row in users]
    finally:
        cur.close()
        conn.close()


def toggle_grocery_list_pin(list_id: int, user_id: int) -> Optional[Dict[str, Any]]:
    """Toggle pin status for a grocery list (owner only)"""
    conn = get_db_connection()
    cur = conn.cursor(cursor_factory=RealDictCursor)
    try:
        # Check if user is owner
        cur.execute(
            "SELECT id, is_pinned FROM grocery_lists WHERE id = %s AND owner_id = %s",
            (list_id, user_id)
        )
        result = cur.fetchone()
        if not result:
            return None
        
        # Toggle pin status
        new_pin_status = not result["is_pinned"]
        cur.execute(
            """
            UPDATE grocery_lists
            SET is_pinned = %s, updated_at = CURRENT_TIMESTAMP
            WHERE id = %s
            RETURNING id, name, items, owner_id, is_pinned, created_at, updated_at
            """,
            (new_pin_status, list_id)
        )
        updated = cur.fetchone()
        conn.commit()
        return dict(updated) if updated else None
    finally:
        cur.close()
        conn.close()