2025年P(guān)ython數(shù)據(jù)庫(kù)操作專項(xiàng)訓(xùn)練試卷：MySQL與SQLAlchemy模擬實(shí)戰(zhàn)考試時(shí)間：______分鐘總分：______分姓名：______一、請(qǐng)編寫Python代碼，使用`pymysql`庫(kù)連接到名為`test_db`的MySQL數(shù)據(jù)庫(kù)，該數(shù)據(jù)庫(kù)運(yùn)行在本地服務(wù)器（默認(rèn)端口），用戶名為`user`，密碼為`password`。連接成功后，打印連接對(duì)象。二、假設(shè)存在一個(gè)名為`employees`的MySQL表，其結(jié)構(gòu)如下：|ColumnName|DataType|Constraints||:----------|:--------|:-------------||id|INT|PRIMARYKEY,AUTO_INCREMENT||first_name|VARCHAR(50)|NOTNULL||last_name|VARCHAR(50)|NOTNULL||email|VARCHAR(100)|UNIQUE||department|VARCHAR(50)||請(qǐng)使用SQLAlchemy的ORM方式，定義一個(gè)`Employee`模型類來映射該表。三、在上一題定義的`Employee`模型類基礎(chǔ)上，請(qǐng)編寫Python代碼：1.創(chuàng)建一個(gè)指向名為`test_db`的MySQL數(shù)據(jù)庫(kù)的SQLAlchemy引擎（使用`mysql+pymysql://`驅(qū)動(dòng)）。2.創(chuàng)建一個(gè)Session類，并實(shí)例化一個(gè)Session對(duì)象。3.使用Session添加一個(gè)新的`Employee`對(duì)象到數(shù)據(jù)庫(kù)中，該員工的信息為：`first_name='John'`,`last_name='Doe'`,`email='john.doe@'`,`department='HR'`。4.提交事務(wù)，并關(guān)閉Session。四、繼續(xù)使用第三題的`Employee`模型類和數(shù)據(jù)庫(kù)連接：請(qǐng)編寫Python代碼，使用SQLAlchemyORM查詢`employees`表中`department`為`'Engineering'`的所有員工記錄，并將結(jié)果打印出來。查詢應(yīng)使用`filter()`方法。五、請(qǐng)編寫Python代碼，使用SQLAlchemyORM更新`employees`表中`id`為`5`的員工的`department`字段值為`'Marketing'`。要求使用原生SQL表達(dá)式語言進(jìn)行更新操作，并提交事務(wù)。六、請(qǐng)編寫Python代碼，使用SQLAlchemyORM刪除`employees`表中`id`為`10`的員工記錄。要求使用ORM對(duì)象方式（即先獲取對(duì)象再刪除）進(jìn)行刪除操作，并提交事務(wù)。七、假設(shè)`employees`表和另一個(gè)名為`departments`的表之間存在一對(duì)多關(guān)系，`departments`表結(jié)構(gòu)如下：|ColumnName|DataType|Constraints|Relationship||:----------|:--------|:-------------|:-----------||id|INT|PRIMARYKEY,AUTO_INCREMENT||name|VARCHAR(50)|UNIQUE|||manager_id|INT|FOREIGNKEYREFERENCESemployees(id)|一對(duì)多|請(qǐng)編寫Python代碼，使用SQLAlchemyORM查詢`departments`表中所有記錄，并打印出每個(gè)部門的名稱及其管理者的`first_name`和`last_name`。要求使用`join()`方法進(jìn)行關(guān)聯(lián)查詢。試卷答案一、```pythonimportpymysqlconn=pymysql.connect(host='localhost',user='user',password='password',database='test_db')print(conn)```解析：使用`pymysql.connect()`函數(shù)建立與MySQL數(shù)據(jù)庫(kù)的連接。需要提供數(shù)據(jù)庫(kù)的主機(jī)名（host）、用戶名（user）、密碼（password）和數(shù)據(jù)庫(kù)名（database）。成功連接后，連接對(duì)象`conn`會(huì)被返回，并打印出來。二、```pythonfromsqlalchemyimportColumn,Integer,String,create_enginefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmakerBase=declarative_base()classEmployee(Base):__tablename__='employees'id=Column(Integer,primary_key=True,autoincrement=True)first_name=Column(String(50),nullable=False)last_name=Column(String(50),nullable=False)email=Column(String(100),unique=True)department=Column(String(50))```解析：首先導(dǎo)入必要的SQLAlchemy模塊。使用`declarative_base()`創(chuàng)建一個(gè)基類`Base`。然后定義`Employee`類，繼承自`Base`。在類內(nèi)部使用`__tablename__`指定表名。使用`Column`定義表的列，指定列名、數(shù)據(jù)類型（如`Integer`,`String`）和約束（如`primary_key`,`nullable`,`unique`）。三、```pythonfromsqlalchemy.ormimportsessionmaker#創(chuàng)建引擎engine=create_engine('mysql+pymysql://user:password@localhost/test_db')#創(chuàng)建Session類Session=sessionmaker(bind=engine)#實(shí)例化Session對(duì)象session=Session()#創(chuàng)建Employee對(duì)象new_employee=Employee(first_name='John',last_name='Doe',email='john.doe@',department='HR')#添加到Sessionsession.add(new_employee)#提交事務(wù)mit()#關(guān)閉Sessionsession.close()```解析：使用`create_engine()`創(chuàng)建一個(gè)數(shù)據(jù)庫(kù)引擎對(duì)象，連接到指定的數(shù)據(jù)庫(kù)。`sessionmaker()`創(chuàng)建一個(gè)Session類，通過`bind`參數(shù)綁定引擎。實(shí)例化Session類得到Session對(duì)象。創(chuàng)建一個(gè)新的`Employee`對(duì)象，設(shè)置其屬性。使用`session.add()`將對(duì)象添加到Session的臨時(shí)集合中。調(diào)用`mit()`提交事務(wù)，將更改保存到數(shù)據(jù)庫(kù)。最后調(diào)用`session.close()`關(guān)閉Session。四、```pythonfromsqlalchemy.ormimportsessionmaker#假設(shè)engine和Session已經(jīng)定義如第三題Session=sessionmaker(bind=engine)session=Session()#使用filter()方法進(jìn)行查詢query=session.query(Employee).filter(Employee.department=='Engineering')results=query.all()#打印結(jié)果forempinresults:print(emp.id,emp.first_name,emp.last_name,emp.email,emp.department)#關(guān)閉Sessionsession.close()```解析：實(shí)例化Session對(duì)象。使用`session.query(Employee)`創(chuàng)建一個(gè)查詢對(duì)象，指定要查詢的模型類。使用`filter()`方法添加過濾條件，這里是`Employee.department=='Engineering'`。調(diào)用`all()`方法執(zhí)行查詢并獲取所有結(jié)果。遍歷結(jié)果并打印每個(gè)員工的詳細(xì)信息。最后關(guān)閉Session。五、```pythonfromsqlalchemyimportselect#假設(shè)engine和Session已經(jīng)定義如第三題Session=sessionmaker(bind=engine)session=Session()#使用原生SQL表達(dá)式語言進(jìn)行更新stmt=select([Employee]).where(Employee.id==5)result=session.execute(stmt)ifresult.rowcount>0:session.execute(Employee.__table__.update().where(Employee.id==5).values(department='Marketing'))mit()#關(guān)閉Sessionsession.close()```解析：實(shí)例化Session對(duì)象。使用`select()`創(chuàng)建一個(gè)SELECT語句對(duì)象，用于檢查id為5的記錄是否存在。執(zhí)行SELECT語句。如果查詢結(jié)果行數(shù)大于0，說明記錄存在，使用`Employee.__table__.update()`創(chuàng)建一個(gè)UPDATE語句，指定where條件和新的值（department='Marketing'）。執(zhí)行UPDATE語句并提交事務(wù)。最后關(guān)閉Session。六、```python#假設(shè)engine和Session已經(jīng)定義如第三題Session=sessionmaker(bind=engine)session=Session()#獲取id為10的Employee對(duì)象employee_to_delete=session.query(Employee).filter(Employee.id==10).first()#如果對(duì)象存在，則刪除ifemployee_to_delete:session.delete(employee_to_delete)mit()#關(guān)閉Sessionsession.close()```解析：實(shí)例化Session對(duì)象。使用`session.query(Employee).filter(Employee.id==10).first()`查詢id為10的員工對(duì)象。如果查詢到的對(duì)象不為空（`first()`返回非None），則使用`session.delete()`方法刪除該對(duì)象。調(diào)用`mit()`提交刪除操作。最后關(guān)閉Session。七、```python#假設(shè)engine和Session已經(jīng)定義如第三題Session=sessionmaker(bind=engine)session=Session()#使用join()方法進(jìn)行關(guān)聯(lián)查詢query=session.query(D,Employee.first_name,Employee.last_name).join(Employee).filter(Department.manager_id==Employee.id)results=query.all()#打印結(jié)果forname,first_name,last_nameinresults:print(f"Department:{name},Manager:{first_name}{last_name}")#關(guān)閉Sessionsession.close()```解析：實(shí)例化