flask之六 sqlachemy详解 scoped_session线程安全 基本增删改查 多对

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测试数据mit()
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测试数据mit()
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测试数据mit()
# 关闭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测试数据mit()
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测试数据mit()
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测试数据  "  ))
db.session测试数据mit()

users  = User.query.all()

 

flask之六 sqlachemy详解 scoped_session线程安全 基本增删改查 多对多关系建立和操作 flask-sqlalchemy的使用

标签：结构   style   example   metadata   大小   名称   ams   合并   bin   

查看更多关于flask之六 sqlachemy详解 scoped_session线程安全 基本增删改查 多对的详细内容...