SQLAlchemy教学版

Posted 2021-01-29 DragonFire

好久没有更新Blog了

今天来聊一聊 Python 的 ORM 框架 SQLAlchemy 有些同学已经听说过这个框架的大名了,也听说了 SQLAlchemy 没有 Django 的 Models 好用

我在这里官宣辟谣一下啊, Models 紧紧只是配置和使用比较简单(这特么就是废话),因为他是Django自带的ORM框架,也正是因为是Django原生的,所以兼容性远远不如SQLAlchemy

真正算得上全面的ORM框架必然是我们的SQLAlchemy ORM框架,它可以在任何使用SQL查询时使用

当然了,无论是使用什么ORM框架,都是为了方便不熟练数据库的同学使用的,我个人还是比较推崇原生 SQL ,也建议每位同学攻克 SQL 的难关

废话不多说,我们来看一下 SQLAlchemy 如何使用:

单表操作:

1.创建

 1 from sqlalchemy import create_engine
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy import Column, Integer, String
 4 
 5 #创建基类,相当于Django中的 models.Model,被各个数据表类所继承
 6 Base = declarative_base()
 7 
 8 
 9 # ##################### 单表示例 #########################
10 # 创建一张数据表
11 class Users(Base):
12     __tablename__ = ‘users‘
13 
14     id = Column(Integer, primary_key=True)
15     name = Column(String(32), index=True)
16 
17     __table_args__ = (
18         # UniqueConstraint(‘id‘, ‘name‘, name=‘uix_id_name‘),
19         # Index(‘ix_id_name‘, ‘name‘, ‘extra‘),
20     )
21 
22 # 创建另一张数据表
23 class School(Base):
24     __tablename__ = "school"
25 
26     id = Column(Integer,primary_key=True,autoincrement=True)
27     name = Column(String,unique=True)
28 
29 
30 # 创建数据库链接
31 engine = create_engine(
32         "mysql+pymysql://root:[email protected]:3306/dragon?charset=utf8",
33         max_overflow=0,  # 超过连接池大小外最多创建的连接
34         pool_size=5,  # 连接池大小
35         pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
36         pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
37     )
38 
39 
40 # 通过Base.metadata找到所有继承 Base 的数据表class
41 Base.metadata.create_all(engine)
42 
43 # SQLAlchemy数据表进行修改后,无法直接进行更新,只能删除表后进行操作,重新进行操作

创建基类和第一张数据表

2.增删改查相关操作:

 1 from CreateDB import Users, School
 2 
 3 # 1. 创建一个用户添加到数据库
 4 # 创建连接
 5 from sqlalchemy import create_engine
 6 from sqlalchemy.orm import sessionmaker
 7 engine = create_engine("mysql+pymysql://root:[email protected]:3306/dragon?charset=utf8",
 8                        )
 9 
10 # 创建数据表操作对象 sessionmaker
11 DB_session = sessionmaker(engine)
12 db_session = DB_session()
13 
14 # 1.增加 操作数据表
15 # 单行操作
16 # obj1 = Users(name="123") # 通过 Users数据表类创建一条数据
17 # db_session.add(obj1) # 将创建好的数据添加到 数据表操作对象的内存中,此时数据库还并不知道要添加数据
18 # db_session.commit() # 将数据表操作对象内存中的所有数据及操作提交到数据库中
19 # 多行操作
20 # db_session.add_all([
21 #     Users(name="zhangsan"),
22 #     Users(name="lisi"),
23 # ])
24 # db_session.commit()
25 
26 # 2.查询 数据表操作
27 # user_list = db_session.query(Users).all() # 查询所有数据
28 # user_list = db_session.query(Users).filter(Users.id >=2 ) # 查询带有条件的数据 表达式 返回sql语句,循环依然可以获取到数据
29 # user_list = db_session.query(Users).filter(Users.id >=2 ).all() # 查询带有条件的数据 表达式 返回数据列表
30 # print(user_list)
31 # for row in user_list:
32 #     print(row.id,row.name)
33 
34 # 3.删除数据 数据表操作
35 # db_session.query(Users).filter(Users.id == 1).delete() # 删除带有条件的数据
36 # db_session.commit()
37 
38 # 4.修改
39 # db_session.query(Users).filter(Users.id == 3).update({"name":"alexDSB"}) # 更新id=3的数据
40 # db_session.commit()
41 
42 # 关闭连接
43 db_session.close()

