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

场景应用

老大我让爬取内部网站获取数据，插入到新建的表中，并每天进行爬取更新数据（后面做了定时任务）。然后根据该表统计每日的新增数量/更新数量进行制图制表，向上级汇报。

思路构建

选用sqlalchemy+mysqlconnector，连接数据库，创建表，对指定表进行crud

from sqlalchemy import exists, column, integer, string, foreignkey, datetime, text, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from conf.parseconfig import parseconf

# 从配置文件中获取数据库信息
host = parseconf.get_conf('mysqlinfo', 'host')
port = parseconf.get_conf('mysqlinfo', 'port')
dbname = parseconf.get_conf('mysqlinfo', 'dbname')
usernm = parseconf.get_conf('mysqlinfo', 'usernm')
passwd = parseconf.get_conf('mysqlinfo', 'passwd')
# 连接数据库
engine_str = "mysql+mysqlconnector://{0}:{1}@{2}:{3}/{4}".format(usernm, passwd, host, port, dbname)
# 创建的数据库引擎
engine = create_engine(engine_str, encoding='utf-8')

#创建session类型
dbsession = sessionmaker(bind=engine)
# 创建session对象，进行增删改查:
session = dbsession()

# 实例化官宣模型 - base 就是 orm 模型
base = declarative_base()


# 创建服务单表 继承base基类
class serviceorder(base):
  __tablename__ = 'serviceordertable'
  serviceorderid = column(string(32), primary_key=true, comment='服务单id')
  servicedesc = column(string(512), comment='服务说明')
  transfertimes = column(string(32), comment='转派次数')
  # 创建更新时间，对数据的更新进行记录
  updatetime = column(datetime, server_default=func.now(), onupdate=func.now())


def init_db():
  base.metadata.create_all(engine)


def drop_db():
  base.metadata.drop_all(engine)


if __name__ == "__main__":
  # 每次执行时 会判断表的存在性 对于数据库中不存在的表进行创建 已存在的表则可以直接进行增删改查
  init_db()

  ### 首先讲一下使用sqlalchemy执行原生的sql语句###
  # 方式一：
  res = session.execute('select * from serviceorder') # res是获取的对象
  all_res_list = res.fetchall() # all_res_list具体的结果 是列表
  print(all_res_list ) # 结果： [('数据提取',), ('非数据提取',)]
  # 方式二：
  conn = engine.connect()
  res = conn.execute('select * from serviceorder') 
  all_res_list = res.fetchall()

  ### 使用创建好的session对象进行增删改查 ###
  # 插入单条数据
  # 创建新service0rder对象
  new_serviceorder = serviceorder(serviceorderid='001', servicedesc='ack', transfertimes='9')
  # 添加到session
  session.add(new_serviceorder)
  # 提交即保存到数据库
  session.commit()

  # 插入多条数据
  serviceorder_list = [serviceorder(serviceorderid='002', servicedesc='好的', transfertimes='9'),serviceorder(serviceorderid='003', servicedesc='起床', transfertimes='9')]
  session.add_all(serviceorder_list)
  session.commit()
  # session.close()

  # 查询
  # 查询是否存在 结果是布尔值
  it_exists = session.query(
    exists().where(serviceorder.serviceorderid == '002')
  ).scalar()
  # 创建query查询，filter是where条件
  # 调用one() first()返回唯一行，如果调用all()则已列表的形式返回所有行:
  server_order = session.query(serviceorder).filter(serviceorder.serviceorderid == '003').first()
  print(server_order.servicedesc)
  serciceorders = session.query(serviceorder).filter(serviceorder.servicedesc == '好的').all()

  # 改 更新数据
  # 数据更新，将值为mack的servicedesc修改为danny
  update_obj = session.query(serviceorder).filter(serviceorder.servicedesc == 'mack').update({"servicedesc": "danny"})
  # 或则
  update_objp = session.query(serviceorder).filter(serviceorder.servicedesc == 'mack').first()
  update_objp.servicedesc = 'danny'
  session.commit()

  # 删除
  update_objk = session.query(serviceorder).filter(serviceorder.servicedesc == 'mack').delete()
  # 或则
  update_objkp = session.query(serviceorder).filter(serviceorder.servicedesc == 'mack').one()
  update_objkp.delete()
  session.commit()
  session.close()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。