2025年Python數據庫操作專項訓練試卷：MySQL與SQLAlchemy模擬實戰

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