基于SQLAlchemy实现操作MySQL并执行原生sql语句

admin

1. from sqlalchemy import exists, Column, Integer, String, ForeignKey, DateTime, Text, func
   
2. from sqlalchemy.ext.declarative import declarative_base
   
3. from sqlalchemy import create_engine
   
4. from sqlalchemy.orm import sessionmaker
   
5. from conf.parseConfig import parseConf
   
6. # 从配置文件中获取数据库信息
   
7. host = parseConf.get_conf('MySQLInfo', 'host')
   
8. port = parseConf.get_conf('MySQLInfo', 'port')
   
9. dbname = parseConf.get_conf('MySQLInfo', 'dbname')
   
10. usernm = parseConf.get_conf('MySQLInfo', 'usernm')
    
11. passwd = parseConf.get_conf('MySQLInfo', 'passwd')
    
12. # 连接数据库
    
13. engine_str = "mysql+mysqlconnector://{0}:{1}@{2}:{3}/{4}".format(usernm, passwd, host, port, dbname)
    
14. # 创建的数据库引擎
    
15. engine = create_engine(engine_str, encoding='utf-8')
    
16. #创建session类型
    
17. DBSession = sessionmaker(bind=engine)
    
18. # 创建session对象，进行增删改查:
    
19. session = DBSession()
    
20. # 实例化官宣模型 - Base 就是 ORM 模型
    
21. Base = declarative_base()
    
22. # 创建服务单表 继承Base基类
    
23. class ServiceOrder(Base):
    
24. __tablename__ = 'serviceOrderTable'
    
25. serviceOrderId = Column(String(32), primary_key=True, comment='服务单ID')
    
26. serviceDesc = Column(String(512), comment='服务说明')
    
27. transferTimes = Column(String(32), comment='转派次数')
    
28. # 创建更新时间，对数据的更新进行记录
    
29. updateTime = Column(DateTime, server_default=func.now(), onupdate=func.now())
    
30. def init_db():
    
31. Base.metadata.create_all(engine)
    
32. def drop_db():
    
33. Base.metadata.drop_all(engine)
    
34. if __name__ == "__main__":
    
35. # 每次执行时 会判断表的存在性 对于数据库中不存在的表进行创建 已存在的表则可以直接进行增删改查
    
36. init_db()
    
37. ### 首先讲一下使用sqlalchemy执行原生的sql语句###
    
38. # 方式一：
    
39. res = session.execute('select * from ServiceOrder') # res是获取的对象
    
40. all_res_list = res.fetchall() # all_res_list具体的结果 是列表
    
41. print(all_res_list ) # 结果： [('数据提取',), ('非数据提取',)]
    
42. # 方式二：
    
43. conn = engine.connect()
    
44. res = conn.execute('select * from ServiceOrder')
    
45. all_res_list = res.fetchall()
    
46. ### 使用创建好的session对象进行增删改查 ###
    
47. # 插入单条数据
    
48. # 创建新service0rder对象
    
49. new_serviceorder = ServiceOrder(serviceOrderId='001', serviceDesc='ack', transferTimes='9')
    
50. # 添加到session
    
51. session.add(new_serviceorder)
    
52. # 提交即保存到数据库
    
53. session.commit()
    
54. # 插入多条数据
    
55. serviceorder_list = [ServiceOrder(serviceOrderId='002', serviceDesc='好的', transferTimes='9'),ServiceOrder(serviceOrderId='003', serviceDesc='起床', transferTimes='9')]
    
56. session.add_all(serviceorder_list)
    
57. session.commit()
    
58. # session.close()
    
59. # 查询
    
60. # 查询是否存在 结果是布尔值
    
61. it_exists = session.query(
    
62. exists().where(ServiceOrder.serviceOrderId == '002')
    
63. ).scalar()
    
64. # 创建Query查询，filter是where条件
    
65. # 调用one() first()返回唯一行，如果调用all()则已列表的形式返回所有行:
    
66. server_order = session.query(ServiceOrder).filter(ServiceOrder.serviceOrderId == '003').first()
    
67. print(server_order.serviceDesc)
    
68. serciceorders = session.query(ServiceOrder).filter(ServiceOrder.serviceDesc == '好的').all()
    
69. # 改 更新数据
    
70. # 数据更新，将值为Mack的serviceDesc修改为Danny
    
71. update_obj = session.query(ServiceOrder).filter(ServiceOrder.serviceDesc == 'Mack').update({"serviceDesc": "Danny"})
    
72. # 或则
    
73. update_objp = session.query(ServiceOrder).filter(ServiceOrder.serviceDesc == 'Mack').first()
    
74. update_objp.serviceDesc = 'Danny'
    
75. session.commit()
    
76. # 删除
    
77. update_objk = session.query(ServiceOrder).filter(ServiceOrder.serviceDesc == 'Mack').delete()
    
78. # 或则
    
79. update_objkp = session.query(ServiceOrder).filter(ServiceOrder.serviceDesc == 'Mack').one()
    
80. update_objkp.delete()
    
81. session.commit()
    
82. session.close()

复制代码