SQLAlchemy的使用

sqlalchemy和django中的ORM框架做比較

sqlalchemy是能夠在任何使用sql語句查詢的時候使用的
而django中的orm操做只是在django中才能夠使用

SQLAlchemy的使用：

單表操做：

建立數據表
from sqlalchemy.ext.declarative import declarative_base

#基類，和django中的models.Model同樣
BaseModel = declarative_base()

from sqlalchemy import Column,Integer,String

#像django中建立表結構同樣
class User(BaseModel):
    #這裏的tablename通常都是類名的小寫
    __tablename__ = 'user'
    id = Cloumn(Integer,primary_key=True,autoincrement=True)
    username = Column(String(32),nullable=False,unique=True,index=True)
    password=Column(String(32),nullable=False)

from sqlalchemy.engine import create_engine
#建立引擎 
　　　　 數據庫名+引擎名://數據庫用戶名:數據庫密碼@ip地址：端口/數據庫名?charset=utf8
engine=create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s18?charset=utf8")
#執行建立的指令
BaseModel.metadata.create_all(engine)

單表的增刪改查(C增R查U改D刪)

from sqlalchemy.orm import sessionmaker
from sqlalchemy.engine import create_engine

engine=create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s18?charset=utf8")

#選擇數據庫
db = sessionmaker(engine)
#新建查詢窗口
db_session = db()
#建立一個對象
u=User(name='alex')

#新增數據
db_session.add(u)
#db_session.add_all(u_list)
#只要對數據庫進行了修改，就須要commit一下
db_session.commit()
#關閉窗口
#db_session.close()

#查詢數據
db_session.query(User).all() #查詢全部數據，獲得的是一個列表中裝着查詢到的數據，若是須要取出他們裏面的屬性，直接打點調用就好了

#帶條件的查詢,查詢第一條數據
db_session.query(User).filter(id==1).first()

#並列條件,就是或者的意思
db_session.query(User).filter(id==1,name=='alex').first()

#修改數據
db_session.query(User).filter(id==1).update({name:'wusir'})
db_session.commit()

#刪除數據
db_session.query(User).filter(id==2).delete()
db_session.commit()

from sqlalchemy.sql import and_,or_

#就是而且的意思，兩個條件都要知足
db_session.query(User).filter(and_(User.id >= 3,User.name == 'alex')).all()
#就是或者的意思，知足一個條件就行
db_session.query(User).filter(or_(User.id >3 ,User.name == 'wusir')).all()
#結合着使用
db_session.query(User).filter(or_(User.id==1,User.name=='佩琪'，and_(User.id==2,User.name=='alex'))).all()

#使用原生的sql篩選條件
db_session.query(User).filter(id=1).first()

 # 字符串匹配方式篩選條件 並使用 order_by進行排序
 r6 = db_session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='DragonFire').order_by(User.id).all()

 #原生SQL查詢
 r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='DragonFire').all()

 # 篩選查詢列
 # query的時候咱們不在使用User ORM對象,而是使用User.name來對內容進行選取
 user_list = db_session.query(User.name).all()
 print(user_list)
 for row in user_list:
 print(row.name)

 # 別名映射 name as nick
 user_list = db_session.query(User.name.label("nick")).all()
 print(user_list)
 for row in user_list:
 print(row.nick) # 這裏要寫別名了

 # 篩選條件格式
 user_list = db_session.query(User).filter(User.name == "DragonFire").all()
 user_list = db_session.query(User).filter(User.name == "DragonFire").first()
 user_list = db_session.query(User).filter_by(name="DragonFire").first()
 for row in user_list:
 print(row.nick)

 # 複雜查詢
 from sqlalchemy.sql import text
 user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="DragonFire")

 # 查詢語句
 from sqlalchemy.sql import text
 user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="DragonFire")

 # 排序 :
 user_list = db_session.query(User).order_by(User.id).all()
 user_list = db_session.query(User).order_by(User.id.desc()).all()
 for row in user_list:
 print(row.name,row.id)

 #其餘查詢條件
 """
 ret = session.query(User).filter_by(name='DragonFire').all()
 ret = session.query(User).filter(User.id > 1, User.name == 'DragonFire').all()
 ret = session.query(User).filter(User.id.between(1, 3), User.name == 'DragonFire').all() # between 大於1小於3的
 ret = session.query(User).filter(User.id.in_([1,3,4])).all() # in_([1,3,4]) 只查詢id等於1,3,4的
 ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查詢不等於1,3,4的
 ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='DragonFire'))).all() 子查詢
 from sqlalchemy import and_, or_
 ret = session.query(User).filter(and_(User.id > 3, User.name == 'DragonFire')).all()
 ret = session.query(User).filter(or_(User.id < 2, User.name == 'DragonFire')).all()
 ret = session.query(User).filter(
 or_(
 User.id < 2,
 and_(User.name == 'eric', User.id > 3),
 User.extra != ""
 )).all()
 # select * from User where id<2 or (name="eric" and id>3) or extra != ""

 # 通配符
 ret = db_session.query(User).filter(User.name.like('e%')).all()
 ret = db_session.query(User).filter(~User.name.like('e%')).all() 

 # 限制
 ret = db_session.query(User)[1:2]

 # 排序
 ret = db_session.query(User).order_by(User.name.desc()).all()
 ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all()

 # 分組
 from sqlalchemy.sql import func

 ret = db_session.query(User).group_by(User.extra).all()
 ret = db_session.query(
 func.max(User.id),
 func.sum(User.id),
 func.min(User.id)).group_by(User.name).all()

 ret = db_session.query(
 func.max(User.id),
 func.sum(User.id),
 func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()
 """

 # 關閉鏈接
 db_session.close()