增删改查

3.搞基高几高技告急膏剂高集高吉高级版查询操作

 1 from CreateDB import Users, School
 2 
 3 # 1. 创建一个用户添加到数据库
 4 # 创建连接
 5 from sqlalchemy import create_engine
 6 from sqlalchemy.orm import sessionmaker
 7 engine = create_engine("mysql+pymysql://root:[email protected]7.0.0.1:3306/dragon?charset=utf8",
 8                        )
 9 
10 # 创建数据表操作对象 sessionmaker
11 DB_session = sessionmaker(engine)
12 db_session = DB_session()
13 
14 # 查询数据表操作
15 """
16 r1 = session.query(Users).all()
17 r2 = session.query(Users.name.label(‘xx‘), Users.age).all()
18 r3 = session.query(Users).filter(Users.name == "alex").all()
19 r4 = session.query(Users).filter_by(name=‘alex‘).all()
20 r5 = session.query(Users).filter_by(name=‘alex‘).first()
21 r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name=‘fred‘).order_by(Users.id).all()
22 r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name=‘ed‘).all()
23 """
24 # 筛选查询列
25 # user_list = db_session.query(Users.name).all()
26 # print(user_list) # [(‘alexDSB‘,), (‘zhangsan‘,)] 虽然看到的是元祖,但是依然可以通过对象打点儿调用属性
27 # for row in user_list:
28 #     print(row.name)
29 
30 # 别名映射  name as nick
31 # user_list = db_session.query(Users.name.label("nick")).all()
32 # print(user_list) # [(‘alexDSB‘,), (‘zhangsan‘,)] 虽然看到的是元祖,但是依然可以通过对象打点儿调用属性
33 # for row in user_list:
34 #     print(row.nick) # 这里要写别名了
35 
36 # 筛选条件格式
37 # user_list = db_session.query(Users).filter(Users.name == "alexDSB").all()
38 # user_list = db_session.query(Users).filter(Users.name == "alexDSB").first()
39 # user_list = db_session.query(Users).filter_by(name="alexDSB").first()
40 # for row in user_list:
41 #     print(row.nick)
42 
43 # 复杂查询
44 # from sqlalchemy.sql import text
45 # user_list = db_session.query(Users).filter(text("id<:value and name=:name")).params(value=3,name="alexDSB")
46 
47 # 查询语句
48 # from sqlalchemy.sql import text
49 # user_list = db_session.query(Users).filter(text("select * from users id<:value and name=:name")).params(value=3,name="alexDSB")
50 
51 # 排序 :
52 # user_list = db_session.query(Users).order_by(Users.id).all()
53 # user_list = db_session.query(Users).order_by(Users.id.desc()).all()
54 # for row in user_list:
55 #     print(row.name,row.id)
56 
57 #其他查询条件
58 """
59 ret = session.query(Users).filter_by(name=‘alex‘).all()
60 ret = session.query(Users).filter(Users.id > 1, Users.name == ‘eric‘).all()
61 ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == ‘eric‘).all() # between 大于1小于3的
62 ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
63 ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
64 ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=‘eric‘))).all() 子查询
65 from sqlalchemy import and_, or_
66 ret = session.query(Users).filter(and_(Users.id > 3, Users.name == ‘eric‘)).all()
67 ret = session.query(Users).filter(or_(Users.id < 2, Users.name == ‘eric‘)).all()
68 ret = session.query(Users).filter(
69     or_(
70         Users.id < 2,
71         and_(Users.name == ‘eric‘, Users.id > 3),
72         Users.extra != ""
73     )).all()
74 # select * from users where id<2 or (name="eric" and id>3) or extra != "" 
75 """
76 
77 # 关闭连接
78 db_session.close()

高级版查询操作

4.高级版更新操作

1 """
2 db_session.query(Users).filter(Users.id > 0).update({"name" : "099"})
3 db_session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
4 db_session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
5 db_session.commit()
6 """

高级版更新操作

5.扩展内容

 1 # 通配符
 2 ret = session.query(Users).filter(Users.name.like(‘e%‘)).all()
 3 ret = session.query(Users).filter(~Users.name.like(‘e%‘)).all()
 4 
 5 # 限制
 6 ret = session.query(Users)[1:2]
 7 
 8 # 排序
 9 ret = session.query(Users).order_by(Users.name.desc()).all()
