基于sqlalchemy對(duì)mysql實(shí)現(xiàn)增刪改查操作的方法-創(chuàng)新互聯(lián)

這篇文章將為大家詳細(xì)講解有關(guān)基于sqlalchemy對(duì)mysql實(shí)現(xiàn)增刪改查操作的方法，小編覺得挺實(shí)用的，因此分享給大家做個(gè)參考，希望大家閱讀完這篇文章后可以有所收獲。

在鳳岡等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強(qiáng)發(fā)展的系統(tǒng)性、市場(chǎng)前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供做網(wǎng)站、網(wǎng)站制作 網(wǎng)站設(shè)計(jì)制作按需定制開發(fā),公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),高端網(wǎng)站設(shè)計(jì),成都營(yíng)銷網(wǎng)站建設(shè),外貿(mào)網(wǎng)站建設(shè),鳳岡網(wǎng)站建設(shè)費(fèi)用合理。

需求場(chǎng)景：

老大讓我利用爬蟲爬取的數(shù)據(jù)寫到或更新到mysql數(shù)據(jù)庫中，百度了兩種方法

1 是使用pymysql連接mysql,通過操作原生的sql語句進(jìn)行增刪改查數(shù)據(jù)；

2 是使用sqlalchemy連接mysql,通過ORM模型建表并操作數(shù)據(jù)庫，不需要寫原生的sql語句，相對(duì)簡(jiǎn)單些；

以下就是本次使用sqlalchemy的經(jīng)驗(yàn)之談。

實(shí)現(xiàn)流程：連接數(shù)據(jù)庫》通過模型類創(chuàng)建表》建立會(huì)話》執(zhí)行創(chuàng)建表語句》通過會(huì)話進(jìn)行增刪改查

from sqlalchemy import exists, Column, Integer, String, ForeignKey, exists
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 創(chuàng)建的數(shù)據(jù)庫引擎
engine = create_engine("mysql+pymysql://user:pwd@ip/數(shù)據(jù)庫名?charset=utf8")

#創(chuàng)建session類型
DBSession = sessionmaker(bind=engine)

# 實(shí)例化官宣模型 - Base 就是 ORM 模型
Base = declarative_base()


# 創(chuàng)建服務(wù)單表
class ServiceOrder(Base):
  __tablename__ = 'serviceOrderTable'
  id = Column(Integer, primary_key=True, autoincrement=True)
  serviceOrderId = Column(String(32), nullable=False, index=True, comment='服務(wù)單ID')
  serviceDesc = Column(String(268), comment='服務(wù)說明')
  oneLevelName = Column(String(32), comment='C類別')
  twoLevelName = Column(String(32), comment='T子類')
  threeLevelName = Column(String(32), comment='I項(xiàng)目')
  fourLevelName = Column(String(32), comment='S子項(xiàng)')
  transferTimes = Column(String(32), comment='轉(zhuǎn)派次數(shù)')
  overDueStatus = Column(String(32), comment='過期狀態(tài)')
  serviceTimeLimit = Column(String(32), comment='服務(wù)時(shí)限')
  serTimeLimitTypeName = Column(String(16), comment='時(shí)限類型')  
  # 一對(duì)多:
  # serviceWorkOrder = relationship("ServiceWorkOrder", backref="serviceorder")


# 多對(duì)一:多個(gè)服務(wù)工單可以屬于服務(wù)單
class ServiceWorkOrder(Base):
  __tablename__ = 'serviceWorkOrderTable'
  id = Column(Integer, primary_key=True, autoincrement=True)
  serviceWorkOrderId = Column(String(32), nullable=False, index=True, comment='服務(wù)工單ID')
  workOrderName = Column(String(268), comment='工單名稱')
  fromId = Column(String(32), comment='服務(wù)單ID')
  createUserSectionName = Column(String(32), comment='創(chuàng)建人室')
  createUserName = Column(String(32), comment='創(chuàng)建人')
  handlerName = Column(String(32), comment='處理人')
  statusName = Column(String(32), comment='工單狀態(tài)')
  createTime = Column(String(32), comment='創(chuàng)建時(shí)間') 
  # “多”的一方的book表是通過外鍵關(guān)聯(lián)到user表的:
  # serviceOrder_id = Column(Integer, ForeignKey('serviceOrderTable.id'))

# 創(chuàng)建數(shù)據(jù)庫 如果數(shù)據(jù)庫已存在 則不會(huì)創(chuàng)建 會(huì)根據(jù)庫名直接連接已有的庫
def init_db():
  Base.metadata.create_all(engine)

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

def insert_update():
  # all_needed_data_lists 是需要插入數(shù)據(jù)庫的數(shù)據(jù) 格式[{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判斷目標(biāo)對(duì)象是否存在，返回True或Faults
      it_exists = session.query(
          exists().where(ServiceOrder.serviceOrderId == item['serviceOrderId'] )
        ).scalar()
    except Exception as e:
      self.log.error(e)
      break
    try:
      # 如果不存在，進(jìn)行新增；存在的話就更新現(xiàn)存的數(shù)據(jù)
      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('數(shù)據(jù)更新成功！')
  except:
    session.rollback()
    self.log.info('數(shù)據(jù)更新失?。?#39;)

if __name__ == "__main__":
  # 創(chuàng)建數(shù)據(jù)庫 如果數(shù)據(jù)庫已存在 則不會(huì)創(chuàng)建 會(huì)根據(jù)庫名直接連接已有的庫
  init_db()
  # 創(chuàng)建session對(duì)象，進(jìn)行增刪改查:
  session = DBSession()
  # 利用session 增 改數(shù)據(jù) 記得提交
  insert_update()

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國(guó)服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡(jiǎn)單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢(shì)，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場(chǎng)景需求。

當(dāng)前標(biāo)題：基于sqlalchemy對(duì)mysql實(shí)現(xiàn)增刪改查操作的方法-創(chuàng)新互聯(lián)
URL鏈接：http://muchs.cn/article30/dhocso.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)建站、ChatGPT、云服務(wù)器、網(wǎng)站導(dǎo)航、動(dòng)態(tài)網(wǎng)站、建站公司

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)

網(wǎng)站托管運(yùn)營(yíng)