Python使用psycopg2操作PostgreSQL數(shù)據(jù)庫(kù)的完全指南

更新時(shí)間：2025年06月25日 09:18:47 作者：小龍?jiān)谏綎|

psycopg2 是 Python 中最流行的 PostgreSQL 數(shù)據(jù)庫(kù)適配器,它實(shí)現(xiàn)了 Python DB API 2.0 規(guī)范,同時(shí)提供了許多 PostgreSQL 特有的功能支持,下面我們來(lái)看看如何使用psycopg2操作PostgreSQL進(jìn)行連接和增刪改查操作吧

安裝

pip install psycopg2-binary

連接數(shù)據(jù)庫(kù)

使用連接參數(shù)直接連接

import psycopg2

# 基本連接參數(shù)
conn_params = {
    "dbname": "test",
    "user": "postgres",
    "password": "password",
    "host": "localhost",
    "port": "5432"
}

try:
    conn = psycopg2.connect(**conn_params)
    print("數(shù)據(jù)庫(kù)連接成功")
    
    # 執(zhí)行數(shù)據(jù)庫(kù)操作...
    
except psycopg2.Error as e:
    print(f"連接數(shù)據(jù)庫(kù)失敗: {e}")
finally:
    if 'conn' in locals():
        conn.close()

使用連接字符串 (DSN)

import psycopg2

# 連接字符串格式
dsn = "dbname=test user=postgres password=password host=localhost port=5432"

try:
    conn = psycopg2.connect(dsn)
    print("數(shù)據(jù)庫(kù)連接成功")
    
    # 執(zhí)行數(shù)據(jù)庫(kù)操作...
    
except psycopg2.Error as e:
    print(f"連接數(shù)據(jù)庫(kù)失敗: {e}")
finally:
    if 'conn' in locals():
        conn.close()

創(chuàng)建表

import psycopg2

conn = psycopg2.connect(host='127.0.0.1',
                        port='5432',
                        dbname="test", 
                        user="postgres",
                        password="password")

