# 安裝PyMySQL（推薦）
pip install pymysql
# 若需安裝MySQLdb（僅Python 2.x，不推薦）
# pip install mysqldb

import pymysql
from pymysql import Error
def mysql_demo():
    # 1. 數(shù)據(jù)庫(kù)連接參數(shù)（根據(jù)實(shí)際環(huán)境修改）
    db_config = {
        "host": "localhost",  # 數(shù)據(jù)庫(kù)地址（本地為localhost）
        "port": 3306,         # 數(shù)據(jù)庫(kù)端口（MySQL默認(rèn)3306）
        "user": "root",       # 用戶(hù)名
        "password": "123456", # 密碼
        "database": "test_db" # 要連接的數(shù)據(jù)庫(kù)名
    }
    # 初始化連接和游標(biāo)
    connection = None
    cursor = None
    try:
        # 2. 建立數(shù)據(jù)庫(kù)連接
        connection = pymysql.connect(**db_config)
        if connection.open:
            print("? 成功連接到MySQL數(shù)據(jù)庫(kù)")
        # 3. 創(chuàng)建游標(biāo)對(duì)象（用于執(zhí)行SQL語(yǔ)句）
        cursor = connection.cursor()
        # 4. 執(zhí)行操作（建表→插入→查詢(xún)→更新→刪除）
        # 4.1 建表（若表不存在）
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS user_info (
            id INT AUTO_INCREMENT PRIMARY KEY,  # 自增主鍵
            name VARCHAR(50) NOT NULL,         # 姓名（非空）
            age INT,                           # 年齡
            gender VARCHAR(10),                # 性別
            create_time DATETIME DEFAULT CURRENT_TIMESTAMP  # 創(chuàng)建時(shí)間（默認(rèn)當(dāng)前時(shí)間）
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
        """
        cursor.execute(create_table_sql)
        print("? 表創(chuàng)建成功（若不存在）")
        # 4.2 插入數(shù)據(jù)（單條插入）
        insert_sql = "INSERT INTO user_info (name, age, gender) VALUES (%s, %s, %s)"
        insert_data = ("張三", 25, "男")  # 占位符%s，避免SQL注入（重點(diǎn)?。?
        cursor.execute(insert_sql, insert_data)
        connection.commit()  # 插入/更新/刪除操作需提交事務(wù)
        print(f"? 插入數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 4.3 批量插入數(shù)據(jù)（效率高于單條插入）
        batch_insert_sql = "INSERT INTO user_info (name, age, gender) VALUES (%s, %s, %s)"
        batch_data = [("李四", 28, "男"), ("王五", 23, "女"), ("趙六", 30, "男")]
        cursor.executemany(batch_insert_sql, batch_data)
        connection.commit()
        print(f"? 批量插入數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 4.4 查詢(xún)數(shù)據(jù)（單條查詢(xún)+多條查詢(xún)）
        # 單條查詢(xún)（根據(jù)id查詢(xún)）
        select_one_sql = "SELECT * FROM user_info WHERE id = %s"
        cursor.execute(select_one_sql, (1,))  # 注意：參數(shù)需是元組，即使只有一個(gè)值
        result_one = cursor.fetchone()  # 獲取單條結(jié)果（返回元組，無(wú)結(jié)果返回None）
        print(f"? 單條查詢(xún)結(jié)果：{result_one}")
        # 多條查詢(xún)（查詢(xún)所有數(shù)據(jù)）
        select_all_sql = "SELECT * FROM user_info"
        cursor.execute(select_all_sql)
        result_all = cursor.fetchall()  # 獲取所有結(jié)果（返回列表，每個(gè)元素是元組）
        print(f"? 所有查詢(xún)結(jié)果（共{len(result_all)}條）：")
        for row in result_all:
            print(f"id: {row[0]}, 姓名: {row[1]}, 年齡: {row[2]}, 性別: {row[3]}, 創(chuàng)建時(shí)間: {row[4]}")
        # 4.5 更新數(shù)據(jù)
        update_sql = "UPDATE user_info SET age = %s WHERE name = %s"
        update_data = (26, "張三")
        cursor.execute(update_sql, update_data)
        connection.commit()
        print(f"? 更新數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 4.6 刪除數(shù)據(jù)
        delete_sql = "DELETE FROM user_info WHERE name = %s"
        delete_data = ("趙六",)
        cursor.execute(delete_sql, delete_data)
        connection.commit()
        print(f"? 刪除數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
    except Error as e:
        # 異常處理（捕獲連接、SQL執(zhí)行等異常）
        print(f"? 數(shù)據(jù)庫(kù)操作異常：{e}")
        # 若有事務(wù)未提交，回滾（避免數(shù)據(jù)錯(cuò)亂）
        if connection:
            connection.rollback()
    finally:
        # 5. 關(guān)閉資源（先關(guān)游標(biāo)，再關(guān)連接，順序不可顛倒）
        if cursor:
            cursor.close()
            print("? 游標(biāo)已關(guān)閉")
        if connection and connection.open:
            connection.close()
            print("? 數(shù)據(jù)庫(kù)連接已關(guān)閉")
# 執(zhí)行示例
if __name__ == "__main__":
    mysql_demo()

import sqlite3
from sqlite3 import Error
def sqlite_demo():
    # 1. 數(shù)據(jù)庫(kù)連接參數(shù)（SQLite無(wú)需用戶(hù)名密碼，僅需指定數(shù)據(jù)庫(kù)文件路徑）
    db_path = "test_sqlite.db"  # 數(shù)據(jù)庫(kù)文件存儲(chǔ)路徑（不存在則自動(dòng)創(chuàng)建）
    # 初始化連接和游標(biāo)
    connection = None
    cursor = None
    try:
        # 2. 建立數(shù)據(jù)庫(kù)連接（SQLite連接是文件連接，不存在則創(chuàng)建文件）
        connection = sqlite3.connect(db_path)
        print("? 成功連接到SQLite數(shù)據(jù)庫(kù)（文件：test_sqlite.db）")
        # 3. 創(chuàng)建游標(biāo)對(duì)象
        cursor = connection.cursor()
        # 4. 執(zhí)行操作（建表→插入→查詢(xún)→更新→刪除）
        # 4.1 建表（若表不存在）
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS book_info (
            id INTEGER PRIMARY KEY AUTOINCREMENT,  # SQLite自增用INTEGER，而非INT
            book_name TEXT NOT NULL,               # 書(shū)名（非空）
            author TEXT,                           # 作者
            price REAL,                            # 價(jià)格（浮點(diǎn)型）
            publish_time TEXT                      # 出版時(shí)間（字符串格式）
        );
        """
        cursor.execute(create_table_sql)
        print("? 表創(chuàng)建成功（若不存在）")
        # 4.2 插入數(shù)據(jù)（單條+批量）
        # 單條插入
        insert_sql = "INSERT INTO book_info (book_name, author, price, publish_time) VALUES (?, ?, ?, ?)"
        # SQLite占位符用?，而非%s，注意區(qū)別！
        insert_data = ("Python編程：從入門(mén)到精通", "張三", 89.9, "2023-01-01")
        cursor.execute(insert_sql, insert_data)
        # 批量插入
        batch_data = [
            ("Java編程思想", "李四", 99.9, "2022-05-01"),
            ("SQL必知必會(huì)", "王五", 59.9, "2023-03-15"),
            ("Redis實(shí)戰(zhàn)", "趙六", 79.9, "2023-06-20")
        ]
        cursor.executemany(insert_sql, batch_data)
        connection.commit()  # SQLite同樣需要提交事務(wù)
        print(f"? 插入數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 4.3 查詢(xún)數(shù)據(jù)
        # 單條查詢(xún)（根據(jù)id）
        select_one_sql = "SELECT * FROM book_info WHERE id = ?"
        cursor.execute(select_one_sql, (1,))
        result_one = cursor.fetchone()
        print(f"? 單條查詢(xún)結(jié)果：{result_one}")
        # 多條查詢(xún)（查詢(xún)價(jià)格>60的書(shū)籍）
        select_condition_sql = "SELECT book_name, author, price FROM book_info WHERE price > ?"
        cursor.execute(select_condition_sql, (60,))
        result_condition = cursor.fetchall()
        print(f"? 價(jià)格>60的書(shū)籍（共{len(result_condition)}條）：")
        for row in result_condition:
            print(f"書(shū)名: {row[0]}, 作者: {row[1]}, 價(jià)格: {row[2]}元")
        # 4.4 更新數(shù)據(jù)
        update_sql = "UPDATE book_info SET price = ? WHERE book_name = ?"
        update_data = (79.9, "Python編程：從入門(mén)到精通")
        cursor.execute(update_sql, update_data)
        connection.commit()
        print(f"? 更新數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 4.5 刪除數(shù)據(jù)
        delete_sql = "DELETE FROM book_info WHERE author = ?"
        delete_data = ("趙六",)
        cursor.execute(delete_sql, delete_data)
        connection.commit()
        print(f"? 刪除數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
    except Error as e:
        print(f"? 數(shù)據(jù)庫(kù)操作異常：{e}")
        if connection:
            connection.rollback()
    finally:
        # 關(guān)閉資源
        if cursor:
            cursor.close()
            print("? 游標(biāo)已關(guān)閉")
        if connection:
            connection.close()
            print("? 數(shù)據(jù)庫(kù)連接已關(guān)閉")
# 執(zhí)行示例
if __name__ == "__main__":
    sqlite_demo()

# 安裝psycopg2二進(jìn)制版本（推薦）
pip install psycopg2-binary
# 安裝編譯版本（需提前安裝PostgreSQL開(kāi)發(fā)環(huán)境，不推薦）
# pip install psycopg2

import psycopg2
from psycopg2 import Error
def postgresql_demo():
    # 1. 數(shù)據(jù)庫(kù)連接參數(shù)
    db_config = {
        "host": "localhost",
        "port": 5432,          # PostgreSQL默認(rèn)端口5432（區(qū)別于MySQL的3306）
        "user": "postgres",    # PostgreSQL默認(rèn)用戶(hù)名postgres
        "password": "123456",
        "database": "test_pg"
    }
    connection = None
    cursor = None
    try:
        # 2. 建立連接
        connection = psycopg2.connect(**db_config)
        cursor = connection.cursor()
        print("? 成功連接到PostgreSQL數(shù)據(jù)庫(kù)")
        # 3. 建表（支持JSON類(lèi)型，PostgreSQL特色）
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS student_info (
            id SERIAL PRIMARY KEY,  # PostgreSQL自增用SERIAL
            name VARCHAR(50) NOT NULL,
            age INT,
            info JSONB  # 支持JSONB類(lèi)型（高效存儲(chǔ)JSON數(shù)據(jù)）
        );
        """
        cursor.execute(create_table_sql)
        print("? 表創(chuàng)建成功（若不存在）")
        # 4. 插入數(shù)據(jù)（支持JSON類(lèi)型數(shù)據(jù)）
        insert_sql = "INSERT INTO student_info (name, age, info) VALUES (%s, %s, %s)"
        # PostgreSQL占位符同樣是%s，與PyMySQL一致
        insert_data = ("張三", 20, '{"class": "一班", "score": 90}')  # JSON數(shù)據(jù)傳字符串
        cursor.execute(insert_sql, insert_data)
        # 批量插入
        batch_data = [
            ("李四", 21, '{"class": "二班", "score": 85}'),
            ("王五", 19, '{"class": "一班", "score": 95}')
        ]
        cursor.executemany(insert_sql, batch_data)
        connection.commit()
        print(f"? 插入數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 5. 查詢(xún)數(shù)據(jù)（支持JSON字段查詢(xún)，PostgreSQL特色）
        select_sql = "SELECT * FROM student_info WHERE info->>'class' = %s"
        cursor.execute(select_sql, ("一班",))  # 查詢(xún)class為一班的學(xué)生
        result = cursor.fetchall()
        print(f"? 一班學(xué)生（共{len(result)}條）：")
        for row in result:
            print(f"id: {row[0]}, 姓名: {row[1]}, 年齡: {row[2]}, 信息: {row[3]}")
        # 6. 更新、刪除操作（與PyMySQL一致）
        update_sql = "UPDATE student_info SET age = %s WHERE name = %s"
        cursor.execute(update_sql, (22, "張三"))
        connection.commit()
        delete_sql = "DELETE FROM student_info WHERE name = %s"
        cursor.execute(delete_sql, ("李四",))
        connection.commit()
    except Error as e:
        print(f"? 數(shù)據(jù)庫(kù)操作異常：{e}")
        if connection:
            connection.rollback()
    finally:
        # 關(guān)閉資源
        if cursor:
            cursor.close()
        if connection:
            connection.close()
            print("? 數(shù)據(jù)庫(kù)連接已關(guān)閉")
if __name__ == "__main__":
    postgresql_demo()

# 安裝pymongo
pip install pymongo
# 若需支持異步操作（如FastAPI項(xiàng)目），安裝motor（pymongo的異步版本）
# pip install motor

from pymongo import MongoClient
from pymongo.errors import PyMongoError
def mongodb_demo():
    # 1. 數(shù)據(jù)庫(kù)連接參數(shù)
    mongo_url = "mongodb://localhost:27017/"  # MongoDB默認(rèn)地址+端口
    db_name = "test_mongo"  # 數(shù)據(jù)庫(kù)名（不存在則自動(dòng)創(chuàng)建）
    collection_name = "user"  # 集合名（對(duì)應(yīng)關(guān)系型數(shù)據(jù)庫(kù)的表，不存在則自動(dòng)創(chuàng)建）
    client = None
    try:
        # 2. 建立連接（創(chuàng)建MongoClient對(duì)象）
        client = MongoClient(mongo_url)
        print("? 成功連接到MongoDB數(shù)據(jù)庫(kù)")
        # 3. 獲取數(shù)據(jù)庫(kù)對(duì)象
        db = client[db_name]
        # 4. 獲取集合對(duì)象（對(duì)應(yīng)表）
        collection = db[collection_name]
        # 5. 執(zhí)行操作（插入→查詢(xún)→更新→刪除）
        # 5.1 插入文檔（單條+批量）
        # 單條插入（文檔是字典格式，對(duì)應(yīng)JSON）
        single_doc = {
            "name": "張三",
            "age": 25,
            "gender": "男",
            "info": {"city": "北京", "job": "程序員"}
        }
        insert_result = collection.insert_one(single_doc)
        print(f"? 單條插入成功，文檔ID：{insert_result.inserted_id}")
        # 批量插入（傳入列表，每個(gè)元素是字典）
        batch_docs = [
            {"name": "李四", "age": 28, "gender": "男", "info": {"city": "上海", "job": "產(chǎn)品經(jīng)理"}},
            {"name": "王五", "age": 23, "gender": "女", "info": {"city": "廣州", "job": "設(shè)計(jì)師"}},
            {"name": "趙六", "age": 30, "gender": "男", "info": {"city": "深圳", "job": "測(cè)試工程師"}}
        ]
        batch_result = collection.insert_many(batch_docs)
        print(f"? 批量插入成功，文檔IDs：{batch_result.inserted_ids}")
        # 5.2 查詢(xún)文檔（單條+多條+條件查詢(xún)）
        # 單條查詢(xún)（根據(jù)條件查詢(xún)，find_one返回第一個(gè)匹配文檔）
        single_query = collection.find_one({"name": "張三"})
        print(f"? 單條查詢(xún)結(jié)果：{single_query}")
        # 多條查詢(xún)（查詢(xún)所有g(shù)ender為男的文檔，返回游標(biāo)，需遍歷）
        multi_query = collection.find({"gender": "男"})
        print("? 所有男性用戶(hù)：")
        for doc in multi_query:
            print(doc)
        # 條件查詢(xún)（查詢(xún)age>25的文檔，只返回name和age字段，隱藏_id）
        condition_query = collection.find(
            {"age": {"$gt": 25}},  # 條件：age>25（MongoDB條件語(yǔ)法）
            {"name": 1, "age": 1, "_id": 0}  # 1：顯示，0：隱藏
        )
        print("? 年齡>25的用戶(hù)（僅顯示姓名和年齡）：")
        for doc in condition_query:
            print(doc)
        # 5.3 更新文檔（單條+多條）
        # 單條更新（更新name為張三的文檔，將age改為26）
        update_single = collection.update_one(
            {"name": "張三"},  # 查詢(xún)條件
            {"$set": {"age": 26, "info.city": "杭州"}}  # 更新操作：$set表示修改指定字段
        )
        print(f"? 單條更新成功，匹配行數(shù)：{update_single.matched_count}，修改行數(shù)：{update_single.modified_count}")
        # 多條更新（更新所有g(shù)ender為男的文檔，將age加1）
        update_multi = collection.update_many(
            {"gender": "男"},
            {"$inc": {"age": 1}}  # $inc表示自增
        )
        print(f"? 多條更新成功，匹配行數(shù)：{update_multi.matched_count}，修改行數(shù)：{update_multi.modified_count}")
        # 5.4 刪除文檔（單條+多條）
        # 單條刪除（刪除name為趙六的文檔）
        delete_single = collection.delete_one({"name": "趙六"})
        print(f"? 單條刪除成功，刪除行數(shù)：{delete_single.deleted_count}")
        # 多條刪除（刪除age>30的文檔）
        delete_multi = collection.delete_many({"age": {"$gt": 30}})
        print(f"? 多條刪除成功，刪除行數(shù)：{delete_multi.deleted_count}")
    except PyMongoError as e:
        print(f"? MongoDB操作異常：{e}")
    finally:
        # 關(guān)閉連接
        if client:
            client.close()
            print("? MongoDB連接已關(guān)閉")
if __name__ == "__main__":
    mongodb_demo()

# 安裝redis-py（推薦最新版本）
pip install redis
# 若需異步操作（如異步Web項(xiàng)目），安裝aioredis
# pip install aioredis

import redis
from redis.exceptions import RedisError
def redis_demo():
    # 1. 數(shù)據(jù)庫(kù)連接參數(shù)
    redis_config = {
        "host": "localhost",
        "port": 6379,          # Redis默認(rèn)端口6379
        "password": "",        # 若有密碼，傳入密碼字符串
        "db": 0,               # 選擇數(shù)據(jù)庫(kù)（Redis默認(rèn)有16個(gè)數(shù)據(jù)庫(kù)，編號(hào)0-15）
        "decode_responses": True  # 關(guān)鍵：設(shè)置為T(mén)rue，返回字符串（默認(rèn)返回bytes類(lèi)型）
    }
    r = None
    try:
        # 2. 建立連接（創(chuàng)建Redis對(duì)象，推薦使用連接池）
        # 方式1：直接創(chuàng)建連接（簡(jiǎn)單場(chǎng)景）
        r = redis.Redis(**redis_config)
        # 方式2：使用連接池（生產(chǎn)環(huán)境推薦，避免頻繁創(chuàng)建/關(guān)閉連接）
        # pool = redis.ConnectionPool(**redis_config)
        # r = redis.Redis(connection_pool=pool)
        # 驗(yàn)證連接（ping()返回True表示連接成功）
        if r.ping():
            print("? 成功連接到Redis數(shù)據(jù)庫(kù)")
        # 3. 執(zhí)行操作（常用數(shù)據(jù)結(jié)構(gòu)：字符串、哈希、列表、集合）
        # 3.1 字符串類(lèi)型（String）：最基礎(chǔ)的鍵值對(duì)，適合緩存、計(jì)數(shù)器
        # 設(shè)置值（key=name, value=張三，過(guò)期時(shí)間60秒）
        r.set("name", "張三", ex=60)
        # 獲取值
        name = r.get("name")
        print(f"? 字符串查詢(xún)：name = {name}")
        # 自增（計(jì)數(shù)器）
        r.incr("count")  # count從0變?yōu)?
        r.incr("count", 2)  # count從1變?yōu)?
        count = r.get("count")
        print(f"? 計(jì)數(shù)器：count = {count}")
        # 刪除值
        r.delete("count")
        print(f"? 字符串刪除：count是否存在 = {r.exists('count')}")  # exists返回0表示不存在
        # 3.2 哈希類(lèi)型（Hash）：適合存儲(chǔ)對(duì)象（如用戶(hù)信息），鍵值對(duì)嵌套
        # 設(shè)置哈希值（key=user:1，字段name、age、gender）
        r.hset("user:1", mapping={
            "name": "李四",
            "age": 28,
            "gender": "男"
        })
        # 獲取單個(gè)字段值
        user_name = r.hget("user:1", "name")
        print(f"? 哈希查詢(xún)（單個(gè)字段）：user:1.name = {user_name}")
        # 獲取所有字段和值
        user_all = r.hgetall("user:1")
        print(f"? 哈希查詢(xún)（所有字段）：user:1 = {user_all}")
        # 獲取所有字段名
        user_fields = r.hkeys("user:1")
        print(f"? 哈希查詢(xún)（所有字段名）：user:1.fields = {user_fields}")
        # 3.3 列表類(lèi)型（List）：有序、可重復(fù)，適合消息隊(duì)列、排行榜
        # 向列表左側(cè)添加值
        r.lpush("book_list", "Python編程", "Java編程", "SQL必知必會(huì)")
        # 向列表右側(cè)添加值
        r.rpush("book_list", "Redis實(shí)戰(zhàn)")
        # 獲取列表長(zhǎng)度
        book_len = r.llen("book_list")
        print(f"? 列表長(zhǎng)度：book_list.length = {book_len}")
        # 獲取列表所有值（0表示第一個(gè)，-1表示最后一個(gè)）
        book_all = r.lrange("book_list", 0, -1)
        print(f"? 列表查詢(xún)（所有值）：book_list = {book_all}")
        # 彈出列表左側(cè)值（消息隊(duì)列常用）
        book_pop = r.lpop("book_list")
        print(f"? 列表彈出（左側(cè)）：彈出值 = {book_pop}，剩余列表 = {r.lrange('book_list', 0, -1)}")
        # 3.4 集合類(lèi)型（Set）：無(wú)序、不可重復(fù)，適合去重、交集/并集查詢(xún)
        # 向集合添加值
        r.sadd("user_set", "張三", "李四", "王五", "張三")  # 重復(fù)值會(huì)自動(dòng)去重
        # 獲取集合所有值
        user_set_all = r.smembers("user_set")
        print(f"? 集合查詢(xún)（所有值）：user_set = {user_set_all}")
        # 判斷值是否在集合中
        is_exist = r.sismember("user_set", "張三")
        print(f"? 集合判斷：張三是否在user_set中 = {is_exist}")
        # 獲取集合長(zhǎng)度
        user_set_len = r.scard("user_set")
        print(f"? 集合長(zhǎng)度：user_set.length = {user_set_len}")
        # 3.5 事務(wù)操作（Redis事務(wù)：批量執(zhí)行，要么全部成功，要么全部失敗）
        pipe = r.pipeline()  # 創(chuàng)建管道（事務(wù)載體）
        pipe.set("tx1", "test1")
        pipe.set("tx2", "test2")
        pipe.incr("tx3")
        pipe.execute()  # 執(zhí)行事務(wù)
        print(f"? 事務(wù)執(zhí)行成功：tx1 = {r.get('tx1')}, tx2 = {r.get('tx2')}, tx3 = {r.get('tx3')}")
    except RedisError as e:
        print(f"? Redis操作異常：{e}")
    finally:
        # 關(guān)閉連接（使用連接池時(shí)，無(wú)需手動(dòng)關(guān)閉，連接池會(huì)自動(dòng)管理）
        if r:
            r.close()
            print("? Redis連接已關(guān)閉")
if __name__ == "__main__":
    redis_demo()