标签：结构 style example metadata 大小 名称 ams 合并 bin
一、sqlachemy(是一个orm框架)
1 orm框，可以独立出来用 2 orm执行原生sql 生成engine(链接池) 详见上一篇博客 engine = create_engine() 获取链接 conn = engine.raw_connection() 后续就一样了
3 创建表，删除表和生成字段（不能创建数据库）
-写一个类
Base = declarative_base()
class Users(Base):
__tablename__ = ‘users‘ # 数据库表名称
id = Column(Integer, primary_key=True) # id 主键
name = Column(String(32), index=True, nullable=False) # name列，索引，不可为空
-把被Base管理的所有表创建和删除
engine = create_engine()
Base.metadata.create_all(engine)
Base.metadata.drop_all(engine)
4 单表新增
engine = create_engine()
Connection = sessionmaker(bind=engine)
# 每次执行数据库操作时，都需要创建一个Connection
conn = Connection()
user=User(name=‘lqz‘)
conn.add(user)
conn.add_all([对象1，对象2])
# 提交
conn.commit()
con.close()
5 一对多关系建立
class Hobby(Base):
__tablename__ = ‘hobby‘
id = Column(Integer, primary_key=True)
caption = Column(String(50), default=‘篮球‘)
class Person(Base):
__tablename__ = ‘person‘
nid = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
# hobby指的是tablename而不是类名，uselist=False
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关，不会新增字段，只用于快速链表操作
# 类名，backref用于反向查询
hobby=relationship(‘Hobby‘,backref=‘pers‘)
5 一对多新增
engine = create_engine()
Connection = sessionmaker(bind=engine)
# 每次执行数据库操作时，都需要创建一个Connection
conn = Connection()
hobby=Hobby(caption=‘美女‘)
person=Person(name=‘lqz‘,hobby_id=1)
conn.add_all([hobby，person])
# 第二种方式
hobby=Hobby(caption=‘美女‘)
person=Person(name=‘lqz‘,hobby=hobby) # 通过对象来匹配hobby_id
conn.add_all([hobby，person])
# 提交
conn.commit()
con.close()
二、scoped_session线程安全
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session from models import Users engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) # 从连接池中拿一个链接 conn = sessionmaker(bind=engine) session = scoped_session(conn) # @app.route(‘/‘) # def index(): # # session.add() obj1 = Users(name="lqz") session.add(obj1) """ # 线程安全，基于本地线程实现每个线程用同一个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‘ ) """ # 提交事务 session.commit() # 关闭session session.close()
三、基本增删改查
# 基本增删查改 from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session from models import Users,Person,Hobby from sqlalchemy.sql import text engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) # 从连接池中拿一个链接 conn = sessionmaker(bind=engine) session = scoped_session(conn) # 增 add ，add_all # 查询 # # 查询Users表中id为1的第一条记录 # filter_by写等式，filter写比较 # res=session.query(Users).filter_by(id=1).first() # res=session.query(Users).filter(Users.id<2) # for user in res: # print(user.name) # 影响的条数 # res=session.query(Users).filter_by(id=1).delete() # res=session.query(Users).filter(Users.id>1).delete() # print(res) ### 修改 # res是影响的行数 # res=session.query(Users).filter(Users.id > 0).update({"name" : "lqz"}) #类似于django的F查询 # synchronize_session=False 表示加字符串 # res=session.query(Users).filter(Users.id > 0).update({‘name‘: Users.name + "099"}, synchronize_session=False) # res=session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate") # res=session.query(Users).filter(Users.id > 0).update({Users.age: Users.age + 1}, synchronize_session="evaluate") # print(res) #查询 # res=session.query(Users).all() # 查所有 # res=session.query(Users).all()[0] #取第一个 # res=session.query(Users).first() ##取第一个 # select name as xx,age form User; # r2 = session.query(Users.name.label(‘xx‘), Users.age).all() # # r2 = session.query(Users.name.label(‘xx‘), Users.age) # # # print(r2) # for user in r2: # print(user.xx) #filter传的是表达式，filter_by传的是参数 # r3 = session.query(Users).filter(Users.name == "lqz099").all() # r3 = session.query(Users).filter(Users.id == 1).all() # r4 = session.query(Users).filter_by(name=‘lqz‘).all() # r5 = session.query(Users).filter_by(name=‘lqz‘).first() # print(r3) #:value 和:name 相当于占位符，用params传参数 # select * from user where id<6 and name=lqz099 order by id; # r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=6, name=‘lqz099‘).order_by(Users.id).all() #自定义查询sql # r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name=‘lqz099‘).all() # # for u in r7: # print(u.id) ### 其它查询操作 # 条件 # ret = session.query(Users).filter_by(name=‘lqz‘).all() #表达式，and条件连接 # ret = session.query(Users).filter(Users.id > 1, Users.name == ‘lqz099‘).all() # ret = session.query(Users).filter(Users.id.between(5, 6), Users.name == ‘lqz099‘).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(name=‘lqz099‘))) 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(Users.id > 3, Users.name == ‘eric‘).all() # ret = session.query(Users).filter(or_(Users.id < 5, Users.name == ‘lqz099‘)).all() # ret = session.query(Users).filter( # or_( # Users.id < 2, # and_(Users.name == ‘eric‘, Users.id > 3), # Users.age != 8 # )) # print(ret) # 通配符，以e开头，不以e开头 # ret = session.query(Users).filter(Users.name.like(‘%9%‘)).all() # ret = session.query(Users).filter(~Users.name.like(‘%9%‘)) # 限制，用于分页，区间 # select *from user limit(1:4) # select * from users limit 1,4; 从第一条数据往后取4条 # ret = session.query(Users)[0:4] # ret = session.query(Users)[2:6] # print(ret[0].id) # 排序，根据name降序排列（从大到小） # ret = session.query(Users).order_by(Users.name.desc()).all() #第一个条件重复后，再按第二个条件升序排 # ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()) # print(ret) ### 分组 # 分组 from sqlalchemy.sql import func #分组之后取最大id，id之和，最小id # select max(id),sum(id),min(id) from user group by name; # ret = session.query( # func.max(Users.id), # func.sum(Users.id), # func.min(Users.id)).group_by(Users.name).all() #haviing筛选 # select max(id),sum(id),min(id) from user group by name having min(id)>2; # Users.objects.value(Users.name).filter().annotate(a=max(User.id),b=min(User.id)).filter(b__gt=2) # 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关联） # select * from person,hobby where person.hobby_id=hobby.id; # ret = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all() # # print(ret)[(person,hobby),(person,hobby)] # # # for p in ret: # print(p[0]) # print(p[1].caption) # print(‘-----‘) #join表，默认是inner join,没有指定on的字段，默认用外键关联 # select * from Person inner join Hobby on person.hobby_id =hobby.id; # ret = session.query(Person).join(Hobby) # print(ret) # #isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可 # ret = session.query(Person).join(Hobby, isouter=True) ret = session.query(Hobby).join(Person, isouter=True) # print(ret) # 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上 # select * from Person left join Hobby on person.id=hobby.id # ret = session.query(Person).join(Hobby,Person.hobby_id==Hobby.id, isouter=True) # print(ret) # select * from Person inner join Hobby on Person.hobby_id=Hobby.id where person.id>1 and hobby.caption=‘篮球‘ # ret = session.query(Person).join(Hobby).filter(Person.id>1,Hobby.caption==‘篮球‘) # # 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集 # #union和union all的区别？ q1 = session.query(Person.name).filter(Person.nid > 2) # q1 = session.query(Person.name).filter(Person.nid > 2).all() # q2 = session.query(Hobby.caption).filter(Hobby.id < 2) # ret = q1.union(q2).all() # print(ret) q1 = session.query(Person.name).filter(Person.nid > 2) q2 = session.query(Hobby.caption).filter(Hobby.id < 2) ret = q1.union_all(q2).all() print(ret) session.commit() session.close()
四、多对多关系的建立
import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index 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) class Boy(Base): __tablename__ = ‘boy‘ id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(64), unique=True, nullable=False) # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以 girls = relationship(‘Girl‘, secondary=‘boy2girl‘, backref=‘boys‘) # engine = create_engine( # "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", # max_overflow=0, # 超过连接池大小外最多创建的连接 # pool_size=5, # 连接池大小 # pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 # pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） # ) # Base.metadata.create_all(engine)
4.1 基本增删改查
# 基本增删查改 from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session from models import Boy, Girl, Boy2Girl from sqlalchemy.sql import text engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) # 从连接池中拿一个链接 conn = sessionmaker(bind=engine) session = scoped_session(conn) # boy = Boy(name=‘lqz‘) # boy2 = Boy(name=‘egon‘) # # girl = Girl(name=‘刘亦菲‘) # girl2 = Girl(name=‘迪丽热巴‘) # session.add_all([boy,boy2,girl,girl2]) # 建立关系 # b=Boy2Girl(girl_id=1,boy_id=1) # b=Boy2Girl(girl=对象,boy=对象) # session.add(b) # lqz=session.query(Boy).filter(Boy.id==1).first() # # dlrb=session.query(Girl).filter(Girl.id==2).first() # lyf=session.query(Girl).filter(Girl.id==1).first() # # # # lqz.girls=[dlrb,] # lqz.girls.append(lyf) # egon=session.query(Boy).filter(Boy.id==2).first() # lyf=session.query(Girl).filter(Girl.id==1).first() # lyf.boys.append(egon) # session.add(lyf) # lqz=session.query(Boy).filter(Boy.id==1).first() # lyf=session.query(Girl).filter(Girl.id==1).first() # print(lyf.boys) session.commit() session.close()
五、flask-sqlalchemy的使用
from flask import Flask from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///example.sqlite" db = SQLAlchemy(app) class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String, unique=True, nullable=False) email = db.Column(db.String, unique=True, nullable=False) db.session.add(User(username="Flask", email="example@example.com")) db.session.commit() users = User.query.all()
flask之六 sqlachemy详解 scoped_session线程安全 基本增删改查 多对多关系建立和操作 flask-sqlalchemy的使用
标签：结构 style example metadata 大小 名称 ams 合并 bin
原文地址：https://www.cnblogs.com/ltyc/p/14449125.html
