sqlalchemy和flask-sqlalchemy

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了sqlalchemy和flask-sqlalchemy，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含15596字，纯文字阅读大概需要23分钟。

内容图文

from sqlalchemy.ext.declarative import declarative_base  # 导入基类
from sqlalchemy import Column, Integer, String  # 数据类型

# Base = ORM基类 - 要按照ORM的规则定义你的类
Base = declarative_base()

class Users(Base):
    __tablename__ = "user"
    # 创建ID数据字段 , 那么ID是不是一个数据列呢? 也就是说创建ID字段 == 创建ID数据列
    # id = Column(数据类型,索引,主键,外键,等等)
    id = Column(Integer, primary_key=True, autoincrement=True,index=True)
    name = Column(String(32), nullable=False)  # nullable=False 不能为空


# 打开数据库的连接 -- 创建数据库引擎
from sqlalchemy import create_engine
# 创建数据库引擎
# engine = create_engine("mysql://scott:tiger@hostname/dbname",encoding=‘latin1‘, echo=True)
engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8")

Base.metadata.create_all(engine)  # Base自动检索所有继承Base的ORM 对象 并且创建所有的数据表

2.单表的增删改查：

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from creatTable import Users

# 创建引擎
engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8")

Session = sessionmaker(engine)
db_session = Session()

# 1. 增加数据add(创建表结构的类名(字段名=添加的数据))
db_session.add(Users(name="ZWQ"))  # 相当于建立一条添加数据的sql语句
db_session.commit()  # 执行
db_session.close()   # 结束关闭

# 批量添加
db_session.add_all([Users(name="清风徐来"), Users(name="水波不兴")])
db_session.commit()
db_session.close()

# 2.查询 query(表结构的类名)
sqlres = db_session.query(Users)
print(sqlres)  # 直接翻译输出对应的SQL查询语句

res = db_session.query(Users).all()  # 返回表中所有数据对象
print(res)# [<creatTable.Users object at 0x00000000038A1B00>,<creatTable.Users object at 0x00000000038A1B70>]

for u in res:
    print(u.id, u.name)

res = db_session.query(Users).first()  # 取第一个，返回是对象
print(res.id, res.name)

res = db_session.query(Users).filter(Users.id == 3).first()  # 返回符合条件查询结果
print(res.name)

res = db_session.query(Users).filter(Users.id <= 2, Users.name == "ZWQ").all() # filter中的条件可以是模糊条件,多个条件
for u in res:
    print(u.id,u.name)

# 3.更改数据 update({k:v})
res = db_session.query(Users).filter(Users.id == 1).update({"name":"DragonFire"})
print(res)
db_session.commit()

res = db_session.query(Users).update({"name":"ZWQ"})  # 全部修改，返回修改的数据个数
print(res)
db_session.commit()

# 4.删除 delete()结合查询条件删除
res = db_session.query(Users).filter(Users.id == 1).delete()  # 删除否合条件的数据，返回删除数量
print(res)
db_session.commit()

res = db_session.query(Users).delete()  # 删除表中所有数据，返回删除数量
print(res)
db_session.commit()

 

3.创建外键关联的表结构：

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

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

class Student(Base):
    __tablename__ = "student"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)
    sch_id = Column(Integer,ForeignKey("school.id"))  # 关联的表的字段,表间的关系

    stu2sch = relationship("School",backref="sch2stu")  # 写在哪边那边就是正向查询，对象间的关系，backref(反向查询)


class School(Base):
    __tablename__ = "school"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),nullable=False)


from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8")

Base.metadata.create_all(engine)

 4、外键关联的表添加与查询操作：

from sqlalchemy.orm import sessionmaker

from creatTableFk import engine   
from creatTableFk import Student,School  # 导入创建表结构的类

Session = sessionmaker(engine)

db_session = Session()

# 1.添加数据
db_session.add(School(name="NCU"))
db_session.commit()

# relationship  正向添加
stu = Student(name="清风徐来",stu2sch=School(name="NCU"))
db_session.add(stu)
db_session.commit()

# relationship  反向添加
sch = School(name="NCU")
sch.sch2stu = [Student(name="YWB"),Student(name="CT")]
db_session.add(sch)
db_session.commit()