cur = conn.cursor()
cur.execute("""
                CREATE TABLE students (
                    id SERIAL PRIMARY KEY,
                    name VARCHAR(100) NOT NULL,
                    class VARCHAR(50) NOT NULL,
                    math_score NUMERIC(5, 2) CHECK (math_score >= 0 AND math_score <= 100),
                    english_score NUMERIC(5, 2) CHECK (english_score >= 0 AND english_score <= 100),
                    science_score NUMERIC(5, 2) CHECK (science_score >= 0 AND science_score <= 100),
                    history_score NUMERIC(5, 2) CHECK (history_score >= 0 AND history_score <= 100),
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
conn.commit()

插入隨機(jī)數(shù)據(jù)

import psycopg2
import random
from faker import Faker

conn = psycopg2.connect(host='127.0.0.1',
                        port='5432',
                        dbname="test", 
                        user="postgres",
                        password="password")


cursor = conn.cursor()
fake = Faker('zh_CN')
        
# 準(zhǔn)備隨機(jī)數(shù)據(jù)
classes = ['一年一班', '一年二班', '二年一班', '二年二班', '三年一班', '三年二班']
students = []
count = 10

for _ in range(count):
    name = fake.name()
    class_name = random.choice(classes)
    math = round(random.uniform(50, 100), 1)
    english = round(random.uniform(50, 100), 1)
    science = round(random.uniform(50, 100), 1)
    history = round(random.uniform(50, 100), 1)
    
    students.append((name, class_name, math, english, science, history))

# 插入數(shù)據(jù)
cursor.executemany("""
    INSERT INTO students 
    (name, class, math_score, english_score, science_score, history_score)
    VALUES (%s, %s, %s, %s, %s, %s)
""", students)

conn.commit()
print(f"成功插入 {count} 條隨機(jī)學(xué)生數(shù)據(jù)")

查詢數(shù)據(jù)

def get_students_as_dict(dbname, user, password, host='localhost', port=5432):
    """以字典形式返回學(xué)生數(shù)據(jù)"""
    try:
        conn = psycopg2.connect(host=host,
                        port=port,
                        dbname=dbname, 
                        user=user,
                        password=password)
        
        # 使用DictCursor可以返回字典形式的結(jié)果
        cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
        
        cursor.execute("""
            SELECT id, name, class, 
                   math_score, english_score, 
                   science_score, history_score
            FROM students
            LIMIT 3
        """)
        
        print("\n字典形式的學(xué)生數(shù)據(jù):")
        for row in cursor:
            # 可以直接通過列名訪問
            print(dict(row))
            
    except psycopg2.Error as e:
        print(f"查詢數(shù)據(jù)時(shí)出錯(cuò): {e}")
    finally:
        if conn:
            conn.close()

更改數(shù)據(jù)

import psycopg2

def update_student_score(dbname, user, password, student_id, subject, new_score, host='localhost', port=5432):
    """
    更新指定學(xué)生的單科成績(jī)
    
    參數(shù):
        student_id: 學(xué)生ID
        subject: 科目名稱 ('math_score', 'english_score', 'science_score', 'history_score')
        new_score: 新成績(jī) (0-100)
    """
    try:
        conn = psycopg2.connect(
            dbname=dbname,
            user=user,
            password=password,
            host=host,
            port=port
        )
        cursor = conn.cursor()
        
        # 驗(yàn)證科目名稱
        valid_subjects = ['math_score', 'english_score', 'science_score', 'history_score']
        if subject not in valid_subjects:
            raise ValueError(f"無(wú)效科目名稱，必須是: {', '.join(valid_subjects)}")
        
        # 執(zhí)行更新
        cursor.execute(f"""
            UPDATE students
            SET {subject} = %s
            WHERE id = %s
            RETURNING id, name, {subject}
        """, (new_score, student_id))
        
        updated_row = cursor.fetchone()
        
        if updated_row:
            conn.commit()
            print(f"成功更新學(xué)生 {updated_row[1]} (ID: {updated_row[0]}) 的{subject.replace('_', '')}為 {updated_row[2]}")
        else:
            print(f"未找到ID為 {student_id} 的學(xué)生")
        
    except psycopg2.Error as e:
        print(f"更新數(shù)據(jù)時(shí)出錯(cuò): {e}")
        conn.rollback()
    except ValueError as e:
        print(f"參數(shù)錯(cuò)誤: {e}")
    finally:
        if conn:
            conn.close()

# 使用示例
update_student_score(
    dbname='test',
    user='postgres',
    password='password',
    student_id=1,  # 要更新的學(xué)生ID
    subject='math_score',  # 要更新的科目
    new_score=95.5,  # 新成績(jī)
    host='localhost'
)

刪除數(shù)據(jù)

import psycopg2

def delete_student_by_id(dbname, user, password, student_id, host='localhost', port=5432):
    """根據(jù)學(xué)生ID刪除記錄"""
    try:
        conn = psycopg2.connect(
            dbname=dbname,
            user=user,
            password=password,
            host=host,
            port=port
        )
        cursor = conn.cursor()
        
        # 先查詢學(xué)生是否存在
        cursor.execute("""
            SELECT name, class FROM students WHERE id = %s
        """, (student_id,))
        
        student = cursor.fetchone()
        
        if not student:
            print(f"未找到ID為 {student_id} 的學(xué)生")
            return
        
        # 確認(rèn)刪除
        confirm = input(f"確定要?jiǎng)h除學(xué)生 {student[0]} (班級(jí): {student[1]}) 嗎? (y/n): ")
        if confirm.lower() != 'y':
            print("刪除操作已取消")
            return
        
        # 執(zhí)行刪除
        cursor.execute("""
            DELETE FROM students
            WHERE id = %s
            RETURNING id, name, class
        """, (student_id,))
        
        deleted_student = cursor.fetchone()
        
        if deleted_student:
            conn.commit()
            print(f"已刪除學(xué)生: ID {deleted_student[0]}, 姓名: {deleted_student[1]}, 班級(jí): {deleted_student[2]}")
        else:
            print("刪除失敗，未找到該學(xué)生")
        
    except psycopg2.Error as e:
        print(f"刪除數(shù)據(jù)時(shí)出錯(cuò): {e}")
        conn.rollback()
    finally:
        if conn:
            conn.close()

# 使用示例
delete_student_by_id(
    dbname='test',
    user='postgres',
    password='password',
    student_id=3,  # 要?jiǎng)h除的學(xué)生ID
    host='localhost'
)

到此這篇關(guān)于Python使用psycopg2操作PostgreSQL數(shù)據(jù)庫(kù)的完全指南的文章就介紹到這了,更多相關(guān)Python psycopg2操作PostgreSQL內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: