您的位置: 首页  >  IT文章  >  饮冰三年-人工智能-Python-44 Python之初始SQLAlchemy SQLAlchemy：基于Python的ORM框架

饮冰三年-人工智能-Python-44 Python之初始SQLAlchemy SQLAlchemy：基于Python的ORM框架

分类: IT文章 • 2022-08-24 23:35:26

下载安装SQLAlchemy 

一、单表操作

　　1：创建表　　 

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from AlchemyDemo.OneTable.creat_table import Student
ENGINE = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")
Session = sessionmaker(ENGINE)
session = Session()
#两种修改方式，filter_by 和 filter
session.query(Student).filter_by(name='张三').update({Student.email:"zhangsan33@qq.com"})
session.query(Student).filter(Student.name=="李四").update({Student.email:"lisi44@qq.com"})
session.commit()
session.close()
create_table

　   2：查询数据　 

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from AlchemyDemo.OneTable.creat_table import Student

ENGINE = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")
Session = sessionmaker(ENGINE)
session = Session()

resEntityList = session.query(Student).all()
resEntity = session.query(Student).filter(Student.id > 0).first()  # type:Student
resEntity2 = session.query(Student).filter_by(name="李四").first()  # type:Student
for en in resEntityList:
    print(en.name, en.email)
print(resEntity.name, resEntity.email)
print(resEntity2.name, resEntity2.email)
session.close()
search_table

　　3：添加实体

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker,scoped_session
from AlchemyDemo.OneTable.creat_table import Student
import threading

ENGINE=create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")
Session=sessionmaker(bind=ENGINE)
session = Session()

# 插入单交数据
stuEntity=Student(name="张三",email="zhangsan@qq.com")
session.add(stuEntity)

# 插入多条提交数据
stuList=[Student(name="李四",email="lisi@qq.com"),
         Student(name="王五",email="wangwu@qq.com"),
         Student(name="赵六",email="zhaoliu@qq.com")]
session.add_all(stuList)
session.commit()

session.close()
add_table

　　4：修改实体　 

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from AlchemyDemo.OneTable.creat_table import Student
ENGINE = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")
Session = sessionmaker(ENGINE)
session = Session()
#两种修改方式，filter_by 和 filter
session.query(Student).filter_by(name='张三').update({Student.email:"zhangsan33@qq.com"})
session.query(Student).filter(Student.name=="李四").update({Student.email:"lisi44@qq.com"})
session.commit()
session.close()
mod_table

　　5：删除实体　 

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from AlchemyDemo.OneTable.creat_table import Student

ENGINE=create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")

Session=sessionmaker(ENGINE)
session=Session()
session.query(Student).filter(Student.name=="赵六").delete()

session.commit()
session.close()
del_table

二、一对多操作　　

　　1：创建表　　 

from sqlalchemy import create_engine
from sqlalchemy.orm import  relationship
from sqlalchemy import Column,Integer,NVARCHAR,DateTime,ForeignKey
from sqlalchemy import Index, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
import datetime

ENGINE = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")
Base=declarative_base()

#创建模型
class UserInfo(Base):
    __tablename__="user_info"
    id=Column(Integer,primary_key=True,autoincrement=True)
    name = Column(NVARCHAR(32), index=True, nullable=False)
    email = Column(NVARCHAR(32), unique=True)
    create_time = Column(DateTime, default=datetime.datetime.now)
    #ForeignKey字段的建立，需要指定能够外键绑定哪个表那个字段
    hobby_id = Column(Integer,ForeignKey("hobby.id"))
    # 方便数据查询和增加，只停留在orm关系表
    # 参数一：关联到哪个类。backref，被关联类的反向查询
    hobby=relationship("Hobby",backref="UserInfo")

    # 相当于django的orm的class Meta，是一些元信息
    __table_args__ = (
        UniqueConstraint('id', 'name', name="uni_id_name"),
        Index("name", 'email')
    )

class Hobby(Base):
    __tablename__="hobby"
    id=Column(Integer,primary_key=True,autoincrement=True)
    title=Column(NVARCHAR(300),default="码代码")

def create_db():
    Base.metadata.create_all(ENGINE)


def drop_db():
    Base.metadata.drop_all(ENGINE)



if __name__ == '__main__':
    create_db()
    # drop_db()
create_table

