Python 的 ORM 框架 SQLAlchemy , SQLAlchemy 没有 Django 的 Models 好用

Models 紧紧只是配置和使用比较简单, 因为他是Django自带的ORM框架,也正是因为是Django原生的,所以兼容性远远不如SQLAlchemy

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

SQLAlchemy 如何使用:

单表操作:

1.创建

from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String#创建基类,相当于Django中的 models.Model,被各个数据表类所继承Base = declarative_base()# ##################### 单表示例 ########################## 创建一张数据表class Users(Base):    __tablename__ = 'users'    id = Column(Integer, primary_key=True)    name = Column(String(32), index=True)    __table_args__ = (        # UniqueConstraint('id', 'name', name='uix_id_name'),        # Index('ix_id_name', 'name', 'extra'),    )# 创建另一张数据表class School(Base):    __tablename__ = "school"    id = Column(Integer,primary_key=True,autoincrement=True)    name = Column(String,unique=True)# 创建数据库链接engine = create_engine(        "mysql+pymysql://root:root@localhost:3306/dragon?charset=utf8",        max_overflow=0,  # 超过连接池大小外最多创建的连接        pool_size=5,  # 连接池大小        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）    )# 通过Base.metadata找到所有继承 Base 的数据表classBase.metadata.create_all(engine)# SQLAlchemy数据表进行修改后,无法直接进行更新,只能删除表后进行操作,重新进行操作创建基类和第一张数据表

　　2.增删改查相关操作:

from CreateDB import Users, School# 1. 创建一个用户添加到数据库# 创建连接from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmakerengine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/dragon?charset=utf8",                       )# 创建数据表操作对象 sessionmakerDB_session = sessionmaker(engine)db_session = DB_session()# 1.增加 操作数据表# 单行操作# obj1 = Users(name="123") # 通过 Users数据表类创建一条数据# db_session.add(obj1) # 将创建好的数据添加到 数据表操作对象的内存中,此时数据库还并不知道要添加数据# db_session.commit() # 将数据表操作对象内存中的所有数据及操作提交到数据库中# 多行操作# db_session.add_all([#     Users(name="zhangsan"),#     Users(name="lisi"),# ])# db_session.commit()# 2.查询 数据表操作# user_list = db_session.query(Users).all() # 查询所有数据# user_list = db_session.query(Users).filter(Users.id >=2 ) # 查询带有条件的数据 表达式 返回sql语句,循环依然可以获取到数据# user_list = db_session.query(Users).filter(Users.id >=2 ).all() # 查询带有条件的数据 表达式 返回数据列表# print(user_list)# for row in user_list:#     print(row.id,row.name)# 3.删除数据 数据表操作# db_session.query(Users).filter(Users.id == 1).delete() # 删除带有条件的数据# db_session.commit()# 4.修改# db_session.query(Users).filter(Users.id == 3).update({"name":"alexDSB"}) # 更新id=3的数据# db_session.commit()# 关闭连接db_session.close()增删改查

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

from CreateDB import Users, School# 1. 创建一个用户添加到数据库# 创建连接from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmakerengine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/dragon?charset=utf8",                       )# 创建数据表操作对象 sessionmakerDB_session = sessionmaker(engine)db_session = DB_session()# 查询数据表操作"""r1 = session.query(Users).all()r2 = session.query(Users.name.label('xx'), Users.age).all()r3 = session.query(Users).filter(Users.name == "alex").all()r4 = session.query(Users).filter_by(name='alex').all()r5 = session.query(Users).filter_by(name='alex').first()r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all()"""# 筛选查询列# user_list = db_session.query(Users.name).all()# print(user_list) # [('alexDSB',), ('zhangsan',)] 虽然看到的是元祖,但是依然可以通过对象打点儿调用属性# for row in user_list:#     print(row.name)# 别名映射  name as nick# user_list = db_session.query(Users.name.label("nick")).all()# print(user_list) # [('alexDSB',), ('zhangsan',)] 虽然看到的是元祖,但是依然可以通过对象打点儿调用属性# for row in user_list:#     print(row.nick) # 这里要写别名了# 筛选条件格式# user_list = db_session.query(Users).filter(Users.name == "alexDSB").all()# user_list = db_session.query(Users).filter(Users.name == "alexDSB").first()# user_list = db_session.query(Users).filter_by(name="alexDSB").first()# for row in user_list:#     print(row.nick)# 复杂查询# from sqlalchemy.sql import text# user_list = db_session.query(Users).filter(text("id<:value and name=:name")).params(value=3,name="alexDSB")# 查询语句# from sqlalchemy.sql import text# user_list = db_session.query(Users).filter(text("select * from users id<:value and name=:name")).params(value=3,name="alexDSB")# 排序 :# user_list = db_session.query(Users).order_by(Users.id).all()# user_list = db_session.query(Users).order_by(Users.id.desc()).all()# for row in user_list:#     print(row.name,row.id)#其他查询条件"""ret = session.query(Users).filter_by(name='alex').all()ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() # between 大于1小于3的ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() 子查询from sqlalchemy import and_, or_ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()ret = session.query(Users).filter(    or_(        Users.id < 2,        and_(Users.name == 'eric', Users.id > 3),        Users.extra != ""    )).all()# select * from users where id<2 or (name="eric" and id>3) or extra != "" """# 关闭连接db_session.close()高级版查询操作

