table 에 crud 할 수 있는 python code

Windsurf 를 통해서 Claude 3.7 Sonnet 에게 테이블 명을 알려주고, CRUD 함수를 추가한 코드를 작성해 달라고 했습니다.

 

먼저, 테이블 생성 함수 create_user_test_table 를 정의할 때 테이블이 이미 존재하는제를 확인하면서 생성하네요.

그리고 데이터를 입력하고, 조회하고, 수정하고, 삭제하도록 했습니다.

 

import sys
from db_connect import connect_to_postgres, execute_query, close_connection

def create_user_test_table(connection):
    """
    Create the user_test table if it doesn't exist
    """
    create_table_query = """
    CREATE TABLE IF NOT EXISTS user_test (
        id SERIAL PRIMARY KEY,
        name VARCHAR(100) NOT NULL
    );
    """
    return execute_query(connection, create_table_query)

def insert_user(connection, user_id, name):
    """
    Insert a new user into the user_test table
    """
    insert_query = f"""
    INSERT INTO user_test (id, name)
    VALUES ({user_id}, '{name}')
    """
    return execute_query(connection, insert_query)

def update_user(connection, user_id, new_name):
    """
    Update a user's name in the user_test table
    """
    update_query = f"""
    UPDATE user_test
    SET name = '{new_name}'
    WHERE id = {user_id}
    """
    return execute_query(connection, update_query)

def select_all_users(connection):
    """
    Select all users from the user_test table
    """
    select_query = """
    SELECT id, name FROM user_test
    """
    return execute_query(connection, select_query)

def select_user_by_id(connection, user_id):
    """
    Select a user by ID from the user_test table
    """
    select_query = f"""
    SELECT id, name FROM user_test
    WHERE id = {user_id}
    """
    return execute_query(connection, select_query)

def delete_user(connection, user_id):
    """
    Delete a user from the user_test table by ID
    """
    delete_query = f"""
    DELETE FROM user_test
    WHERE id = {user_id}
    """
    return execute_query(connection, delete_query)

if __name__ == "__main__":
    # Connect to the database
    print("Connecting to PostgreSQL database...")
    conn = connect_to_postgres()
    
    if conn:
        try:
            # Create the user_test table if it doesn't exist
            create_user_test_table(conn)
            
            # Insert test data
            print("\nInserting test users...")
            insert_user(conn, 1, "John Doe")
            insert_user(conn, 2, "Jane Smith")
            insert_user(conn, 3, "Bob Johnson")
            
            # Select all users
            print("\nSelecting all users:")
            users = select_all_users(conn)
            if users:
                for user in users:
                    print(f"ID: {user[0]}, Name: {user[1]}")
            
            # Update a user
            print("\nUpdating user with ID 2...")
            update_user(conn, 2, "Jane Wilson")
            
            # Select the updated user
            print("\nSelecting updated user:")
            updated_user = select_user_by_id(conn, 2)
            if updated_user:
                for user in updated_user:
                    print(f"ID: {user[0]}, Name: {user[1]}")
            
            # Delete a user
            print("\nDeleting user with ID 3...")
            delete_user(conn, 3)
            
            # Select all users after deletion
            print("\nSelecting all users after deletion:")
            remaining_users = select_all_users(conn)
            if remaining_users:
                for user in remaining_users:
                    print(f"ID: {user[0]}, Name: {user[1]}")
            
        except Exception as e:
            print(f"Error: {e}")
        finally:
            # Close the connection
            close_connection(conn)