　　2：查询数据　　　

from sqlalchemy.orm import sessionmaker
from AlchemyDemo.OneToMany.crate_tables import UserInfo,Hobby,ENGINE

Session=sessionmaker(ENGINE)
db_session = Session()

#基于relationship的正向查询
user_obj = db_session.query(UserInfo).filter(UserInfo.name=="李四").first() #type:UserInfo
print(user_obj.email)
print(user_obj.hobby.title)

#基于relationship的反向查询
hob_obj = db_session.query(Hobby).first() #type:Hobby
for ui in hob_obj.UserInfo:
    print(ui.name)

db_session.close()
search_table

　　

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from AlchemyDemo.OneToMany.crate_tables import UserInfo, Hobby

ENGINE = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")

# 每次执行数据库操作的时候，都需要创建一个session,相当于管理器(相当于Django的ORM的objects)
Session = sessionmaker(bind=ENGINE)
# 线程安全，基于本地线程实现每个线程用同一个session
session = scoped_session(Session)

db_session = Session()
# 添加数据--方式一
hobby_obj = Hobby(title="跑步")
db_session.add(hobby_obj)
db_session.commit()  # 先保存

hobby_obj = db_session.query(Hobby).filter(Hobby.title == "跑步").first()  # 再查取

user_obj = UserInfo(name="张三", email="zhangsan@qq.com", hobby_id=hobby_obj.id)
db_session.add(user_obj)
db_session.commit()

# 添加数据--方式二 通过relationship实现正向插入
user_obj = UserInfo(name="李四", email="lisi@qq.com", hobby=Hobby(title="游泳"))
db_session.add(user_obj)
db_session.commit()

# 添加数据--方式三 通过relationship实现反向插入
hob_obj=Hobby(title="阅读",UserInfo=[UserInfo(name="王五", email="wangwu@qq.com"),UserInfo(name="赵六", email="zhaoliu@qq.com")])
db_session.add(hob_obj)
db_session.commit()
add_entity

　   4：修改和刪除操作沒有什麽特殊之处　

二、多对多操作　

　　1：创建表　　 

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy import Column,INTEGER,NVARCHAR,DateTime
from sqlalchemy import Index,UniqueConstraint,ForeignKey
from sqlalchemy.orm import relationship
import datetime

ENGINE = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemyDemo?charset=utf8")
Base=declarative_base()

class Book(Base):
    __tablename__="book"
    id=Column(INTEGER,primary_key=True,autoincrement=True)
    title=Column(NVARCHAR(64))
    # 不生成表字段 仅用于查询和增加方便
    #多对多的relationship还需要设置额外的参数secondary：绑定多对多的中间表
    tags =relationship("Tag",secondary="book2tag",backref="books")

class Tag(Base):
    __tablename__ = "tag"
    id = Column(INTEGER, primary_key=True, autoincrement=True)
    title = Column(NVARCHAR(64))

class Book2Tag(Base):
    __tablename__ = "book2tag"
    id = Column(INTEGER, primary_key=True, autoincrement=True)
    book_id = Column(INTEGER,ForeignKey("book.id"))
    tag_id = Column(INTEGER,ForeignKey("tag.id"))

def create_db():
    Base.metadata.create_all(ENGINE)

def drop_db():
    Base.metadata.drop_all(ENGINE)

if __name__ == '__main__':
    create_db()
    # drop_db()
创建表

　　2：查询数据　 

from sqlalchemy.orm import sessionmaker
from AlchemyDemo.ManyToMany.create_tables import ENGINE,Book2Tag,Book,Tag

Session = sessionmaker(ENGINE)
db_sessioin=Session()

#查询某标签对应的书
tag_obj= db_sessioin.query(Tag).filter(Tag.title=="开发语言").first() #type:Tag
for bk in tag_obj.books:
    print(bk.title)

#查询某本书对应的标签
book_obj= db_sessioin.query(Book).filter(Book.title==".Net从入门到放弃").first() #type:Book
for tg in book_obj.tags:
    print(tg.title)
查询表

　   3：添加实体

from AlchemyDemo.ManyToMany.create_tables import ENGINE,Book,Tag,Book2Tag
from sqlalchemy.orm import sessionmaker

Session=sessionmaker(ENGINE)
db_session = Session()

