基于sqlalchemy对mysql实现增删改查操作的方法

这篇文章将为大家详细讲解有关基于sqlalchemy对mysql实现增删改查操作的方法，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。

需求场景：

老大让我利用爬虫爬取的数据写到或更新到mysql数据库中，百度了两种方法

1 是使用pymysql连接mysql,通过操作原生的sql语句进行增删改查数据；

2 是使用sqlalchemy连接mysql,通过ORM模型建表并操作数据库，不需要写原生的sql语句，相对简单些；

以下就是本次使用sqlalchemy的经验之谈。

实现流程：连接数据库》通过模型类创建表》建立会话》执行创建表语句》通过会话进行增删改查

from sqlalchemy import exists, Column, Integer, String, ForeignKey, existsfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker# 创建的数据库引擎engine = create_engine("mysql+pymysql://user:pwd@ip/数据库名?charset=utf8")#创建session类型DBSession = sessionmaker(bind=engine)# 实例化官宣模型 - Base 就是 ORM 模型Base = declarative_base()# 创建服务单表class ServiceOrder(Base):  __tablename__ = 'serviceOrderTable'  id = Column(Integer, primary_key=True, autoincrement=True)  serviceOrderId = Column(String(32), nullable=False, index=True, comment='服务单ID')  serviceDesc = Column(String(268), comment='服务说明')  oneLevelName = Column(String(32), comment='C类别')  twoLevelName = Column(String(32), comment='T子类')  threeLevelName = Column(String(32), comment='I项目')  fourLevelName = Column(String(32), comment='S子项')  transferTimes = Column(String(32), comment='转派次数')  overDueStatus = Column(String(32), comment='过期状态')  serviceTimeLimit = Column(String(32), comment='服务时限')  serTimeLimitTypeName = Column(String(16), comment='时限类型')    # 一对多:  # serviceWorkOrder = relationship("ServiceWorkOrder", backref="serviceorder")# 多对一:多个服务工单可以属于服务单class ServiceWorkOrder(Base):  __tablename__ = 'serviceWorkOrderTable'  id = Column(Integer, primary_key=True, autoincrement=True)  serviceWorkOrderId = Column(String(32), nullable=False, index=True, comment='服务工单ID')  workOrderName = Column(String(268), comment='工单名称')  fromId = Column(String(32), comment='服务单ID')  createUserSectionName = Column(String(32), comment='创建人室')  createUserName = Column(String(32), comment='创建人')  handlerName = Column(String(32), comment='处理人')  statusName = Column(String(32), comment='工单状态')  createTime = Column(String(32), comment='创建时间')   # "多"的一方的book表是通过外键关联到user表的:  # serviceOrder_id = Column(Integer, ForeignKey('serviceOrderTable.id'))# 创建数据库 如果数据库已存在 则不会创建 会根据库名直接连接已有的库def init_db():  Base.metadata.create_all(engine)def drop_db():  Base.metadata.drop_all(engine)def insert_update():  # all_needed_data_lists 是需要插入数据库的数据 格式[{key: value, ... }, { }, { }...]  for item in all_needed_data_lists:    ServiceOrderRow = ServiceOrder(serviceOrderId=item['serviceOrderId'],                    serviceDesc=item['serviceDesc'],                    oneLevelName=item['oneLevelName'],                    twoLevelName=item['twoLevelName'],                    threeLevelName=item['threeLevelName'],                    fourLevelName=item['fourLevelName'],                    transferTimes=item['transferTimes'],                    overDueStatus=item['overDueStatus'],                    serviceTimeLimit=item['serviceTimeLimit'],                    serTimeLimitTypeName=item['serTimeLimitTypeName'],                    )    try:      # 利用exists判断目标对象是否存在，返回True或Faults      it_exists = session.query(          exists().where(ServiceOrder.serviceOrderId == item['serviceOrderId'] )        ).scalar()    except Exception as e:      self.log.error(e)      break    try:      # 如果不存在，进行新增；存在的话就更新现存的数据      if not it_exists:        session.add(ServiceOrderRow)      else:        session.query(ServiceOrder).filter(ServiceOrder.serviceOrderId == item['serviceOrderId'])\          .update(item)    except Exception as e:      self.log.error(e)      break  try:    session.commit()    self.log.info('数据更新成功！')  except:    session.rollback()    self.log.info('数据更新失败！')if __name__ == "__main__":  # 创建数据库 如果数据库已存在 则不会创建 会根据库名直接连接已有的库  init_db()  # 创建session对象，进行增删改查:  session = DBSession()  # 利用session 增 改数据 记得提交  insert_update()  

关于基于sqlalchemy对mysql实现增删改查操作的方法就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。