database.py

import sqlite3


DATABASE = "users.db"


def get_connection():
    return sqlite3.connect(DATABASE)


def create_table():

    connection = get_connection()
    cursor = connection.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT NOT NULL UNIQUE
        )
    """)

    connection.commit()
    connection.close()


def get_all_users():

    connection = get_connection()
    cursor = connection.cursor()

    cursor.execute("SELECT id, name, email FROM users")

    rows = cursor.fetchall()

    connection.close()

    users = []

    for row in rows:
        users.append({
            "id": row[0],
            "name": row[1],
            "email": row[2]
        })

    return users



def create_user(name, email):

    connection = get_connection()
    cursor = connection.cursor()

    try:
        cursor.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            (name, email)
        )
        
        connection.commit()
        
        user_id = cursor.lastrowid
        
        connection.close()
        
        return {
            "id": user_id,
            "name": name,
            "email": email
        }
        
    except sqlite3.IntegrityError:
        connection.close()
        return None
    
    
    
def get_user_by_id(user_id):

    connection = get_connection()
    cursor = connection.cursor()

    cursor.execute(
        "SELECT id, name, email FROM users WHERE id = ?",
        (user_id,)
    )

    row = cursor.fetchone()

    connection.close()

    if row is None:
        return None

    return {
        "id": row[0],
        "name": row[1],
        "email": row[2]
    }
    
    
def update_user(user_id, name, email):

    connection = get_connection()
    cursor = connection.cursor()

    try:
        cursor.execute(
            """
            UPDATE users
            SET name = ?, email = ?
            WHERE id = ?
            """,
            (name, email, user_id)
        )
        
        connection.commit()
        
        if cursor.rowcount == 0:
            connection.close()
            return None
        
        connection.close()
        
        return {
            "id": user_id,
            "name": name,
            "email": email
        }
    
    except sqlite3.IntegrityError:
        connection.close()
        return "duplicate_email"
    
def patch_user(user_id, name=None, email=None):

    connection = get_connection()
    cursor = connection.cursor()

    try:
        if name is not None and email is not None:
        
            cursor.execute(
                """
                UPDATE users
                SET name = ?, email = ?
                WHERE id = ?
                """,
                (name, email, user_id)
            )
        
        elif name is not None:
        
            cursor.execute(
                """
                UPDATE users
                SET name = ?
                WHERE id = ?
                """,
                (name, user_id)
            )
        
        elif email is not None:
        
            cursor.execute(
                """
                UPDATE users
                SET email = ?
                WHERE id = ?
                """,
                (email, user_id)
            )
        
        else:
            connection.close()
            return None
        
        connection.commit()
        
        if cursor.rowcount == 0:
            connection.close()
            return None
        
        cursor.execute(
            "SELECT id, name, email FROM users WHERE id = ?",
            (user_id,)
        )
        
        row = cursor.fetchone()
        
        connection.close()
        
        return {
            "id": row[0],
            "name": row[1],
            "email": row[2]
        }
        
    except sqlite3.IntegrityError:
        connection.close()
        return "duplicate_email"
    
    
def delete_user(user_id):

    connection = get_connection()
    cursor = connection.cursor()

    cursor.execute(
        "SELECT id, name, email FROM users WHERE id = ?",
        (user_id,)
    )

    row = cursor.fetchone()

    if row is None:
        connection.close()
        return None

    user = {
        "id": row[0],
        "name": row[1],
        "email": row[2]
    }

    cursor.execute(
        "DELETE FROM users WHERE id = ?",
        (user_id,)
    )

    connection.commit()
    connection.close()

    return user


def search_users(query):
    connection = get_connection()
    cursor = connection.cursor()

    cursor.execute(
        """
        SELECT id, name, email
        FROM users
        WHERE name LIKE ? OR email LIKE ?
        """,
        (f"%{query}%", f"%{query}%")
    )

    rows = cursor.fetchall()
    connection.close()

    users = []

    for row in rows:
        users.append({
            "id": row[0],
            "name": row[1],
            "email": row[2]
        })

    return users