# 2.查询
res = db_session.query(Student).all()
for stu in res:
    print(stu.name,stu.stu2sch.name)  # 正向跨表

res = db_session.query(School).all()
for sch in res:
    for stu in sch.sch2stu:
        print(sch.name,stu.name)  # 反向跨表

 

5.多对多的表的创建

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

Base = declarative_base()


# 多对多关联通过第三张表关联，sqlalchemy要自己创建第三张表
class Girl(Base):
    __tablename__ = "girls"
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

    g2b = relationship("Boy", backref="b2g", secondary="hotels")


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


class Hotel(Base):
    __tablename__ = "hotels"
    id = Column(Integer, primary_key=True)
    boy_id = Column(Integer, ForeignKey("boys.id"), nullable=False)
    girl_id = Column(Integer, ForeignKey("girls.id"), nullable=False)


engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy_Pro?charset=utf8")
Base.metadata.create_all(engine)

6.多对多表数据的添加与查询：

from sqlalchemy.orm import sessionmaker

from CreateTableM2M import engine

Session = sessionmaker(engine)
db_session = Session()

from CreateTableM2M import Girl,Boy

# 1.增加数据
# relationship 正向添加
g = Girl(name="ZLY",g2b=[Boy(name="ZWQ"),Boy(name="FSF")])
db_session.add(g)
db_session.commit()

# relationship 反向添加
b = Boy(name="ZS")
b.b2g = [Girl(name="罗玉凤"),Girl(name="娟儿"),Girl(name="芙蓉姐姐")]
db_session.add(b)
db_session.commit()

# 2.查询
# relationship 正向
res = db_session.query(Girl).all()
for g in res:
    for b in g.g2b:
        print(g.name,b.name)

# relationship 反向
res = db_session.query(Boy).all()
for b in res:
    for g in b.b2g:
        print(b.name,g.name)