# 通过relationship插入
book_obj = Book(title=".Net从入门到放弃")
book_obj.tags=[Tag(title="C#"),Tag(title=".NET")]

tag_obj =Tag(title="开发语言")
tag_obj.books=[Book(title="Python全栈"),Book(title="Java入门")]

db_session.add_all([book_obj,tag_obj])
db_session.commit()
添加实体

　  4：复杂查询

from sqlalchemy.orm import sessionmaker
from AlchemyDemo.ManyToMany.create_tables import ENGINE, Tag
from sqlalchemy import or_, and_, between, any_
from sqlalchemy.sql import func

Session = sessionmaker(ENGINE)
db_session = Session()
# 条件查询
ret1 = db_session.query(Tag).filter_by(id=2).first()  # type:Tag
print("1:查询top1", ret1.title)
ret2 = db_session.query(Tag).filter(Tag.id > 1, Tag.title == ".NET").all()
for ent in ret2:
    print("2:默认and条件查询", ent.title)
ret3 = db_session.query(Tag).filter(and_(Tag.id == 3, Tag.title == ".NET")).all()
for ent in ret3:
    print("3:and条件查询", ent.title)
ret4 = db_session.query(Tag).filter(or_(Tag.id == 4, Tag.title == ".NET")).all()
for ent in ret4:
    print("4:or条件查询", ent.title)
ret5 = db_session.query(Tag).filter(Tag.id.between(2, 3)).all()
for ent in ret5:
    print("5:between条件查询", ent.title)
ret6 = db_session.query(Tag).filter(Tag.id.in_([2, 3])).all()
for ent in ret6:
    print("6:in条件查询", ent.title)
ret7 = db_session.query(Tag).filter(Tag.id.notin_([2, 3])).all()
for ent in ret7:
    print("7:not in条件查询", ent.title)

# 条件查询_模糊查询
ret8 = db_session.query(Tag).filter(Tag.title.like('%语言')).all()
for ent in ret8:
    print("8:like条件查询", ent.title)

ret9 = db_session.query(Tag).filter(~Tag.title.like('%语言')).all()
for ent in ret9:
    print("9:~like取反查询", ent.title)

# 选择查询+别名
ret10 = db_session.query(Tag.title.label('biaoti')).filter_by(id=2).first()  # type:Tag
for ent in ret10:
    print("10: 别名，只能查询标题", ent)

# 排序
ret11 = db_session.query(Tag).order_by(Tag.id.desc()).all()
for ent in ret11:
    print("11: 排序，倒序", ent.id, ent.title)

ret11 = db_session.query(Tag).order_by(Tag.id.desc()).all()
for ent in ret11:
    print("11: 排序，倒序", ent.id, ent.title)

# 分组
ret12 = db_session.query(Tag.title, func.count('*').label('count')).group_by(Tag.title).all()
for ent in ret12:
    print("12: 分组", ent.count, ent.title)

# 聚合函数
from sqlalchemy.sql import func

ret13 = db_session.query(
    Tag.title,
    func.max(Tag.id).label('maxTag'),
    func.min(Tag.id).label('minTag')
).group_by(Tag.title).having(func.max(Tag.id > 3)).all()
for ent in ret13:
    print("13: 聚合", ent.maxTag, ent.minTag, ent.title)

from AlchemyDemo.OneToMany.crate_tables import UserInfo, Hobby

# 连表查询
# print(ret16)  得到一个列表套元组 元组里是两个对象
ret14 = db_session.query(UserInfo, Hobby).filter(UserInfo.hobby_id == Hobby.id).all()
# 列表嵌套元组
for tupleRet in ret14:
    print("14: 连表", tupleRet[0].name, tupleRet[1].title)

ret15 = db_session.query(UserInfo).join(Hobby).all()
# 得到列表里面是前一个对象，join相当于inner join
for ent in ret15:
    print("15: 连表", ent.name)
# 或者直接用outerjoin也是相当于left join
ret16 = db_session.query(Hobby).outerjoin(UserInfo).all()
for ent in ret16:
    print("16: 左外连表", ent.title)

# 复杂查询
from sqlalchemy.sql import text
ret17 = db_session.query(UserInfo).filter(text("id<:value and name=:name")).params(value=9,name="张三")
for ent in ret17:
    print("17: 复杂查询，直接sql", ent.name)
复杂查询

相关推荐