4.高级版更新操作

"""db_session.query(Users).filter(Users.id > 0).update({
   "name" : "099"})db_session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)db_session.query(Users).filter(Users.id > 0).update({
   "age": Users.age + 1}, synchronize_session="evaluate")db_session.commit()"""高级版更新操作

5.扩展内容

# 通配符ret = session.query(Users).filter(Users.name.like('e%')).all()ret = session.query(Users).filter(~Users.name.like('e%')).all()# 限制ret = session.query(Users)[1:2]# 排序ret = session.query(Users).order_by(Users.name.desc()).all()ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()# 分组from sqlalchemy.sql import funcret = session.query(Users).group_by(Users.extra).all()ret = session.query(    func.max(Users.id),    func.sum(Users.id),    func.min(Users.id)).group_by(Users.name).all()ret = session.query(    func.max(Users.id),    func.sum(Users.id),    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()排序分组选取通配符

 

一对多的操作 : ForeignKey

1.创建表:

from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKeyfrom 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:DragonFire@localhost:3306/dragon?charset=utf8",        max_overflow=0,  # 超过连接池大小外最多创建的连接        pool_size=5,  # 连接池大小        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）    )# 通过Base.metadata找到所有继承 Base 的数据表classBase.metadata.create_all(engine)创建多表并建立关系

 

2.增删改查

from SQLAlchemy_ready.ss2_ForeignKey_relationship import Author, Book# 1. 创建一个用户添加到数据库# 创建连接from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmakerengine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8",                       )# 创建数据表操作对象 sessionmakerDB_session = sessionmaker(engine)db_session = DB_session()# 1.增加 操作数据表# 添加两个作者:# db_session.add_all([#     Author(name="alex"),#     Author(name="yinwangba")# ])# db_session.commit()# 添加一本书 jinpingmei 作者是 yinwangba# author = db_session.query(Author).filter(Author.name == "yinwangba").first()# db_session.add(Book(name="jinpingmei",author_id=author.id))# db_session.commit()# 2.查询所有数据,并显示作者名称,连表查询# book_list = db_session.query(Book).all()# for row in book_list:#     print(row.name,row.author_id)# book_list = db_session.query(Book.name.label("bname"),Author.name.label ("aname")).join(Author,Book.author_id == Author.id,isouter=True).all()# print(book_list)# for row in book_list:#     print(row.aname,row.bname)# 查询之relationship 快速连表# 创建表的时候加入 relationship#普通版添加# obj = Author(name="yinwangba")# db_session.add(obj)# db_session.commit()# print(obj.id,obj.name)## obj_book = Book(name="jinpingmei",author_id=obj.id)# db_session.add(obj_book)# db_session.commit()# obj = Author(name="yinwangba")# relationship版 添加# bobj = Book(name="jinpingmei",user=Author(name="yinwangba"))# db_session.add(bobj)# db_session.commit()# 查询之relationship 快速连表# book_list = db_session.query(Book).all()# for row in book_list:#     print(row.name,row.user.name)# 查询作者的所有书籍# obj = db_session.query(Author).filter(Author.name=="yinwangba").first()# print(obj.author2book)# 反向字段添加# author_obj = Author(name="alex")# author_obj.author2book = [Book(name="儒林外史之银林大战"),Book(name="邻家小妹妹")]# db_session.add(author_obj)# db_session.commit()# 关闭连接db_session.close()增删改查