受欢迎的唐山网站建设,SUPERW上海网站建设工作室,怎么做原创短视频网站,网站建设的开源平台SQLAlchemy
1.介绍
SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上#xff0c;使用关系对象映射进行数据库操作#xff0c;简言之便是#xff1a;将类和对象转换成SQL#xff0c;然后使用数据API执行SQL并获取执行结果。
pip3 install sqlalchemy组…
SQLAlchemy
1.介绍
SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上使用关系对象映射进行数据库操作简言之便是将类和对象转换成SQL然后使用数据API执行SQL并获取执行结果。
pip3 install sqlalchemy组成部分
Engine框架的引擎
Connection Pooling 数据库连接池
Dialect选择连接数据库的DB API种类
Schema/Types架构和类型
SQL Exprression LanguageSQL表达式语言SQLAlchemy本身无法操作数据库其必须以来pymsql等第三方插件Dialect用于和数据API进行交流根据配置文件的不同调用不同的数据库API从而实现对数据库的操作如
MySQL-Pythonmysqlmysqldb://user:passwordhost[:port]/dbnamepymysqlmysqlpymysql://username:passwordhost/dbname[?options]MySQL-Connectormysqlmysqlconnector://user:passwordhost[:port]/dbnamecx_Oracleoraclecx_oracle://user:passhost:port/dbname[?keyvaluekeyvalue...]更多http://docs.sqlalchemy.org/en/latest/dialects/index.htmldjango中如何反向生成models
python manage.py inspectdb app/models.py2.简单使用能创建表删除表不能修改表
修改表在数据库添加字段类对应上
1执行原生sql不常用
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engineengine create_engine(mysqlpymysql://root:123456127.0.0.1:3306/test?charsetutf8,max_overflow0, # 超过连接池大小外最多创建的连接pool_size5, # 连接池大小pool_timeout30, # 池中没有线程最多等待的时间否则报错pool_recycle-1 # 多久之后对线程池中的线程进行一次连接的回收重置
)
def task(arg):conn engine.raw_connection()cursor conn.cursor()cursor.execute(select * from app01_book)result cursor.fetchall()print(result)cursor.close()conn.close()for i in range(20):t threading.Thread(targettask, args(i,))t.start()2 orm使用
models.py
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base declarative_base()class Users(Base):__tablename__ users # 数据库表名称id Column(Integer, primary_keyTrue) # id 主键name Column(String(32), indexTrue, nullableFalse) # name列索引不可为空# email Column(String(32), uniqueTrue)#datetime.datetime.now不能加括号加了括号以后永远是当前时间# ctime Column(DateTime, defaultdatetime.datetime.now)# extra Column(Text, nullableTrue)__table_args__ (# UniqueConstraint(id, name, nameuix_id_name), #联合唯一# Index(ix_id_name, name, email), #索引)def init_db():根据类创建数据库表:return:engine create_engine(mysqlpymysql://root:123456127.0.0.1:3306/aaa?charsetutf8,max_overflow0, # 超过连接池大小外最多创建的连接pool_size5, # 连接池大小pool_timeout30, # 池中没有线程最多等待的时间否则报错pool_recycle-1 # 多久之后对线程池中的线程进行一次连接的回收重置)Base.metadata.create_all(engine)def drop_db():根据类删除数据库表:return:engine create_engine(mysqlpymysql://root:123456127.0.0.1:3306/aaa?charsetutf8,max_overflow0, # 超过连接池大小外最多创建的连接pool_size5, # 连接池大小pool_timeout30, # 池中没有线程最多等待的时间否则报错pool_recycle-1 # 多久之后对线程池中的线程进行一次连接的回收重置)Base.metadata.drop_all(engine)if __name__ __main__:# drop_db()init_db()app.py
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#mysqlpymysql://root127.0.0.1:3306/aaa
engine create_engine(mysqlpymysql://root:123456127.0.0.1:3306/aaa, max_overflow0, pool_size5)
Connection sessionmaker(bindengine)# 每次执行数据库操作时都需要创建一个Connection
con Connection()# ############# 执行ORM操作 #############
obj1 Users(namelqz)
con.add(obj1)
# 提交事务
con.commit()# 关闭session其实是将连接放回连接池
con.close()3.一对多关系
class Hobby(Base):__tablename__ hobbyid Column(Integer, primary_keyTrue)caption Column(String(50), default篮球)class Person(Base):__tablename__ personnid Column(Integer, primary_keyTrue)name Column(String(32), indexTrue, nullableTrue)# hobby指的是tablename而不是类名uselistFalsehobby_id Column(Integer, ForeignKey(hobby.id))# 跟数据库无关不会新增字段只用于快速链表操作# 类名backref用于反向查询hobbyrelationship(Hobby,backrefpers)4.多对多关系
class Boy2Girl(Base):__tablename__ boy2girlid Column(Integer, primary_keyTrue, autoincrementTrue)girl_id Column(Integer, ForeignKey(girl.id))boy_id Column(Integer, ForeignKey(boy.id))class Girl(Base):__tablename__ girlid Column(Integer, primary_keyTrue)name Column(String(64), uniqueTrue, nullableFalse)class Boy(Base):__tablename__ boyid Column(Integer, primary_keyTrue, autoincrementTrue)hostname Column(String(64), uniqueTrue, nullableFalse)# 与生成表结构无关仅用于查询方便,放在哪个单表中都可以servers relationship(Girl, secondaryboy2girl, backrefboys)5.操作数据表
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Usersengine create_engine(mysqlpymysql://root:123456127.0.0.1:3306/aaa, max_overflow0, pool_size5)
Session sessionmaker(bindengine)# 每次执行数据库操作时都需要创建一个session
session Session()# ############# 执行ORM操作 #############
obj1 Users(namelqz)
session.add(obj1)# 提交事务
session.commit()
# 关闭session
session.close()6.基于scoped_session实现线程安全
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Usersengine create_engine(mysqlpymysql://root:123127.0.0.1:3306/s6, max_overflow0, pool_size5)
Session sessionmaker(bindengine)
# 线程安全基于本地线程实现每个线程用同一个session
# 特殊的scoped_session中有原来方法的Session中的一下方法public_methods (__contains__, __iter__, add, add_all, begin, begin_nested,close, commit, connection, delete, execute, expire,expire_all, expunge, expunge_all, flush, get_bind,is_modified, bulk_save_objects, bulk_insert_mappings,bulk_update_mappings,merge, query, refresh, rollback,scalar
)#scoped_session类并没有继承Session,但是却又它的所有方法
session scoped_session(Session)
# ############# 执行ORM操作 #############
obj1 Users(namealex1)
session.add(obj1)# 提交事务
session.commit()
# 关闭session
session.close()7.基本增删查改
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import textfrom db import Users, Hostsengine create_engine(mysqlpymysql://root:123127.0.0.1:3306/s6, max_overflow0, pool_size5)
Session sessionmaker(bindengine)session Session()# ################ 添加 ################obj1 Users(namewupeiqi)
session.add(obj1)session.add_all([Users(namelqz),Users(nameegon),Hosts(namec1.com),
])
session.commit()
# ################ 删除 ################session.query(Users).filter(Users.id 2).delete()
session.commit()# ################ 修改 #################传字典
session.query(Users).filter(Users.id 0).update({name : lqz})
#类似于django的F查询
session.query(Users).filter(Users.id 0).update({Users.name: Users.name 099}, synchronize_sessionFalse)
session.query(Users).filter(Users.id 0).update({age: Users.age 1}, synchronize_sessionevaluate)
session.commit()# ################ 查询 ################r1 session.query(Users).all()
#只取age列把name重命名为xx
r2 session.query(Users.name.label(xx), Users.age).all()
#filter传的是表达式filter_by传的是参数
r3 session.query(Users).filter(Users.name lqz).all()
r4 session.query(Users).filter_by(namelqz).all()
r5 session.query(Users).filter_by(namelqz).first()
#:value 和:name 相当于占位符用params传参数
r6 session.query(Users).filter(text(id:value and name:name)).params(value224, namefred).order_by(Users.id).all()
#自定义查询sql
r7 session.query(Users).from_statement(text(SELECT * FROM users where name:name)).params(nameed).all()
#增删改都要commit()
session.close()8.常用操作
# 条件
ret session.query(Users).filter_by(namelqz).all()
#表达式and条件连接
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()
#注意下划线
ret session.query(Users).filter(Users.id.in_([1,3,4])).all()
#~非除。。外
ret session.query(Users).filter(~Users.id.in_([1,3,4])).all()
#二次筛选
ret session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(nameeric))).all()
from sqlalchemy import and_, or_
#or_包裹的都是or条件and_包裹的都是and条件
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()# 通配符以e开头不以e开头
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]# 排序根据name降序排列从大到小
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()
#分组之后取最大idid之和最小id
ret session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).all()
#haviing筛选
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()# 连表默认用forinkey关联ret session.query(Users, Favor).filter(Users.id Favor.nid).all()
#join表默认是inner join
ret session.query(Person).join(Favor).all()
#isouterTrue 外连表示Person left join Favor没有右连接反过来即可
ret session.query(Person).join(Favor, isouterTrue).all()
#打印原生sql
aasession.query(Person).join(Favor, isouterTrue)
print(aa)
# 自己指定on条件连表条件,第二个参数支持on多个条件用and_,同上
ret session.query(Person).join(Favor,Person.idFavor.id, isouterTrue).all()
# 组合了解UNION 操作符用于合并两个或多个 SELECT 语句的结果集
#union和union all的区别
q1 session.query(Users.name).filter(Users.id 2)
q2 session.query(Favor.caption).filter(Favor.nid 2)
ret q1.union(q2).all()q1 session.query(Users.name).filter(Users.id 2)
q2 session.query(Favor.caption).filter(Favor.nid 2)
ret q1.union_all(q2).all()9.执行原生sql
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hostsengine create_engine(mysqlpymysql://root:123127.0.0.1:3306/s6, max_overflow0, pool_size5)
Session sessionmaker(bindengine)session Session()# 查询
# cursor session.execute(select * from users)
# result cursor.fetchall()# 添加
cursor session.execute(insert into users(name) values(:value),params{value:lqz})
session.commit()
print(cursor.lastrowid)session.close()10.一对多
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Personengine create_engine(mysqlpymysql://root:123127.0.0.1:3306/s6?charsetutf8, max_overflow0, pool_size5)
Session sessionmaker(bindengine)
session Session()
# 添加session.add_all([Hobby(caption乒乓球),Hobby(caption羽毛球),Person(name张三, hobby_id3),Person(name李四, hobby_id4),
])person Person(name张九, hobbyHobby(caption姑娘))
session.add(person)
#添加二
hb Hobby(caption人妖)
hb.pers [Person(name文飞), Person(name博雅)]
session.add(hb)session.commit()
# 使用relationship正向查询v session.query(Person).first()
print(v.name)
print(v.hobby.caption)
# 使用relationship反向查询v session.query(Hobby).first()
print(v.caption)
print(v.pers)#方式一自己链表
# person_listsession.query(models.Person.name,models.Hobby.caption).join(models.Hobby,isouterTrue).all()
person_listsession.query(models.Person,models.Hobby).join(models.Hobby,isouterTrue).all()
for row in person_list:# print(row.name,row.caption)print(row[0].name,row[1].caption)#方式二通过relationshipperson_listsession.query(models.Person).all()
for row in person_list:print(row.name,row.hobby.caption)
#查询喜欢姑娘的所有人
objsession.query(models.Hobby).filter(models.Hobby.id1).first()
personsobj.pers
print(persons)
session.close()11.多对多
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine create_engine(mysqlpymysql://root:123127.0.0.1:3306/s6?charsetutf8, max_overflow0, pool_size5)
Session sessionmaker(bindengine)
session Session()
# 添加session.add_all([Server(hostnamec1.com),Server(hostnamec2.com),Group(nameA组),Group(nameB组),
])
session.commit()s2g Server2Group(server_id1, group_id1)
session.add(s2g)
session.commit()gp Group(nameC组)
gp.servers [Server(hostnamec3.com),Server(hostnamec4.com)]
session.add(gp)
session.commit()ser Server(hostnamec6.com)
ser.groups [Group(nameF组),Group(nameG组)]
session.add(ser)
session.commit()
# 使用relationship正向查询v session.query(Group).first()
print(v.name)
print(v.servers)
# 使用relationship反向查询v session.query(Server).first()
print(v.hostname)
print(v.groups)
session.close()12.其它
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text, func
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine create_engine(mysqlpymysql://root:123127.0.0.1:3306/s6?charsetutf8, max_overflow0, pool_size5)
Session sessionmaker(bindengine)
session Session()# 关联子查询:correlate(Group)表示跟Group表做关联as_scalar相当于对该sql加括号用于放在后面当子查询
subqry session.query(func.count(Server.id).label(sid)).filter(Server.id Group.id).correlate(Group).as_scalar()
result session.query(Group.name, subqry)SELECT group.name AS group_name, (SELECT count(server.id) AS sid
FROM server
WHERE server.id group.id) AS anon_1
FROM groupselect * from tb where id in [select id from xxx];select id,name,#必须保证此次查询只有一个值(select max(id) from xxx) as mid
from tb例如第三个字段只能有一个值
id name mid
1 lqz 12 不合理
2 egon 2成绩表
id sid cid score
1 1 物理 99
2 1 化学 88
3 2 物理 95学生表
id name 每个学生总分数
1 xx 88
2 yy 77select idname,
(select avr(score) from 成绩表 where 成绩表.sid学生表.id) as x
from 学生表
subqry session.query(func.count(成绩表.scort).label(sc)).filter(学生表.id 成绩表.sid).correlate(学生表).as_scalar()
result session.query(学生表.name, subqry)# 原生SQL# 查询
cursor session.execute(select * from users)
result cursor.fetchall()# 添加
cursor session.execute(insert into users(name) values(:value),params{value:wupeiqi})
session.commit()
print(cursor.lastrowid)
session.close()13.Flask-SQLAlchemy
flask和SQLAchemy的管理者通过他把他们做连接
db SQLAlchemy()- 包含配置- 包含ORM基类- 包含create_all- engine- 创建连接离线脚本创建表
详见代码
flask-migrate python3 manage.py db init 初始化只执行一次
python3 manage.py db migrate 等同于 makemigartions python3 manage.py db upgrade 等同于migrate