10 ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
11 
12 # 分组
13 from sqlalchemy.sql import func
14 
15 ret = session.query(Users).group_by(Users.extra).all()
16 ret = session.query(
17     func.max(Users.id),
18     func.sum(Users.id),
19     func.min(Users.id)).group_by(Users.name).all()
20 
21 ret = session.query(
22     func.max(Users.id),
23     func.sum(Users.id),
24     func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

排序分组选取通配符

 

一对多的操作 : ForeignKey

1.创建表:

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

#创建基类,相当于Django中的 models.Model,被各个数据表类所继承
Base = declarative_base()


# ##################### 多表示例 #########################
# 创建一张数据表
class Author(Base):
    __tablename__ = ‘author‘

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)


#  创建另一张数据表
class Book(Base):
    __tablename__ = ‘book‘

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)

    author_id = Column(Integer,ForeignKey("author.id"))

    # relationshi 不会单独生成字段 ,只用于增加查询操作
    user = relationship("Author",backref="author2book") # backref 反向查找的名字




# 创建数据库链接
engine = create_engine(
        "mysql+pymysql://root:[email protected]:3306/dragon?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )

# 通过Base.metadata找到所有继承 Base 的数据表class
Base.metadata.create_all(engine)

创建多表并建立关系

2.增删改查

 1 from SQLAlchemy_ready.ss2_ForeignKey_relationship import Author, Book
 2 
 3 # 1. 创建一个用户添加到数据库
 4 # 创建连接
 5 from sqlalchemy import create_engine
 6 from sqlalchemy.orm import sessionmaker
 7 engine = create_engine("mysql+pymysql://root:[email protected]:3306/dragon?charset=utf8",
 8                        )
 9 
10 # 创建数据表操作对象 sessionmaker
11 DB_session = sessionmaker(engine)
12 db_session = DB_session()
13 
14 # 1.增加 操作数据表
15 # 添加两个作者:
16 # db_session.add_all([
17 #     Author(name="alex"),
18 #     Author(name="yinwangba")
19 # ])
20 # db_session.commit()
21 # 添加一本书 jinpingmei 作者是 yinwangba
22 # author = db_session.query(Author).filter(Author.name == "yinwangba").first()
23 # db_session.add(Book(name="jinpingmei",author_id=author.id))
24 # db_session.commit()
25 
26 # 2.查询所有数据,并显示作者名称,连表查询
27 # book_list = db_session.query(Book).all()
28 # for row in book_list:
29 #     print(row.name,row.author_id)
30 
31 # book_list = db_session.query(Book.name.label("bname"),Author.name.label ("aname")).join(Author,Book.author_id == Author.id,isouter=True).all()
32 # print(book_list)
33 # for row in book_list:
34 #     print(row.aname,row.bname)
35 
36 # 查询之relationship 快速连表
37 # 创建表的时候加入 relationship
38 #普通版添加
39 # obj = Author(name="yinwangba")
40 # db_session.add(obj)
41 # db_session.commit()
42 # print(obj.id,obj.name)
43 #
44 # obj_book = Book(name="jinpingmei",author_id=obj.id)
45 # db_session.add(obj_book)
46 # db_session.commit()
47 # obj = Author(name="yinwangba")
48 
49 # relationship版 添加
50 # bobj = Book(name="jinpingmei",user=Author(name="yinwangba"))
51 # db_session.add(bobj)
52 # db_session.commit()
53 
54 # 查询之relationship 快速连表
55 # book_list = db_session.query(Book).all()
56 # for row in book_list:
57 #     print(row.name,row.user.name)
58 
59 # 查询作者的所有书籍
60 # obj = db_session.query(Author).filter(Author.name=="yinwangba").first()
61 # print(obj.author2book)
62 
63 # 反向字段添加
64 # author_obj = Author(name="alex")
65 # author_obj.author2book = [Book(name="儒林外史之银林大战"),Book(name="邻家小妹妹")]
66 # db_session.add(author_obj)
67 # db_session.commit()
68 
69 # 关闭连接
70 db_session.close()

增删改查

 

多对多 : ManyToMany