Python連接數(shù)據(jù)庫的3種方法對比(SQLite\MySQL\PostgreSQL)

更新時間：2025年08月26日 10:24:07 作者：小莊-Python辦公

數(shù)據(jù)庫是一個有組織的數(shù)據(jù)集合,通常以表格形式存儲,這篇文章主要為大家詳細介紹了Python連接數(shù)據(jù)庫的3種方法,即SQLite,MySQL和PostgreSQL,希望對大家有所幫助

了解數(shù)據(jù)庫架構(gòu)

在開始使用Python連接數(shù)據(jù)庫之前，了解數(shù)據(jù)庫的基本架構(gòu)是非常重要的。數(shù)據(jù)庫是一個有組織的數(shù)據(jù)集合，通常以表格形式存儲。每個表由行和列組成，其中行代表記錄，列代表字段。

數(shù)據(jù)庫管理系統(tǒng)（DBMS）是用于管理數(shù)據(jù)庫的軟件。常見的DBMS包括：

SQLite：輕量級的文件型數(shù)據(jù)庫
MySQL：開源的關(guān)系型數(shù)據(jù)庫
PostgreSQL：功能強大的開源對象關(guān)系型數(shù)據(jù)庫

使用Python SQL庫連接到數(shù)據(jù)庫

Python提供了多種庫來連接不同類型的數(shù)據(jù)庫。以下是連接三種主要數(shù)據(jù)庫的方法：

SQLite

SQLite是Python內(nèi)置的數(shù)據(jù)庫，無需安裝額外的驅(qū)動程序。

import sqlite3

# 連接到SQLite數(shù)據(jù)庫
# 如果文件不存在，會自動創(chuàng)建
conn = sqlite3.connect('example.db')
cursor = conn.cursor()

print("成功連接到SQLite數(shù)據(jù)庫")

# 關(guān)閉連接
conn.close()

MySQL

連接MySQL需要安裝`mysql-connector-python`或`PyMySQL`庫。

import mysql.connector
from mysql.connector import Error

try:
    # 連接到MySQL數(shù)據(jù)庫
    connection = mysql.connector.connect(
        host='localhost',
        database='your_database',
        user='your_username',
        password='your_password'
    )
    
    if connection.is_connected():
        db_info = connection.get_server_info()
        print(f"成功連接到MySQL服務器，版本：{db_info}")
        
except Error as e:
    print(f"連接MySQL時發(fā)生錯誤：{e}")
    
finally:
    if connection.is_connected():
        connection.close()
        print("MySQL連接已關(guān)閉")

PostgreSQL數(shù)據(jù)庫

連接PostgreSQL需要安裝`psycopg2`庫。

import psycopg2
from psycopg2 import Error

try:
    # 連接到PostgreSQL數(shù)據(jù)庫
    connection = psycopg2.connect(
        host='localhost',
        database='your_database',
        user='your_username',
        password='your_password',
        port='5432'
    )
    
    cursor = connection.cursor()
    cursor.execute("SELECT version();")
    record = cursor.fetchone()
    print(f"成功連接到PostgreSQL：{record}")
    
except Error as e:
    print(f"連接PostgreSQL時發(fā)生錯誤：{e}")
    
finally:
    if connection:
        cursor.close()
        connection.close()
        print("PostgreSQL連接已關(guān)閉")

創(chuàng)建表

SQLite

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 創(chuàng)建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER
    )
''')

conn.commit()
print("SQLite表創(chuàng)建成功")
conn.close()

MySQL

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 創(chuàng)建表
create_table_query = '''
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        email VARCHAR(100) UNIQUE NOT NULL,
        age INT
    )
'''

cursor.execute(create_table_query)
connection.commit()
print("MySQL表創(chuàng)建成功")

cursor.close()
connection.close()

PostgreSQL數(shù)據(jù)庫

import psycopg2

connection = psycopg2.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 創(chuàng)建表
create_table_query = '''
    CREATE TABLE IF NOT EXISTS users (
        id SERIAL PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        email VARCHAR(100) UNIQUE NOT NULL,
        age INTEGER
    )
