SQLAlchemy框架详解
概述：SQLAlchemy是⼀个基于Python实现的ORM框架。该框架建⽴在 DB API之上，使⽤关系对象映射进⾏数据库操作，简⾔之便是：将类和对象转换成SQL，然后使⽤数据API执⾏SQL并获取执⾏结果
1. 快速使⽤
（1）创建模型类⽣成数据表
declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64))
extra = Column(String(16))
if __name__ == '__main__':
# 数据库连接相关
engine = create_engine("mysql+pymysql://root:149789@127.0.0.1:3306/s8day127")
# # 创建表
# 删除表
（2）导⼊模型类，新增记录
from sqlalchemy import create_engine
import sessionmaker
from models import User
engine = create_engine("mysql+pymysql://root:149789@127.0.0.1:3306/s8day127", max_overflow=5) Session = sessionmaker(bind=engine)
session = Session()
user = User(name="admin", extra="110")
user2 = User(name="admin2", extra="119")
session.add(user)
session.add(user2)
sessionmit()
2. 使⽤sqlalchemy执⾏原⽣sql语句，sqlalchemy⽀持数据库连接池
from sqlalchemy import create_engine
# engine = create_engine("mysql+pymysql://root:149789@localhost:3306/s8day127")
engine = create_engine(
"mysql+pymysql://root:149789@127.0.0.1:3306/s8day127?charset=utf8",
max_overflow=0, # 超过连接池⼤⼩外最多创建的连接
pool_size=5, # 连接池⼤⼩
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进⾏⼀次连接的回收（重置）,-1代表⼀直使⽤
)
cursor = ute("select * from users;")
ret = cursor.fetchall()
print(ret)
3. SQLAlchemy创建表结构
from datetime import datetime
declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, UniqueConstraint Base = declarative_base()
class Classes(Base):
__tablename__ = "classes"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
class Student(Base):
__tablename__ = "student"
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(64), nullable=False, unique=True)
password = Column(String(64), nullable=False)
ctime = Column(DateTime, nullable=False, w)
class_id = Column(Integer, ForeignKey("classes.id"))
cls = relationship("Classes", backref="stus")
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
class Student2Hobby(Base):
__tablename__ = 'student2hobby'
id = Column(Integer, primary_key=True, autoincrement=True)
student_id = Column(Integer, ForeignKey('student.id'))
hobby_id = Column(Integer, ForeignKey('hobby.id'))
# 联合唯⼀索引
__table_args__ = (
UniqueConstraint("student_id", "hobby_id", name="uix_student_id_hobby_id"),
)
if __name__ == '__main__':
# 数据库连接相关
engine = create_engine("mysql+pymysql://root:149789@127.0.0.1:3306/s8day127")
# # 创建表
# 删除表
# adata.drop_all(engine)
4. sqlalchemy增删改查、⼦查询、关联查询、构造sql语句查询等
from sqlalchemy import create_engine, text, or_
import sessionmaker
from models import Classes
engine = create_engine("mysql+pymysql://root:149789@127.0.0.1:3306/s8day127?charset=utf8") Session = sessionmaker(bind=engine)
session = Session()
# 1. （1）单条增加
# 1. （1）单条增加
obj = Classes(name="98班")
session.add(obj)
sessionmit()
# 1. （2）多条增加
objs = [
Classes(name="101班"),
Classes(name="102班"),
]
session.add_all(objs)
sessionmit()
# 2.
# （1）修改
obj = session.query(Classes).get(3)
obj.name = "201班"
sessionmit()
# （2）修改
session.query(Classes).filter(Classes.id > 3).update({"name": "002班"})
# 参数：synchronize_session=False时是字符串拼接，="evaluate"时是数值运算
session.query(Classes).filter(Classes.id > 3).update({Classes.name: "1" + Classes.name}, synchronize_session=False) session.query(Classes).filter(Classes.id > 3).update({Classes.id: Classes.id - 40}, synchronize_session="evaluate") sessionmit()
# 3. 删除
obj = session.query(Classes).filter(Classes.id == 2).delete()
sessionmit()
# 4. 查询
# 1. .label就是给字段取别名
objs = session.query(Classes.id, Classes.name.label("Name")).filter(Classes.id > 3).all()
for obj in objs:
print(obj.id, obj.Name)
# 2. 打印obj虽然看上去是个元组，但是它并不是⼀个元组，它的类型是个对象：<class 'w.Row'> objs = session.query(Classes.id, Classes).filter(Classes.id > 3).all()
for obj in objs:
print(type(obj))
print(obj.id, obj[1].name)
# 3. filter和filter_by, filter传递的是表达式，⽽filter_by传递的是关键字参数
# filter_by是依赖于filter的, 如果筛选条件是字典的话，filter_by更⽅便
obj = session.query(Classes).filter(Classes.id == 3).first()
print(obj.name)
obj2 = session.query(Classes).filter_by(**{"id": 10}).all()
sql语句查询结果取反obj2 = session.query(Classes).filter_by(id=10).all()
print(obj2[0].name)
# 4. filter构造复杂的查询条件
objs = session.query(Classes).filter(or_(text("id>:value and name=:name"), text("id=:value2 and name=:name2")))\
.params(value=1, name="11002班", value2=3, name2="201班").all()
print([obj.name for obj in objs])
# 5. from_statement构造sql语句
objs = session.query(Classes).from_statement(text("select * from classes where name=:name")).params(name="201班") print(objs)
print([obj.name for obj in objs])
# 6. ⼦查询
objs = session.query(Classes).filter(Classes.id.in_(
session.query(Classes.id).filter(Classes.name == "201班")
))
print([obj.name for obj in objs])
5. 带外键的关联查询
from sqlalchemy import create_engine, text, or_
import sessionmaker
from models import Classes, Student
engine = create_engine("mysql+pymysql://root:149789@127.0.0.1:3306/s8day127?charset=utf8") Session = sessionmaker(bind=engine)
session = Session()
# ⼀：关联查询：不带外键的查询⽅法
# 答应学⽣信息，包含班级名称⽅法⼀：
objs = session.query(Student).all()
class_ids = {obj.class_id for obj in objs}
classes_objs = session.query(Classes).filter(Classes.id.in_(class_ids))
id_map_name = {obj.id: obj.name for obj in classes_objs}
for obj in objs:
print(obj.id, obj.username, id_map_name[obj.class_id])
# ⼆：关联查询：带外键的查询⽅法 join
# 答应学⽣信息，包含班级名称⽅法⼆：
objs = session.query(Student.id, Student.username, Classes.name).join(Classes, isouter=True).all() for obj in objs:
print(obj.id, obj.username, obj.name)
"""
class Student(Base):
__tablename__ = "student"
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(64), nullable=False, unique=True)
password = Column(String(64), nullable=False)
ctime = Column(DateTime, nullable=False, w)
class_id = Column(Integer, ForeignKey("classes.id"))
cls = relationship("Classes", backref="stus")
"""
# 三：关联查询：带外键的查询⽅法通过relationship
# 由多表查⼀表：答应学⽣信息，包含班级名称⽅法三：
objs = session.query(Student).all()
for obj in objs:
print(obj.id, obj.username, obj.cls.name)
# 四：由⼀查多，到201班所有的学⽣
obj = session.query(Classes).filter_by(**{"name": "201班"}).first()
for stu_obj in obj.stus:
print(obj.id, obj.name, stu_obj.username)
6. 其它查询
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论