7. 高级ORM操作

  1 from CreateTable import Users,engine
  2 from sqlalchemy.orm import sessionmaker
  3 from CreateTableForeignKey import Student
  4 
  5 Session = sessionmaker(engine)
  6 db_session = Session()
  7 
  8 # 查询数据表操作
  9 # and or
 10 from sqlalchemy.sql import and_ , or_,desc
 11 ret = db_session.query(Users).filter(and_(Users.id > 3, Users.name == ‘DragonFire‘)).all()
 12 ret = db_session.query(Users).filter(or_(Users.id < 2, Users.name == ‘DragonFire‘)).all()
 13 
 14 ret = db_session.query(Users).filter(
 15     or_(
 16         Users.id < 2,
 17         and_(
 18             Users.name == ‘eric‘,
 19             Users.id > 3
 20         ),
 21         Users.name != ""
 22     )
 23 )
 24 print(ret)
 25 select * from User where id<2 or (name="eric" and id>3) or extra != ""
 26 
 27 
 28 # 查询所有数据
 29 r1 = db_session.query(User).all()
 30 
 31 查询数据 指定查询数据列 加入别名
 32 r2 = db_session.query(Student.name.label(‘username‘), Student.id).first()
 33 print(r2.id,r2.username) # 15 NBDragon
 34 
 35 # 表达式筛选条件
 36 r3 = db_session.query(User).filter(User.name == "DragonFire").all()
 37 
 38 # 原生SQL筛选条件
 39 r4 = db_session.query(User).filter_by(name=‘DragonFire‘).all()
 40 r5 = db_session.query(Users).filter_by(name=‘DragonFire‘).first()
 41 
 42 字符串匹配方式筛选条件 并使用 order_by进行排序
 43 r6 = db_session.query(Student).order_by(Student.name.desc()).all()
 44 for i in r6:
 45     print(i.id,i.name)
 46 
 47 原生SQL查询
 48 from sqlalchemy.sql import text
 49 r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name=‘DragonFire‘).all()
 50 
 51 # 筛选查询列
 52 # query的时候我们不在使用User ORM对象,而是使用User.name来对内容进行选取
 53 user_list = db_session.query(User.name).all()
 54 print(user_list)
 55 for row in user_list:
 56     print(row.name)
 57 
 58 # 别名映射  name as nick
 59 user_list = db_session.query(User.name.label("nick")).all()
 60 print(user_list)
 61 for row in user_list:
 62     print(row.nick) # 这里要写别名了
 63 
 64 # 筛选条件格式
 65 user_list = db_session.query(User).filter(User.name == "DragonFire").all()
 66 user_list = db_session.query(User).filter(User.name == "DragonFire").first()
 67 user_list = db_session.query(User).filter_by(name="DragonFire").first()
 68 for row in user_list:
 69     print(row.nick)
 70 
 71 # 复杂查询
 72 from sqlalchemy.sql import text
 73 user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="DragonFire")
 74 
 75 # 查询语句
 76 from sqlalchemy.sql import text
 77 user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="DragonFire")
 78 
 79 # 排序 :
 80 user_list = db_session.query(User).order_by(User.id).all()
 81 user_list = db_session.query(User).order_by(User.id.desc()).all()
 82 for row in user_list:
 83     print(row.name,row.id)
 84 
 85 #其他查询条件
 86 """
 87 ret = session.query(User).filter_by(name=‘DragonFire‘).all()
 88 ret = session.query(User).filter(User.id > 1, User.name == ‘DragonFire‘).all()
 89 ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name=‘DragonFire‘))).all() 子查询
 90 from sqlalchemy import and_, or_
 91 ret = session.query(User).filter(and_(User.id > 3, User.name == ‘DragonFire‘)).all()
 92 ret = session.query(User).filter(or_(User.id < 2, User.name == ‘DragonFire‘)).all()
 93 
 94 
 95 
 96 
 97 # 限制
 98 ret = db_session.query(User)[1:2]
 99 
100 # 排序
101 ret = db_session.query(User).order_by(User.name.desc()).all()
102 ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all()
103 
104 # 分组
105 from sqlalchemy.sql import func
106 
107 ret = db_session.query(User).group_by(User.extra).all()
108 ret = db_session.query(
109     func.max(User.id),
110     func.sum(User.id),
111     func.min(User.id)).group_by(User.name).all()
112 
113 ret = db_session.query(
114     func.max(User.id),
115     func.sum(User.id),
116     func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()
117 """
118 
119 # 关闭连接
120 db_session.close()
121 
122 
123 ret = db_session.query(Student).filter(Student.id.between(1, 3)).all() # between 大于1小于3的
124 for i in ret:
125     print(i.id,i.name)
126 ret = db_session.query(Student).filter(~Student.id.in_([1,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
127 for i in ret:
128     print(i.id,i.name)
129 ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
130 # 通配符
131 ret = db_session.query(Student).filter(Student.name.like(‘%e%‘)).all()
132 
133 
134 ret = db_session.query(Student).filter(~Student.name.like(‘Z%‘)).all()
135 for i in ret:
136     print(i.id,i.name)
137 
138 
139 
140 高级版更新操作
141 from my_create_table import User,engine
142 from sqlalchemy.orm import sessionmaker
143 
144 Session = sessionmaker(engine)
145 db_session = Session()
146 
147 #直接修改
148 db_session.query(Student).filter(Student.id > 3).update({Student.name: Student.name + "099"}, synchronize_session=False)
149 db_session.commit()
150 
151 db_session.query(Student).filter(Student.id > 3).update({"name": Student.name + "123"}, synchronize_session=False)
152 db_session.commit()
153 
154 
155 #在原有值基础上添加 - 1
156 db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)
157 
158 #在原有值基础上添加 - 2
159 db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")
160 db_session.commit()

三、Flask_SQLAlchemy

1. 安装 Flask-SQLAlchemy模块，pip安装 

　　pip install Flask-SQLAlchemy

2.先创建一个Flask项目

3.基于这个Flask项目,加入Flask-SQLAlchemy让项目变得生动起来

 1 from flask import Flask
 2 
 3 # 导入Flask-SQLAlchemy中的SQLAlchemy
 4 from flask_sqlalchemy import SQLAlchemy
 5 
 6 # 实例化SQLAlchemy
 7 db = SQLAlchemy()
 8 # PS : 实例化SQLAlchemy的代码必须要在引入蓝图之前
 9 
10 from .views.users import user
11 
12 
13 def create_app():
14     app = Flask(__name__)
15 
16     # 初始化App配置 这个app配置就厉害了,专门针对 SQLAlchemy 进行配置
17     # SQLALCHEMY_DATABASE_URI 配置 SQLAlchemy 的链接字符串儿
18     app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8"
19     # SQLALCHEMY_POOL_SIZE 配置 SQLAlchemy 的连接池大小
20     app.config["SQLALCHEMY_POOL_SIZE"] = 5
21     # SQLALCHEMY_POOL_TIMEOUT 配置 SQLAlchemy 的连接超时时间
22     app.config["SQLALCHEMY_POOL_TIMEOUT"] = 15
23     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
24 
25     # 初始化SQLAlchemy , 本质就是将以上的配置读取出来
26     db.init_app(app)
27 
28     app.register_blueprint(user)
29 
30     return app
31 
32 MyApp/__init__.py

4.建立models.py ORM模型文件

 1 from MyApp import db
 2 
 3 Base = db.Model # 这句话你是否还记的?
 4 # from sqlalchemy.ext.declarative import declarative_base
 5 # Base = declarative_base()
 6 # 每一次我们在创建数据表的时候都要做这样一件事
 7 # 然而Flask-SQLAlchemy已经为我们把 Base 封装好了
 8 
 9 # 建立User数据表
10 class Users(Base): # Base实际上就是 db.Model
11     __tablename__ = "users"
12     __table_args__ = {"useexisting": True}
13     # 在SQLAlchemy 中我们是导入了Column和数据类型 Integer 在这里
14     # 就和db.Model一样,已经封装好了
15     id = db.Column(db.Integer,primary_key=True)
16     username = db.Column(db.String(32))
17     password = db.Column(db.String(32))
18 
19 
20 if __name__ == ‘__main__‘:
21     from MyApp import create_app
22     app = create_app()
23     # 这里你要回顾一下Flask应用上下文管理了
24     # 离线脚本:
25     with app.app_context():
26         db.drop_all()
27         db.create_all()
28 
29 MyApp/models.py

5.实现一个简单的用户管理，登录视图

 1 from flask import Blueprint, request, render_template, redirect
 2 from MyApp.models import User,db
 3 
 4 users = Blueprint("user",__name__)
 5 
 6 @users.route("/add_user",methods=["POST","GET"])
 7 def add_user():
 8     if request.method == "GET":
 9         return render_template("add_user.html")
10     else:
11         username = request.form.get("username")
12         db.session.add(User(name=username))
13         db.session.commit()
14         return redirect("/user_list")
15 
16 @users.route("/user_list",methods=["POST","GET"])
17 def user_list():
18     res = User.query.all()
19     print(res)
20     return render_template("user_list.html",user_list=res)
21 
22 
23 @users.route("/del/<uid>",methods=["POST","GET"])
24 def delete_user(uid):
25     res = User.query.filter(User.id == int(uid)).delete()
26     db.session.commit()
27     print(res)
28     return redirect("/user_list")
29 
30 
31 
32 @user.route("/login",methods=["POST","GET"])
33 def user_login():
34     if request.method == "POST":
35         username = request.form.get("username")
36         password = request.form.get("password")
37 
38         # SQlalchemy需要这样做
39         # from sqlalchemy.orm import sessionmaker
40         # Session = sessionmaker(engine)
41         # db_sesson = Session()
42         # 现在不用了,因为 Flask-SQLAlchemy 也已经为我们做好会话打开的工作
43         # 我们在这里做个弊:
44         db.session.add(Users(username=username,password=password))
45         db.session.commit()
46 
47         # 然后再查询
48         user_info = Users.query.filter(Users.username == username and User.password == password).first()
49         print(user_info.username)
50         if user_info:
51             return f"登录成功{user_info.username}"
52 
53     return render_template("login.html")

其实Flask-SQLAlchemy比起SQLAlchemy更加的简单自如,用法几乎一模一样,就是在配置和启动上需要注意与Flask的配合就好啦

其实Flask-SQLAlchemy比起SQLAlchemy更加的简单自如,用法几乎一模一样,就是在配置和启动上需要注意与Flask的配合就好啦

sqlalchemy和flask-sqlalchemy

标签：label   model   nbsp   串匹配   asc   排序   orm   import   多个   

本文系统来源：https://www.cnblogs.com/fengff/p/11636823.html

内容总结

以上是互联网集市为您收集整理的sqlalchemy和flask-sqlalchemy全部内容，希望文章能够帮你解决sqlalchemy和flask-sqlalchemy所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---