摘要：相關(guān)用法最大的好處在于結(jié)構(gòu)清晰化，以及遷移數(shù)據(jù)庫(kù)不會(huì)造成過(guò)多的冗余。但是缺點(diǎn)是沒(méi)有純粹的的功能多，也沒(méi)有純粹的來(lái)的方便。但對(duì)于開(kāi)發(fā)人員來(lái)講，最大的好處在于閱讀代碼更加直觀。

sqlalchemy 最大的好處在于結(jié)構(gòu)清晰化，以及遷移數(shù)據(jù)庫(kù)不會(huì)造成過(guò)多的冗余。但是缺點(diǎn)是沒(méi)有純粹的的sql功能多，也沒(méi)有純粹的sql來(lái)的方便。但對(duì)于開(kāi)發(fā)人員來(lái)講，sqlalchmey最大的好處在于閱讀代碼更加直觀。

本文主要我多年來(lái)使用sqlalchemy遇到的一些坑，以及應(yīng)該如何去做，以及剛開(kāi)始萌新容易遇到的錯(cuò)誤，希望我這篇文章能夠給予我個(gè)人的一些積累。

如何構(gòu)建一個(gè)鏈接

以MYSQL為例,注意password中不要包含@

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
db_url = "mysql+pymysql://username:password@localhost:port/datebase"
app.config["SQLALCHEMY_DATABASE_URI"] = db_url
app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True # 自動(dòng)提交
app.config["SQLALCHEMY_ECHO"] = False  # 如果True在打印出sql日志
db = SQLAlchemy(app)

如何創(chuàng)建一個(gè)數(shù)據(jù)表并可以支持繼承重構(gòu)

class BaseRoom(db.Model):
    __abstract__ = True  # 生命跳過(guò)該class表的生成
    id = db.Column(db.String(50), primary_key=True, comment="用戶(hù)的id")
    third_id = db.Column(db.String(50), nullable=False, comment="網(wǎng)站的id")
    title = db.Column(db.String(500), nullable=False, comment="題目")
    price = db.Column(db.Float(asdecimal=True), nullable=False, comment="價(jià)格")
    
   
   
class DaxiangRoom(BaseRoom):
    __tablename__ = "daxiang_daxiangroom" # 表名字
    __table_args__ = {"useexisting": True}

    landlord_id = db.Column(db.ForeignKey("daxiang_daxianglandlord.id"), index=True)
    landlord = db.relationship("DaxiangLandlord",
                               primaryjoin="DaxiangRoom.landlord_id == DaxiangLandlord.id",
                               backref="daxiang_daxiangrooms")

進(jìn)行一個(gè)session的提交，查找，更新

class CtripRoomSqlAlchemyPipeline(object):

    # 將每行更新或?qū)懭霐?shù)據(jù)庫(kù)中
    def process_item(self, item, spider):
    

        model = CtripRoom(hotel_id=item["hotel_id"],
                          hotel_name=item["hotel_name"],
                          city=item["city"],
                          location=item["location"],
                          type=item["type"],
                          score=item["score"]
                          )
        try:
            db.session.add(model)
            db.session.commit()
        except Exception as e:
            # print(e)
            db.session.rollback()
            pass
        return item
        
        
class lukeRoomUpdateSqlAlchemyPipeline(object):

    # 更新sql
    def process_item(self, item, spider):
        landlord_id = item["landlord_id"]
        model = LukeRoom.query.filter_by(id=item["house_id"]).first()

        model.landlord_id = landlord_id
        model.is_finished = 1

        db.session.add(model)
        db.session.commit()
        
   

使用sqlalchemy 實(shí)現(xiàn)ON DUPLICATE KEY UPDATE

from dbs.database import db
from sqlalchemy.dialects.mysql import insert
token = content["token"]
city = content["city"]
account = content["account"]
platform = "airbnb"
user_id = content["user_id"]
_id = platform + "_" + content["user_id"]
_time = datetime.now(timezone("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")

# 插入的數(shù)據(jù)
comment_model = insert(User).values(
    id=_id, platform=platform, city=city, token=token, user_id=user_id, create_time=_time, account=account
)

# 出現(xiàn)沖突后需要更新的數(shù)據(jù)
do_update_comment_model = comment_model.on_duplicate_key_update(
    id=_id, platform=platform, city=city, token=token, user_id=user_id, create_time=_time, account=account,
    user_status=""
)
db.session.execute(do_update_comment_model)
db.session.commit()