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

import pymysql
from pymysql import Error
def mysql_demo():
    # 1. 數(shù)據(jù)庫連接參數(shù)（根據(jù)實際環(huán)境修改）
    db_config = {
        "host": "localhost",  # 數(shù)據(jù)庫地址（本地為localhost）
        "port": 3306,         # 數(shù)據(jù)庫端口（MySQL默認3306）
        "user": "root",       # 用戶名
        "password": "123456", # 密碼
        "database": "test_db" # 要連接的數(shù)據(jù)庫名
    }
    # 初始化連接和游標
    connection = None
    cursor = None
    try:
        # 2. 建立數(shù)據(jù)庫連接
        connection = pymysql.connect(**db_config)
        if connection.open:
            print("? 成功連接到MySQL數(shù)據(jù)庫")
        # 3. 創(chuàng)建游標對象（用于執(zhí)行SQL語句）
        cursor = connection.cursor()
        # 4. 執(zhí)行操作（建表→插入→查詢→更新→刪除）
        # 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)建時間（默認當前時間）
        ) 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注入（重點?。?
        cursor.execute(insert_sql, insert_data)
        connection.commit()  # 插入/更新/刪除操作需提交事務
        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 查詢數(shù)據(jù)（單條查詢+多條查詢）
        # 單條查詢（根據(jù)id查詢）
        select_one_sql = "SELECT * FROM user_info WHERE id = %s"
        cursor.execute(select_one_sql, (1,))  # 注意：參數(shù)需是元組，即使只有一個值
        result_one = cursor.fetchone()  # 獲取單條結果（返回元組，無結果返回None）
        print(f"? 單條查詢結果：{result_one}")
        # 多條查詢（查詢所有數(shù)據(jù)）
        select_all_sql = "SELECT * FROM user_info"
        cursor.execute(select_all_sql)
        result_all = cursor.fetchall()  # 獲取所有結果（返回列表，每個元素是元組）
        print(f"? 所有查詢結果（共{len(result_all)}條）：")
        for row in result_all:
            print(f"id: {row[0]}, 姓名: {row[1]}, 年齡: {row[2]}, 性別: {row[3]}, 創(chuàng)建時間: {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ù)庫操作異常：{e}")
        # 若有事務未提交，回滾（避免數(shù)據(jù)錯亂）
        if connection:
            connection.rollback()
    finally:
        # 5. 關閉資源（先關游標，再關連接，順序不可顛倒）
        if cursor:
            cursor.close()
            print("? 游標已關閉")
        if connection and connection.open:
            connection.close()
            print("? 數(shù)據(jù)庫連接已關閉")
# 執(zhí)行示例
if __name__ == "__main__":
    mysql_demo()

import sqlite3
from sqlite3 import Error
def sqlite_demo():
    # 1. 數(shù)據(jù)庫連接參數(shù)（SQLite無需用戶名密碼，僅需指定數(shù)據(jù)庫文件路徑）
    db_path = "test_sqlite.db"  # 數(shù)據(jù)庫文件存儲路徑（不存在則自動創(chuàng)建）
    # 初始化連接和游標
    connection = None
    cursor = None
    try:
        # 2. 建立數(shù)據(jù)庫連接（SQLite連接是文件連接，不存在則創(chuàng)建文件）
        connection = sqlite3.connect(db_path)
        print("? 成功連接到SQLite數(shù)據(jù)庫（文件：test_sqlite.db）")
        # 3. 創(chuàng)建游標對象
        cursor = connection.cursor()
        # 4. 執(zhí)行操作（建表→插入→查詢→更新→刪除）
        # 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,               # 書名（非空）
            author TEXT,                           # 作者
            price REAL,                            # 價格（浮點型）
            publish_time TEXT                      # 出版時間（字符串格式）
        );
        """
        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編程：從入門到精通", "張三", 89.9, "2023-01-01")
        cursor.execute(insert_sql, insert_data)
        # 批量插入
        batch_data = [
            ("Java編程思想", "李四", 99.9, "2022-05-01"),
            ("SQL必知必會", "王五", 59.9, "2023-03-15"),
            ("Redis實戰(zhàn)", "趙六", 79.9, "2023-06-20")
        ]
        cursor.executemany(insert_sql, batch_data)
        connection.commit()  # SQLite同樣需要提交事務
        print(f"? 插入數(shù)據(jù)成功，影響行數(shù)：{cursor.rowcount}")
        # 4.3 查詢數(shù)據(jù)
        # 單條查詢（根據(jù)id）
        select_one_sql = "SELECT * FROM book_info WHERE id = ?"
        cursor.execute(select_one_sql, (1,))
        result_one = cursor.fetchone()
        print(f"? 單條查詢結果：{result_one}")
        # 多條查詢（查詢價格>60的書籍）
        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"? 價格>60的書籍（共{len(result_condition)}條）：")
        for row in result_condition:
            print(f"書名: {row[0]}, 作者: {row[1]}, 價格: {row[2]}元")
        # 4.4 更新數(shù)據(jù)
        update_sql = "UPDATE book_info SET price = ? WHERE book_name = ?"
        update_data = (79.9, "Python編程：從入門到精通")
        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ù)庫操作異常：{e}")
        if connection:
            connection.rollback()
    finally:
        # 關閉資源
        if cursor:
            cursor.close()
            print("? 游標已關閉")
        if connection:
            connection.close()
            print("? 數(shù)據(jù)庫連接已關閉")
# 執(zhí)行示例
if __name__ == "__main__":
    sqlite_demo()

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

import psycopg2
from psycopg2 import Error
def postgresql_demo():
    # 1. 數(shù)據(jù)庫連接參數(shù)
    db_config = {
        "host": "localhost",
        "port": 5432,          # PostgreSQL默認端口5432（區(qū)別于MySQL的3306）
        "user": "postgres",    # PostgreSQL默認用戶名postgres
        "password": "123456",
        "database": "test_pg"
    }
    connection = None
    cursor = None
    try:
        # 2. 建立連接
        connection = psycopg2.connect(**db_config)
        cursor = connection.cursor()
        print("? 成功連接到PostgreSQL數(shù)據(jù)庫")
        # 3. 建表（支持JSON類型，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類型（高效存儲JSON數(shù)據(jù)）
        );
        """
        cursor.execute(create_table_sql)
        print("? 表創(chuàng)建成功（若不存在）")
        # 4. 插入數(shù)據(jù)（支持JSON類型數(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. 查詢數(shù)據(jù)（支持JSON字段查詢，PostgreSQL特色）
        select_sql = "SELECT * FROM student_info WHERE info->>'class' = %s"
        cursor.execute(select_sql, ("一班",))  # 查詢class為一班的學生
        result = cursor.fetchall()
        print(f"? 一班學生（共{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ù)庫操作異常：{e}")
        if connection:
            connection.rollback()
    finally:
        # 關閉資源
        if cursor:
            cursor.close()
        if connection:
            connection.close()
            print("? 數(shù)據(jù)庫連接已關閉")
if __name__ == "__main__":
    postgresql_demo()

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

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

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

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