2025年P(guān)ython數(shù)據(jù)庫(kù)操作專項(xiàng)訓(xùn)練試卷：高效數(shù)據(jù)管理技巧考試時(shí)間：______分鐘總分：______分姓名：______一、簡(jiǎn)述在Python中使用數(shù)據(jù)庫(kù)連接池的主要優(yōu)勢(shì)。請(qǐng)從資源管理、性能提升和開發(fā)體驗(yàn)等角度進(jìn)行說明。二、假設(shè)你正在開發(fā)一個(gè)內(nèi)容管理系統(tǒng)，需要設(shè)計(jì)一個(gè)`Articles`表來存儲(chǔ)文章信息。表結(jié)構(gòu)應(yīng)包含以下字段：*`id`：文章唯一標(biāo)識(shí)符，整數(shù)類型，主鍵，自動(dòng)增長(zhǎng)。*`title`：文章標(biāo)題，字符串類型，非空。*`content`：文章正文，文本類型，非空。*`author_id`：作者ID，整數(shù)類型，外鍵，關(guān)聯(lián)到`Authors`表的`id`字段。*`created_at`：創(chuàng)建時(shí)間，日期時(shí)間類型，非空，默認(rèn)為當(dāng)前時(shí)間。*`updated_at`：更新時(shí)間，日期時(shí)間類型，非空，默認(rèn)為當(dāng)前時(shí)間，每次更新記錄時(shí)自動(dòng)刷新。請(qǐng)編寫創(chuàng)建該`Articles`表的SQL語(yǔ)句。為了提高查詢效率，你需要為某些字段添加索引，請(qǐng)說明你會(huì)為哪些字段添加索引，并給出相應(yīng)的索引創(chuàng)建SQL語(yǔ)句。三、請(qǐng)編寫一段Python代碼，使用`sqlite3`模塊連接到一個(gè)名為`test.db`的SQLite數(shù)據(jù)庫(kù)（如果數(shù)據(jù)庫(kù)不存在則自動(dòng)創(chuàng)建）。然后，執(zhí)行以下操作：1.使用`Articles`表結(jié)構(gòu)（來自第二題）創(chuàng)建`articles`表。2.插入三篇文章的數(shù)據(jù)到`articles`表中。假設(shè)`Authors`表已經(jīng)存在，并且包含`id`和`name`字段。你需要為每篇文章指定一個(gè)作者ID。3.查詢所有標(biāo)題中包含"Python"的文章的標(biāo)題和內(nèi)容。4.將上述查詢結(jié)果中第一篇文章的`updated_at`字段更新為當(dāng)前時(shí)間。5.查詢`articles`表的總文章數(shù)量。6.關(guān)閉數(shù)據(jù)庫(kù)連接。注意：請(qǐng)確保代碼能夠處理可能發(fā)生的數(shù)據(jù)庫(kù)異常，并在操作完成后正確關(guān)閉連接。你可以使用`try...except...finally`結(jié)構(gòu)或`with`語(yǔ)句。四、考慮以下兩個(gè)SQL查詢語(yǔ)句，它們旨在從`Employees`表和`Departments`表中獲取員工姓名及其所在部門名稱。查詢A:```sqlSELECTE.name,D.nameASdepartment_nameFROMEmployeesEJOINDepartmentsDONE.department_id=D.idWHEREE.age>30;```查詢B:```sqlSELECTE.name,D.nameASdepartment_nameFROMEmployeesELEFTJOINDepartmentsDONE.department_id=D.idWHEREE.age>30;```請(qǐng)解釋這兩個(gè)查詢語(yǔ)句在執(zhí)行結(jié)果和返回記錄上的主要區(qū)別。在什么場(chǎng)景下你會(huì)選擇使用查詢A，而在什么場(chǎng)景下會(huì)選擇使用查詢B？五、當(dāng)你需要向一個(gè)包含數(shù)萬條記錄的表中批量插入數(shù)據(jù)時(shí)，直接使用多次單條插入語(yǔ)句（`cursor.execute()`多次調(diào)用）通常效率很低。請(qǐng)描述至少兩種提高批量插入效率的方法，并簡(jiǎn)述其原理。六、請(qǐng)編寫Python代碼，使用`pymysql`或`mysql-connector-python`模塊連接到一臺(tái)運(yùn)行MySQL的數(shù)據(jù)庫(kù)服務(wù)器（提供服務(wù)器地址、數(shù)據(jù)庫(kù)名、用戶名和密碼）。假設(shè)該數(shù)據(jù)庫(kù)中已經(jīng)存在一個(gè)名為`ProductInventory`的表，該表包含`product_id`（產(chǎn)品ID）和`quantity`（庫(kù)存數(shù)量）字段。你的任務(wù)是編寫一個(gè)函數(shù)`update_inventory(product_id,quantity_change)`，該函數(shù)接收產(chǎn)品ID和庫(kù)存變動(dòng)量（正數(shù)表示入庫(kù)，負(fù)數(shù)表示出庫(kù)），更新對(duì)應(yīng)產(chǎn)品的庫(kù)存數(shù)量。如果產(chǎn)品ID不存在，則應(yīng)插入一條新記錄。請(qǐng)確保函數(shù)能夠處理數(shù)據(jù)庫(kù)連接異常，并在更新后正確提交事務(wù)。試卷答案一、數(shù)據(jù)庫(kù)連接池通過預(yù)先創(chuàng)建并維護(hù)一組數(shù)據(jù)庫(kù)連接，可供應(yīng)用程序按需復(fù)用，避免了頻繁建立和銷毀連接的開銷。這顯著減少了連接建立的時(shí)間消耗和系統(tǒng)資源占用（如內(nèi)存、文件描述符）。連接池還能實(shí)現(xiàn)連接的負(fù)載均衡和高效管理，提升系統(tǒng)響應(yīng)速度和吞吐量。對(duì)于應(yīng)用程序而言，使用連接池簡(jiǎn)化了數(shù)據(jù)庫(kù)連接的管理代碼，提高了開發(fā)效率和代碼的可維護(hù)性。二、```sqlCREATETABLEArticles(idINTEGERPRIMARYKEYAUTOINCREMENT,titleVARCHAR(255)NOTNULL,contentTEXTNOTNULL,author_idINTEGER,created_atDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMP,updated_atDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,FOREIGNKEY(author_id)REFERENCESAuthors(id));CREATEINDEXidx_titleONArticles(title);CREATEINDEXidx_author_idONArticles(author_id);--可選：如果經(jīng)常按創(chuàng)建時(shí)間或更新時(shí)間查詢，也可創(chuàng)建索引--CREATEINDEXidx_created_atONArticles(created_at);--CREATEINDEXidx_updated_atONArticles(updated_at);```解析思路：1.根據(jù)字段要求，定義各字段的類型、是否允許為空（`NOTNULL`）、主鍵（`PRIMARYKEY`）和默認(rèn)值（`DEFAULT`）。`id`設(shè)為主鍵并自動(dòng)增長(zhǎng)（SQLite用`AUTOINCREMENT`，其他數(shù)據(jù)庫(kù)可能用`AUTO_INCREMENT`或`IDENTITY`）。`created_at`和`updated_at`使用`DEFAULTCURRENT_TIMESTAMP`實(shí)現(xiàn)自動(dòng)賦值，`updated_at`還使用`ONUPDATECURRENT_TIMESTAMP`實(shí)現(xiàn)自動(dòng)更新。2.定義外鍵約束`FOREIGNKEY(author_id)REFERENCESAuthors(id)`，建立與`Authors`表的關(guān)聯(lián)。3.索引選擇：為提高查詢效率，通常需要為經(jīng)常用作搜索條件或排序依據(jù)的字段創(chuàng)建索引。`title`字段用于包含內(nèi)容的模糊查詢，`author_id`字段可能用于根據(jù)作者篩選文章，因此為這兩個(gè)字段創(chuàng)建索引。對(duì)`created_at`或`updated_at`建立索引可能根據(jù)具體的查詢需求決定。三、```pythonimportsqlite3fromdatetimeimportdatetimedefmain():try:#連接數(shù)據(jù)庫(kù)（或創(chuàng)建）conn=sqlite3.connect('test.db')cursor=conn.cursor()#創(chuàng)建表create_table_sql="""CREATETABLEIFNOTEXISTSArticles(idINTEGERPRIMARYKEYAUTOINCREMENT,titleVARCHAR(255)NOTNULL,contentTEXTNOTNULL,author_idINTEGER,created_atDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMP,updated_atDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,FOREIGNKEY(author_id)REFERENCESAuthors(id));"""cursor.execute(create_table_sql)#插入數(shù)據(jù)insert_sql="""INSERTINTOArticles(title,content,author_id)VALUES(?,?,?);"""articles_data=[("PythonBasics","ContentaboutPythonbasics...",1),("DatabaseTuning","Contentaboutdatabasetuning...",2),("WebDevelopment","ContentaboutwebdevelopmentwithPython...",3)]cursor.executemany(insert_sql,articles_data)#查詢標(biāo)題包含"Python"的文章query_sql="SELECTtitle,contentFROMArticlesWHEREtitleLIKE?"cursor.execute(query_sql,('%Python%',))python_articles=cursor.fetchall()forartinpython_articles:print(f"Title:{art[0]},Content:{art[1][:30]}...")#打印標(biāo)題和內(nèi)容前30個(gè)字符#更新第一篇文章的updated_at（假設(shè)查詢結(jié)果非空）ifpython_articles:update_sql="UPDATEArticlesSETupdated_at=?WHEREid=(SELECTidFROMArticlesWHEREtitleLIKE?ORDERBYidLIMIT1)"cursor.execute(update_sql,(datetime.now(),'%Python%'))#查詢文章總數(shù)count_sql="SELECTCOUNT(*)FROMArticles"cursor.execute(count_sql)count_result=cursor.fetchone()print(f"TotalArticles:{count_result[0]}")#提交事務(wù)mit()exceptsqlite3.Errorase:print(f"Databaseerror:{e}")#發(fā)生錯(cuò)誤時(shí)回滾事務(wù)conn.rollback()finally:#關(guān)閉游標(biāo)和連接ifcursor:cursor.close()ifconn:conn.close()if__name__=="__main__":main()```解析思路：1.導(dǎo)入模塊與連接：導(dǎo)入`sqlite3`模塊，使用`connect`連接到`test.db`，失敗則創(chuàng)建。2.創(chuàng)建表：使用`CREATETABLEIFNOTEXISTS`語(yǔ)句創(chuàng)建`Articles`表，確保表已存在時(shí)不重復(fù)創(chuàng)建。結(jié)構(gòu)參照第二題。3.插入數(shù)據(jù)：使用`INSERTINTO...VALUES(?,?)`語(yǔ)句配合`executemany`進(jìn)行批量插入，提高效率。`author_id`假設(shè)已存在于`Authors`表中。4.查詢：使用`SELECT`語(yǔ)句結(jié)合`LIKE`進(jìn)行模糊查詢，查找標(biāo)題包含"Python"的文章。使用`fetchall()`獲取結(jié)果。5.更新：基于查詢結(jié)果，使用`UPDATE`語(yǔ)句更新第一篇文章的`updated_at`字段為當(dāng)前時(shí)間。這里使用子查詢找到第一篇文章的`id`。6.計(jì)數(shù)：使用`SELECTCOUNT(*)`查詢表的總記錄數(shù)。7.異常處理：使用`try...except...finally`結(jié)構(gòu)。`except`捕獲`sqlite3.Error`，打印錯(cuò)誤并回滾事務(wù)。`finally`確保無論如何都關(guān)閉游標(biāo)和連接。8.事務(wù)管理：使用`mit()`提交所有更改。四、查詢A使用的是`INNERJOIN`。它只返回那些同時(shí)存在于`Employees`和`Departments`表中，并且滿足`age>30`條件的記錄。如果`Employees`表中的某員工對(duì)應(yīng)的`department_id`在`Departments`表中不存在，則該員工記錄不會(huì)被包含在結(jié)果中。查詢B使用的是`LEFTJOIN`。它返回所有滿足`age>30`條件的`Employees`表中的記錄。對(duì)于`Employees`表中的每條記錄，數(shù)據(jù)庫(kù)會(huì)嘗試在`Departments`表中查找匹配的`department_id`。如果找到匹配，則返回部門名稱；如果沒有找到匹配（即該員工的部門不存在于`Departments`表中），則部門名稱列會(huì)顯示為`NULL`。選擇查詢A的場(chǎng)景：當(dāng)你只關(guān)心那些確實(shí)是某個(gè)部門且年齡大于30歲的員工時(shí)。例如，你想生成一份年齡大于30歲在職員工及其部門的名單，且假設(shè)所有在職員工都必須有部門記錄。選擇查詢B的場(chǎng)景：當(dāng)你想獲取所有年齡大于30歲的員工信息，并希望知道他們隸屬于哪個(gè)部門（如果存在的話）時(shí)。例如，你想分析所有超過30歲的員工的部門分布情況，即使某些員工沒有部門（可能是待分配、離職但數(shù)據(jù)未清理等），你也希望記錄下來（部門名稱為NULL）。五、方法一：使用`cursor.executemany()`批量插入。原理：`executemany()`允許一次性傳遞多個(gè)參數(shù)序列（或一個(gè)包含序列的列表）給`execute()`。數(shù)據(jù)庫(kù)驅(qū)動(dòng)或數(shù)據(jù)庫(kù)本身可以優(yōu)化這個(gè)過程，將多個(gè)插入操作打包成一個(gè)或幾個(gè)更高效的批處理命令發(fā)送給數(shù)據(jù)庫(kù)，減少了網(wǎng)絡(luò)往返次數(shù)和數(shù)據(jù)庫(kù)解析命令的開銷，從而顯著提升插入效率。方法二：使用數(shù)據(jù)庫(kù)的批量插入特定語(yǔ)句或工具。原理：例如，MySQL的`INSERT...SELECT`語(yǔ)句，可以從另一個(gè)表或臨時(shí)表批量插入數(shù)據(jù)；或者使用特定工具（如`mysql`命令行工具的`loaddatainfile`）導(dǎo)入數(shù)據(jù)。這些方法通常利用了數(shù)據(jù)庫(kù)底層更高效的I/O和解析機(jī)制，繞過了應(yīng)用層的逐條插入邏輯，速度遠(yuǎn)超單條插入。六、```pythonimportpymysqldefupdate_inventory(host,user,password,db,product_id,quantity_change):try:#連接MySQL數(shù)據(jù)庫(kù)conn=pymysql.connect(host=host,user=user,password=password,db=db,charset='utf8mb4')cursor=conn.cursor()#開始事務(wù)conn.begin()#檢查產(chǎn)品是否存在，并更新或插入ifquantity_change>0:#入庫(kù)操作update_sql="UPDATEProductInventorySETquantity=quantity+%sWHEREproduct_id=%s"cursor.execute(update_sql,(quantity_change,product_id))ifcursor.rowcount==0:#如果沒有行被更新（即產(chǎn)品不存在）insert_sql="INSERTINTOProductInventory(product_id,quantity)VALUES(%s,%s)"cursor.execute(insert_sql,(product_id,quantity_change))else:#出庫(kù)操作（負(fù)數(shù)表示變動(dòng)量）#確保庫(kù)存不會(huì)變成負(fù)數(shù)check_sql="SELECTquantityFROMProductInventoryWHEREproduct_id=%s"cursor.execute(check_sql,(product_id,))result=cursor.fetchone()ifresultandresult[0]>=-quantity_change:#如果庫(kù)存足夠update_sql="UPDATEProductInventorySETquantity=quantity+%sWHEREproduct_id=%s"cursor.execute(update_sql,(quantity_change,product_id))else:raiseValueError(f"Insufficientinventoryforproduct_id{product_id}.")#提交事務(wù)mit()exceptpymysql.MySQLErrorase:print(f"MySQLerror:{e}")#發(fā)生錯(cuò)誤時(shí)回滾事務(wù)conn.rollback()#可以根據(jù)需要重新拋出或處理異常#raiseexceptValueErrorase:print(f"Businesslogicerro