一對多查詢

from sqlalchemy.ext.declarative import declarative_base

basemodel = declarative_base()

from sqlalchemy import Column,Integer,String
from sqlalchemy.orm import relationship

class School(basemodel):
    __tablename__ ='school'
    id = Cloumn(Integer,primary_key=True)
    name = Cloumn(String(10),nummable=False)

class Student(basemodel):
    __tablename__ = 'student'
    id = Cloumn(Integer,primary_key=True)
    name = Cloumn(String,nullable=False)
    sch_id=Cloumn(Integer,ForeignKey('school.id'))

    #重點
    stu2sch = relationship('School',backref='sch2stu')

from sqlalchemy.engine import create_engine
engine = create_engine("mysql_pymysql://root:123@127.0.0.1:3306/s18?charset=utf8")

basemodel.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker

db = sessionmaker(engine)
db_session = db()

#正向查詢，添加數據
s=student(name='alex',stu2sch=School(name='oldboy'))
db_session.add(s)
db_session.commit()

db_session.close()
#反向查詢，添加數據
sch=school(name='oldboy')
sch.sch2stu = [
　　Student(name='li'),
　　Student（name='hua'）
　　　　　　　]
db_session.add(sch)
db_session.commit()

#正向查詢數據
ret = db_session.query(Student).first()
ret = db_session.query(Student).all()
#反向查詢數據
ret = db_session.query(School).all()
for sch in ret:
　　for stu in sch.sch2stu:
　　　　print(sch.name,stu.name)

#修改和刪除都是和上面的相同，在這裏咱們就不寫了

多對多查詢 

from sqlalchemy.ext.declarative import declarative_base

BaseModel = declarative_base()

from sqlalchemy import Column, Integer, String,ForeignKey
from sqlalchemy.orm import relationship

class Boy(BaseModel):
    __tablename__ = "boy"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)

class Girl(BaseModel):
    __tablename__ = "girl"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)
    #精髓
    gyb = relationship('Boy',backref='byg',secondary='hotel')

class Hotel(BaseModel):
    __tablename__ = "hotel"
    id = Column(Integer,primary_key=True)
    b_id = Column(Integer,ForeignKey("boy.id"))
    g_id = Column(Integer,ForeignKey("girl.id"))


from sqlalchemy.engine import create_engine

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s18sqlalchemy?charset=utf8")

BaseModel.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker

db = sessionmaker(engine)
db_session = db()

 1.增長數據 relationship 正向添加
 g = Girl(name="趙麗穎",gyb=[Boy(name="馮紹峯"),Boy(name="DragonFire")])
 db_session.add(g)
 db_session.commit()

 2.增長數據 relationship 反向添加
 b = Boy(name="Alexander.DSB.Li")
 b.byg = [Girl(name="娟兒"),Girl(name="羅玉鳳")]
 db_session.add(b)
 db_session.commit()

 3.查詢數據 relationship 正向
 res = db_session.query(Girl).first()
 for b in res.gyb:
     print(res.name,b.name)

 4.查詢數據 relationship 反向
 res = db_session.query(Boy).all()
 for b in res:
    for g in b.byg:
        print(b.name,g.name)

 Flask中使用sqlalchemy

在新建的flask項目中

views.py

from flask import Blueprint,render_template
from app01.models import User,db

user = Blueprint("user",__name__)


@user.route("/add_user",methods=["POST","GET"])
def add_user():
    u = User(name="123456")
    db.session.add(u)
    db.session.commit()
    return render_template("add_user.html")


@user.route("/login_user",methods=["POST","GET"])
def login_user():
    user_name = "123456"
    user_info = User.query.filter(User.name==user_name).first()
    print(user_info,user_info.name)
    return "200 OK"


@user.route("/del_user",methods=["POST","GET"])
def del_user():
    user_name = "123456"
    user_info = User.query.filter(User.name==user_name).delete()
    db.session.commit()
    print(user_info)
    return "200 OK"

__init__.py中

from flask import Flask
from flask_session import Session
from flask_cors import CORS


from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

from app01.views import users



def create_app():
    app = Flask(__name__)
    app.config["DEBUG"] = True
    app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:123@127.0.0.1:3306/s18sqlalchemy?charset=utf8"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

    Session(app)
    CORS(app)
    db.init_app(app)

    app.register_blueprint(users.user)

    return app

models.py中

# from sqlalchemy.ext.declarative import declarative_base
# BaseModel=declarative_base()
from app01 import db

class User(db.Model):
    __tablename__="user"
    __table_args__={}
    id = db.Column(db.Integer,primary_key=True)
    name = db.Column(db.String(32),nullable=False)


if __name__ == '__main__':
    from app01 import create_app
    app = create_app()
    db.init_app(app)
    db.drop_all(app=app)
    db.create_all(app=app)