'''

cursor.execute(create_table_query)
connection.commit()
print("PostgreSQL表創(chuàng)建成功")

cursor.close()
connection.close()

插入記錄

SQLite

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 插入單條記錄
cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)", 
               ("張三", "zhangsan@example.com", 25))

# 插入多條記錄
users_data = [
    ("李四", "lisi@example.com", 30),
    ("王五", "wangwu@example.com", 28)
]
cursor.executemany("INSERT INTO users (name, email, age) VALUES (?, ?, ?)", users_data)

conn.commit()
print(f"成功插入{cursor.rowcount}條記錄")
conn.close()

MySQL

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 插入記錄
insert_query = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
user_data = ("張三", "zhangsan@example.com", 25)

cursor.execute(insert_query, user_data)
connection.commit()
print(f"成功插入{cursor.rowcount}條記錄")

cursor.close()
connection.close()

PostgreSQL數(shù)據(jù)庫

import psycopg2

connection = psycopg2.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 插入記錄
insert_query = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
user_data = ("張三", "zhangsan@example.com", 25)

cursor.execute(insert_query, user_data)
connection.commit()
print(f"成功插入{cursor.rowcount}條記錄")

cursor.close()
connection.close()

選擇記錄

SQLite

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 查詢所有記錄
cursor.execute("SELECT * FROM users")
records = cursor.fetchall()

print("所有用戶記錄：")
for row in records:
    print(f"ID: {row[0]}, 姓名: {row[1]}, 郵箱: {row[2]}, 年齡: {row[3]}")

# 條件查詢
cursor.execute("SELECT * FROM users WHERE age > ?", (25,))
young_users = cursor.fetchall()

print("\n年齡大于25的用戶：")
for user in young_users:
    print(f"姓名: {user[1]}, 年齡: {user[3]}")

conn.close()

MySQL

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 查詢記錄
select_query = "SELECT * FROM users WHERE age > %s"
cursor.execute(select_query, (25,))
records = cursor.fetchall()

print("查詢結(jié)果：")
for row in records:
    print(f"ID: {row[0]}, 姓名: {row[1]}, 郵箱: {row[2]}, 年齡: {row[3]}")

cursor.close()
connection.close()

PostgreSQL數(shù)據(jù)庫

import psycopg2

connection = psycopg2.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 查詢記錄
select_query = "SELECT * FROM users WHERE age > %s"
cursor.execute(select_query, (25,))
records = cursor.fetchall()

print("查詢結(jié)果：")
for row in records:
    print(f"ID: {row[0]}, 姓名: {row[1]}, 郵箱: {row[2]}, 年齡: {row[3]}")

cursor.close()
connection.close()

更新表記錄

SQLite

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 更新記錄
cursor.execute("UPDATE users SET age = ? WHERE name = ?", (26, "張三"))
conn.commit()

print(f"成功更新{cursor.rowcount}條記錄")
conn.close()

MySQL

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 更新記錄
update_query = "UPDATE users SET age = %s WHERE name = %s"
cursor.execute(update_query, (26, "張三"))
connection.commit()

print(f"成功更新{cursor.rowcount}條記錄")

cursor.close()
connection.close()

PostgreSQL數(shù)據(jù)庫

import psycopg2

connection = psycopg2.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 更新記錄
update_query = "UPDATE users SET age = %s WHERE name = %s"
cursor.execute(update_query, (26, "張三"))
connection.commit()

print(f"成功更新{cursor.rowcount}條記錄")

cursor.close()
connection.close()

刪除表記錄

SQLite

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 刪除記錄
cursor.execute("DELETE FROM users WHERE age < ?", (25,))
conn.commit()

print(f"成功刪除{cursor.rowcount}條記錄")
conn.close()

MySQL

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 刪除記錄
delete_query = "DELETE FROM users WHERE age < %s"
cursor.execute(delete_query, (25,))
connection.commit()

print(f"成功刪除{cursor.rowcount}條記錄")

cursor.close()
connection.close()

PostgreSQL數(shù)據(jù)庫

import psycopg2

connection = psycopg2.connect(
    host='localhost',
    database='your_database',
    user='your_username',
    password='your_password'
)

cursor = connection.cursor()

# 刪除記錄
delete_query = "DELETE FROM users WHERE age < %s"
cursor.execute(delete_query, (25,))
connection.commit()

print(f"成功刪除{cursor.rowcount}條記錄")

cursor.close()
connection.close()