sqlalchemy详解及 快速使用
快速使用
# flask 中没有orm框架,对象关系映射,方便我们快速操作数据库
# flask,fastapi中用sqlalchemy居多
# SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上,使用关系对象映射进行数据库操作,简言之便是:将类和对象转换成SQL,然后使用数据API执行SQL并获取执行结果
# 安装
pip3.8 install sqlalchemy
#了解
SQLAlchemy本身无法操作数据库,其必须以来pymsql等第三方插件
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多:http://docs.sqlalchemy.org/en/latest/dialects/index.html
原生操作的快速使用
# 第一步导入create_engine
from sqlalchemy import create_engine
# 第二步生成引擎对象
engine = create_engine(
'mysql+pymysql://root:[email protected]:3306/luffy',
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 第三步 使用引擎获取连接,操作数据库
conn=engine.raw_connection()
# 创建游标
cursor=conn.cursor()
# sql语句
cursor.execute('select * from luffy_course')
print(cursor.fetchall())
创建操作数据表
from sqlalchemy import create_engine
import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
# 执行declarative_base得到一个类
Base = declarative_base()
# 继承生成的Base类
class User(Base):
# 第四步:写字段
id = Column(Integer, primary_key=True) # 生成一列,类型是Integer,主键
name = Column(String(32), index=True, nullable=False) # name列varchar32,索引,不可为空
email = Column(String(32), unique=True)
# datetime.datetime.now不能加括号,加了括号,以后永远是当前时间
ctime = Column(DateTime, default=datetime.datetime.now)
# extra = Column(Text, nullable=True)
# 第五步:写表名 如果不写以类名为表名
__tablename__ = 'users' # 数据库表名称
# 第六步:建立联合索引,联合唯一
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一
Index('ix_id_name', 'name', 'email'), # 索引
)
class Book(Base):
__tablename__ = 'books'
id = Column(Integer, primary_key=True)
name = Column(String(32))
engine = create_engine(
'mysql+pymysql://root:[email protected]:3306/aaa',
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 把表同步到数据库中
Base.metadata.create_all(engine)
sqlalchemy快速插入数据
# 使用orm插入
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Book
# 第一步:生成engine对象
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 第二步:拿到一个Session类,传入engine
Session=sessionmaker(bind=engine)
# 第三步:拿到session对象,相当于连接对象(会话)
session=Session()
# 第四步,增加数据
book=Book(name='红楼梦',)
session.add(book)
session.commit()
session.close()
scoped_session线程安全
from sqlalchemy.orm import scoped_session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 第一步:生成engine对象
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 第二步:拿到一个Session类,传入engine
Session = sessionmaker(bind=engine)
# 线程不安全
# session = Session()
# 做成线程安全的:如何做的?
# 内部使用了local对象,取当前线程的session,如果当前线程有,就直接返回用,如果没有,创建一个,放到local中
# session 是 scoped_session 的对象
session = scoped_session(Session)
# 以后全局使用session即可,它线程安全
基本增删查改
from models import User,Book
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa",)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
# 增加 :add add_all
user = User(name='杨颖',email='[email protected]',extra='孤寡孤寡孤寡')
user1 = User(name='白鹿',email='[email protected]',extra='发发发发发发付')
# session.add(user)
session.add_all([user,user1])
session.commit()
session.close()
# 2 基本查 filter filter_by filer:写条件 filter_by:等于的值
filter
# 2.1 session.query(User) 中写表模型,可以写多个表模型(连表操作) select * from User;
# 2.2 filter 过滤条件,必须写表达式 == >= <= != select * from user where user.id=1
# 2.3 all:普通列表 first
user = session.query(User).filter(User.name == 'lqz').first()
user = session.query(User).filter(User.name != 'lqz').all()
print(user)
res = session.query(User).filter(User.id > 1).all()
print(res)
# filter_by 直接写等式 不能写成 User.name = 'lqz'
user = session.query(User).filter_by(name='lqz').first()
user = session.query(User).filter_by(id=2).first()
user = session.query(User).filter_by(id=2).first()
print(user)
# 3 删除(查到才能删) filter或filter_by查询的结果 不要all或first出来, .delete()即可
res = session.query(User).filter_by(id=2).delete()
session.commit() # 一定不要忘了
print(res) # 影响的行数
# 4 修改(查到才能改)
# 方式一:update修改
res = session.query(User).filter_by(id=3).update({"name" : "彭于晏"})
print(res)
session.commit()
# 方式二,使用对象修改
res = session.query(User).filter_by(id=3).first()
res = session.query(User).filter_by(name='zzz').first()
res.name='来来来'
print(res.id)
session.add(res) # add 如果有主键,就是修改,如果没有主键就是新增
session.commit()
# 修改方式
# 当字符串相加 djagno中使用 F 函数
res=session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)
print(res)
# 加数字(记住)
res=session.query(User).filter(User.id > 0).update({User.age: User.age + 99}, synchronize_session="evaluate")
print(res)
session.commit()
高级查询
# 4 查询: filer:写条件 filter_by:等于的值
# 4.1 查询所有 是list对象
res = session.query(User).all() # 是个普通列表
print(type(res))
print(len(res))
# 4.1.1 只查询某几个字段
res = session.query(User.name.label('xx'), User.email)
print(res) # 打出原生sql
# print(res.all())
for item in res.all():
print(item[0])
# 4.1.2 filter传的是表达式,filter_by传的是参数
res = session.query(User).filter(User.name == "刘亦菲").all()
res = session.query(User).filter(User.name != "刘亦菲").all()
res = session.query(User).filter(User.name != "刘亦菲", User.email == '[email protected]').all() # django 中使用 Q
res = session.query(User).filter_by(name='lqz099').all()
res = session.query(User).filter_by(name='lqz099',email='[email protected]').all()
print(len(res))
# 4.2 取一个 all了后是list,list 没有first方法
res = session.query(User).first()
print(res)
# 4.3 查询所有,使用占位符(了解) :value :name
# select * from user where id <20 or name=lqz099
res = session.query(User).filter(text("id<:value or name=:name")).params(value=10, name='刘亦菲').all()
print(res)
# 4.4 自定义查询(了解)
# from_statement 写纯原生sql
res=session.query(User).from_statement(text("SELECT * FROM users where email=:email")).params(email='[email protected]').all()
# print(type(res[0])) # 是book的对象,但是查的是User表 不要这样写
print(res[0].name) #
# 4.5 高级查询
# 表达式,and条件连接
res = session.query(User).filter(User.id > 1 , User.name == '杨颖').all() # and条件
print(res)
其他查询方法
# between
res = session.query(User).filter(User.id.between(1, 9), User.name == '白鹿').all()
print(res)
# in
res = session.query(User).filter(User.id.in_([1,3,4])).all()
print(res)
# ~非,除。。外
res = session.query(User).filter(~User.id.in_([1,3,4])).all()
print(res)
# 二次筛选
res = session.query(User).filter(~User.id.in_(session.query(User.id).filter_by(name='白鹿'))).all()
print(res)
# and or条件
from sqlalchemy import and_, or_
# or_包裹的都是or条件,and_包裹的都是and条件
res = session.query(User).filter(and_(User.id >= 3, User.name == '刘亦菲')).all() # and条件
res = session.query(User).filter(User.id < 3, User.name == '刘亦菲').all() # 等同于上面
res = session.query(User).filter(or_(User.id < 2, User.name == 'eric')).all()
res = session.query(User).filter(
or_(
User.id < 2,
and_(User.name == '刘亦菲', User.id > 3),
User.extra != ""
)).all()
# 通配符,以e开头,不以e开头
res = session.query(User).filter(User.email.like('%@%')).all()
# select user.id from user where user.name not like e%;
res = session.query(User.id).filter(~User.name.like('e%'))
print(res)
# 分页
# 一页2条,查第5页
res = session.query(User)[2*5:2*5+2]
# 排序,根据name降序排列(从大到小)
res = session.query(User).order_by(User.email.desc()).all()
res = session.query(Book).order_by(Book.price.desc()).all()
res = session.query(Book).order_by(Book.price.asc()).all()
# 第一个条件重复后,再按第二个条件升序排
res = session.query(User).order_by(User.name.desc(), User.id.asc())
# 分组查询 5个聚合函数
from sqlalchemy.sql import func
res = session.query(User).group_by(User.extra) # 如果是严格模式,就报错
# 分组之后取最大id,id之和,最小id 和分组的字段
res = session.query(
User.extra,
func.max(User.id),
func.sum(User.id),
func.min(User.id)).group_by(User.extra).all()
for item in res:
print(item[2])
# having 过滤
# select max(id),sum(id),min(id) from user group by user.extra having id_max>2;
res = session.query(
func.max(User.id),
func.sum(User.id),
func.min(User.id)).group_by(User.extra).having(func.max(User.id) > 2)
原生sql
# 第一步导入create_engine
from sqlalchemy import create_engine
# 第二步生成引擎对象
engine = create_engine(
'mysql+pymysql://root:[email protected]:3306/luffy',
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 第三步 使用引擎获取连接,操作数据库
conn=engine.raw_connection()
# 创建游标
cursor=conn.cursor()
# sql语句
cursor.execute('select * from luffy_course')
print(cursor.fetchall())
# 使用orm插入
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
from models import Book
# 第一步:生成engine对象
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
# 第二步:拿到一个Session类,传入engine
Session=sessionmaker(bind=engine)
# 第三步:拿到session对象,相当于连接对象(会话)
session=Session()
# 第四步,增加数据
cursor = session.execute(text('insert into books(name) values(:name)'), params={"name": '红楼梦'})
session.commit()
print(cursor.lastrowid)
session.close()
django中执行原生sql
# 选择的查询基表Book.objects.raw ,只是一个傀儡,正常查询出哪些字段,都能打印出来
def index(request):
res = Book.objects.raw('select * from app01_publish where id=1') # RawQuerySet 用起来跟列表一样
print(res[0])
print(type(res[0]))
print(res[0].name)
# book 没有addr,但是也打印出来了
print(res[0].addr)
return HttpResponse('ok')
在Django执行原生SQL有以下三种方式:extra,raw,django connection
raw语法:
models.表名.objecs.raw(sql)models.表名.objecs.raw(sql,[参数1,参数2])
一对多
models.py
# 一对多关系
from sqlalchemy import create_engine
import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship
# 第二步:执行declarative_base,得到一个类
Base = declarative_base()
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
class Person(Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
# hobby指的是tablename而不是类名
# 关联字段写在多的一方,写在Person中,跟hobby表中id字段做外键关联
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关,不会新增字段,只用于快速链表操作
# 基于对象的跨表查询:就要加这个字段,取对象 person.hobby pserson.hobby_id
# 类名,backref用于反向查询
hobby = relationship('Hobby', backref='pers') # 如果有hobby对象,拿到所有人 hobby.pers
def __repr__(self):
return self.name
engine = create_engine("mysql+pymysql://[email protected]:3306/aaa", )
# 把表同步到数据库 (把被Base管理的所有表,都创建到数据库)
Base.metadata.create_all(engine)
# 把所有表删除
# Base.metadata.drop_all(engine)
新增和基于对象的查询
from models1 import Hobby,Person
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa",)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
hobby = Hobby(caption='乒乓球')
session.add(hobby)
person = Person(name='张三')
session.add(person)
hobby=session.query(Hobby).filter(Hobby.caption=='乒乓球').first()
# person = Person(name='王五',hobby_id=hobby.id)
person = Person(name='王五',hobby_id=1)
session.add(person)
# 支持按对象的增加方式,必须加relationship 做关联
# 方式一
hobby=session.query(Hobby).filter(Hobby.caption=='乒乓球').first()
person = Person(name='赵六',hobby=hobby)
# 方式二
hobby = Hobby(caption='羽毛球') # 表中暂时没有
person = Person(name='赵六', hobby=hobby)
session.add_all([person, hobby])
session.commit()
多对多
# 一对多关系
from sqlalchemy import create_engine
import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship
# 第二步:执行declarative_base,得到一个类
Base = declarative_base()
# 中间表 手动创建
class Boy2Girl(Base):
__tablename__ = 'boy2girl'
id = Column(Integer, primary_key=True, autoincrement=True)
girl_id = Column(Integer, ForeignKey('girl.id'))
boy_id = Column(Integer, ForeignKey('boy.id'))
class Girl(Base):
__tablename__ = 'girl'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
def __str__(self):
return self.name
def __repr__(self):
return self.name
class Boy(Base):
__tablename__ = 'boy'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关,仅用于查询方便,放在哪个单表中都可以
# 方便快速查询,写了这个字段,相当于django 的manytomany,快速使用基于对象的跨表查询
girls = relationship('Girl', secondary='boy2girl', backref='boys')
def __str__(self):
return self.name
def __repr__(self):
return self.name
engine = create_engine("mysql+pymysql://[email protected]:3306/aaa", )
# 把表同步到数据库 (把被Base管理的所有表,都创建到数据库)
Base.metadata.create_all(engine)
# 把所有表删除
# Base.metadata.drop_all(engine)
增加和基于对象的跨表查询
from model2 import Girl,Boy,Boy2Girl
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa",)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
# 新增
girl = Girl(name='迪丽热巴')
boy = Boy(name='李现')
session.add_all([girl,boy])
session.add(Boy2Girl(girl_id=1,boy_id=1))
session.commit()
# 使用relationship
boy = Boy(name='lqz')
boy.girls = [Girl(name='杨颖'), Girl(name='白敬亭')]
session.add(boy)
session.commit()
# 基于对象的跨表查询
# 正向
boy = session.query(Boy).filter(Boy.id==1).first()
print(boy.girls)
# 反向
girl = session.query(Girl).filter(Girl.id==1).first()
print(girl.boys)
连表查询
### 关联关系,基于连表的跨表查询
from models1 import Person,Hobby
# 链表操作
# select * from person,hobby where person.hobby_id=hobby.id;
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all()
# 自己连表查询
# join表,默认是inner join,自动按外键关联
# select * from Person inner join Hobby on Person.hobby_id=Hobby.id;
res = session.query(Person).join(Hobby).all()
#isouter=True 外连,表示Person left join Favor,没有右连接,反过来即可
# select * from Person left join Hobby on Person.hobby_id=Hobby.id;
res = session.query(Person).join(Hobby, isouter=True).all()
# 没有right join,通过这个实现
res = session.query(Hobby).join(Person, isouter=True).all()
# # 自己指定on条件(连表条件),第二个参数,支持on多个条件,用and_,同上
# select * from Person left join Hobby on Person.id=Hobby.id;
res = session.query(Person).join(Hobby, Person.hobby_id == Hobby.id, isouter=True)
# 多对多关系连表
# 多对多关系,基于链表的跨表查
#方式一:直接连
res = session.query(Boy, Girl,Boy2Girl).filter(Boy.id == Boy2Girl.boy_id,Girl.id == Boy2Girl.girl_id).all()
# 方式二:join连
res = session.query(Boy).join(Boy2Girl).join(Girl).filter(Person.id>=2).all()
标签:engine,sqlalchemy,name,res,session,User,快速,id,详解
From: https://www.cnblogs.com/zx0524/p/17304552.html