当前位置：Gxlcms > 数据库问题 > Flask SQLAlchemy

Flask SQLAlchemy

时间：2021-07-01 10:21:17 帮助过：13人阅读

目录

  • Flask SQLAlchemy
  • 单表查询
  • 一对多关联
  • 多对多
  • flask_sqlalchemy
  • flask_migrate

Flask SQLAlchemy

models 文件

  1. <code># pip install  sqlalchemy
  2. import datetime
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.ext.declarative import declarative_base
  5. from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
  6. from sqlalchemy.orm import relationship
  7. Base = declarative_base()
  8. class Users(Base):
  9.     __tablename__ = 'users'  # 数据库表名称
  10.     id = Column(Integer, primary_key=True)  # id 主键
  11.     name = Column(String(32), index=True, nullable=False)  # name列，索引，不可为空
  12.     age = Column(Integer)
  13.     #email = Column(String(32), unique=True)
  14.     #datetime.datetime.now#不能加括号，加了括号，以后永远是当前时间
  15.     #ctime = Column(DateTime, default=datetime.datetime.now)
  16.     #extra = Column(Text, nullable=True)
  17.     __table_args__ = (
  18.         # UniqueConstraint('id', 'name', name='uix_id_name'), #联合唯一
  19.         # Index('ix_id_name', 'name', 'email'), #索引
  20.     )
  21.     def __repr__(self):
  22.         return self.name
  23.     
  24. class Hobby(Base):
  25.     __tablename__ = "hobby"
  26.     id  = Column(Integer,primary_key=True)
  27.     catption =Column(String(50),default="双色球")
  28. class Person(Base):
  29.     __tablename__ = "person"
  30.     nid = Column(Integer,primary_key=True)
  31.     name = Column(String(32))
  32.     #hobby值tablename而不是Hobby类名，
  33.     hobby_id = Column(Integer,ForeignKey("hobby.id"))
  34.     # 更数据库没有关系，不会新增加字段，只能用于快速的链表查询操作
  35.     #relationship的第一个参数，是类名，第二个参数backref，用于反向查询
  36.     hobby =relationship("Hobby",backref="pres")
  37.  
  38. # 一个男孩可以喜欢多个女孩，一个女孩也可以喜欢多个男孩
  39. class Boy2Girl(Base):
  40.     __tablename__ = "boy2girl"
  41.     id = Column(Integer, primary_key=True)
  42.     girl_id = Column(Integer,ForeignKey("girl.id"))
  43.     boy_id = Column(Integer,ForeignKey("boy.id"))
  44. class Girl(Base):
  45.     __tablename__ = "girl"
  46.     id  = Column(Integer,primary_key=True)
  47.     name =  Column(String(100),nullable=False)
  48.     def __repr__(self):
  49.         return self.name
  50. class Boy(Base):
  51.     __tablename__ = "boy"
  52.     id = Column(Integer, primary_key=True)
  53.     name = Column(String(100), nullable=False)
  54.     #secondary=boy2girl 中间表的表名
  55.     
  56. def init_db():
  57.     """
  58.     根据类创建数据库表
  59.     :return:
  60.     """
  61.     engine = create_engine(
  62.         "mysql+pymysql://root:@127.0.0.1:3307/python13?charset=utf8",
  63.         #"什么数据库（mysql,orcal）+用什么取链接数据库（pymysql）://数据库用户名:密码@mysqlip:端口/数据库名？charset=字符集"
  64.         max_overflow=0,  # 超过连接池大小外最多创建的连接
  65.         pool_size=5,  # 连接池大小
  66.         pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
  67.         pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
  68.     )
  69.     Base.metadata.create_all(engine)
  70. def drop_db():
  71.     """
  72.     根据类删除数据库表
  73.     :return:
  74.     """
  75.     engine = create_engine(
  76.          "mysql+pymysql://root:@127.0.0.1:3307/python13?charset=utf8",
  77.         max_overflow=0,  # 超过连接池大小外最多创建的连接
  78.         pool_size=5,  # 连接池大小
  79.         pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
  80.         pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
  81.     )
  82.     Base.metadata.drop_all(engine)
  83. if __name__ == '__main__':
  84.     drop_db()
  85.     init_db()</code>

orm.py

  1. <code>from sqlalchemy.orm import sessionmaker
  2. from sqlalchemy import create_engine
  3. from models import Users
  4. #"mysql+pymysql://root@127.0.0.1:3306/aaa"
  5. engine = create_engine("mysql+pymysql://root:@127.0.0.1:3307/python13", max_overflow=0, pool_size=5)
  6. Connection = sessionmaker(bind=engine)
  7. # 每次执行数据库操作时，都需要创建一个Connection
  8. con = Connection()
  9. # 1 单增
  10. # obj1  = Users(name="lsb1",age=12)
  11. # con.add(obj1)
  12. # 2 多个增加
  13. # con.add_all([
  14. #          Users(name="lsb1",age=12),
  15. #         Users(name="esb",age=40),
  16. #         Users(name="jsb",age=30),
  17. #         Users(name="tsb",age=12),
  18. #         #Host(name = "tsb",time=123213)
  19. # ])
  20. # 3 删除
  21. # con.query(Users).delete()
  22. #4 改
  23. # con.query(Users).update({"name":"sb","age":14})
  24. # con.query(Users).update({Users.name:Users.name +" is true","age":1},synchronize_session=False)
  25. # con.query(Users).update({Users.age:Users.age + 10})
  26. # 5查(查是不需要commit,也能拿到结果)
  27. #打印sql
  28. # r1 = con.query(Users)
  29. #查询所有
  30. # r1 = con.query(Users).all()
  31. #
  32. #查单条记录
  33. # r1  = con.query(Users).first()
  34. #查哪些字段
  35. # r1  = con.query(Users.age,Users.name.label("sb")).first()
  36. #过滤用filter_by(传参数)或者filter（传表达式）
  37. # r1 = con.query(Users).filter(Users.name == "tsb").first()
  38. # con.query(Users).filter(Users.name == "tsb").update({"name": "sb", "age": 14})
  39. r1 =  con.query(Users).filter_by(name = "esb").first()
  40. print(r1)
  41. #必须提交才能生效
  42. con.commit()
  43. #关闭链接
  44. con.close()</code>



单表查询

  1. <code>from sqlalchemy.orm import sessionmaker
  2. from sqlalchemy import create_engine
  3. from models import Users
  4. #"mysql+pymysql://root@127.0.0.1:3306/aaa"
  5. engine = create_engine("mysql+pymysql://root:@127.0.0.1:3307/python13", max_overflow=0, pool_size=5)
  6. Connection = sessionmaker(bind=engine)
  7. # 每次执行数据库操作时，都需要创建一个Connection
  8. session = Connection()
  9. # 条件
  10. # ret =  session.query(Users).filter_by(name = "esb").all()
  11. #表达式，and 条件链接
  12. # ret  = session.query(Users).filter(Users.name == "sb",Users.age ==14 ).first()
  13. # print(ret.age,ret.name)
  14. # 表示的between，条件,30<=age<=40
  15. # ret  = session.query(Users).filter(Users.age.between(30,40)).all()
  16. # print(ret)
  17. # sql查询的in_操作,相当于django中的__in
  18. # ret =session.query(Users).filter(Users.id.in_([9,11,13])).all()
  19. # print(ret)
  20. # # sql查询取反
  21. # ret1 = session.query(Users).filter(~Users.id.in_([9,11,13])).all()
  22. # print(ret1)
  23. #or查询 ，or和and ,做整合
  24. from sqlalchemy import or_,and_
  25. # ret =  session.query(Users).filter(or_(Users.id == 9,Users.name=="jsb")).all()
  26. # ret =  session.query(Users).filter(and_(Users.id == 9,Users.name=="lsb1")).all()
  27. # ret =  session.query(Users).filter(or_(
  28. #     Users.id == 9,
  29. #     and_(Users.name=="jsb",Users.id==13),
  30. #
  31. #     )
  32. # ).all()
  33. # like查询,
  34. #必须以b开头
  35. # ret = session.query(Users).filter(Users.name.like("b%")).all()
  36. # #第二字母是b
  37. # ret = session.query(Users).filter(Users.name.like("_b%")).all()
  38. #不以b开头
  39. # ret = session.query(Users).filter(~Users.name.like("b%")).all()
  40. #排序
  41. #desc重大到小排序
  42. # ret = session.query(Users).filter(Users.id>1).order_by(Users.id.desc()).all()
  43. #desc重小到大排序
  44. #ret = session.query(Users).filter(Users.id>1).order_by(Users.id.asc()).all()
  45. #多条件排序,先以年纪从大到小排，如果年龄相同，再以id从小到大排
  46. # ret = session.query(Users).filter(Users.id>1).order_by(Users.age.desc(),Users.id.asc()).all()
  47. # print(ret)
  48. #分组查询
  49. # ret  = session.query(Users).group_by(Users.name).all()
  50. # 再分组的时候如果要用聚合操作，就要导入func
  51. from sqlalchemy.sql import func
  52. #选出组内最小年龄要大于等于30的组
  53. # ret  = session.query(Users).group_by(Users.name).having(func.min(Users.age)>=30).all()
  54. #选出组内最小年龄要大于等于30的组,查询组内的最小年龄，最大年纪，年纪之和，
  55. ret = session.query(
  56.     func.min(Users.age),
  57.     func.max(Users.age),
  58.     func.sum(Users.age),
  59.     Users.name
  60. ).group_by(Users.name).having(func.min(Users.age)>=30).all()
  61. print(ret)</code>

一对多关联

  1. <code>from sqlalchemy.orm import sessionmaker
  2. from sqlalchemy import create_engine
  3. from models import Hobby,Person
  4. #"mysql+pymysql://root@127.0.0.1:3306/aaa"
  5. engine = create_engine("mysql+pymysql://root:@127.0.0.1:3307/python13", max_overflow=0, pool_size=5)
  6. Connection = sessionmaker(bind=engine)
  7. # 每次执行数据库操作时，都需要创建一个Connection
  8. session = Connection()
  9. #1添加,没有用关联关系
  10. # session.add_all([
  11. #     Hobby(catption="淫诗"),
  12. #     Hobby(catption="推背"),
  13. #     Person(name="tank",hobby_id=1),
  14. #     Person(name="jason",hobby_id=2)
  15. # ])
  16. # 2添加 用关联关系
  17. # preson = Person(name="egon",hobby=Hobby(catption="相亲"))
  18. #session.add(preson)
  19. #
  20. # hobb = Hobby(catption="人妖")
  21. # hobb.pres = [Person(name="owen"),Person(name="sean")]
  22. # session.add(hobb)
  23. #session.commit()
  24. #正向查询
  25. # pr = session.query(Person).filter( Person.name == "tank").first()
  26. # print(pr.name)
  27. # print(pr.hobby.catption)
  28. #反向查
  29. # v = session.query(Hobby).filter(Hobby.catption=="人妖").first()
  30. # print(v.catption)
  31. # print(v.pres)
  32. # 自己连表,isouter=True表示是left join，不填默认为inner join
  33. person_list =  session.query( Hobby).join(Person,Person.hobby_id==Hobby.id,isouter=True)
  34. #
  35. print(person_list)
  36. session.close()
  37. </code>

多对多

  1. <code>from sqlalchemy.orm import sessionmaker
  2. from sqlalchemy import create_engine
  3. from models import Boy,Boy2Girl,Girl
  4. #"mysql+pymysql://root@127.0.0.1:3306/aaa"
  5. engine = create_engine("mysql+pymysql://root:@127.0.0.1:3307/python13", max_overflow=0, pool_size=5)
  6. Connection = sessionmaker(bind=engine)
  7. # 每次执行数据库操作时，都需要创建一个Connection
  8. session = Connection()
  9. #添加
  10. # session.add_all([
  11. #     Boy(name="tank"),
  12. #     Boy(name="sean"),
  13. #     Girl(name="仓老师"),
  14. #     Girl(name="小泽老师")
  15. # ])
  16. # b2g = Boy2Girl(boy_id=1,girl_id=2)
  17. # session.add(b2g )
  18. # b2g = Boy2Girl(boy_id=2,girl_id=1)
  19. # session.add(b2g )
  20. # session.commit()
  21. # session.close()
  22. #
  23. # boy = Boy(name="亚峰")
  24. # boy.girl=[Girl(name="迪丽热巴"),Girl(name="三上")]
  25. # session.add(boy)
  26. # session.commit()
  27. #
  28. # girl = Girl(name="丹丹")
  29. # girl.boys=[Boy(name="吴彦祖"),Boy(name="鹿晗")]
  30. # session.add(girl)
  31. # session.commit()
  32. # 使用relationship的关系，正向查
  33. # b = session.query(Boy).filter(Boy.name == "亚峰").first()
  34. # print(b.name)
  35. # print(b.girl)
  36. #反向查询
  37. # g = session.query(Girl).filter(Girl.name=="丹丹").first()
  38. # print(g.name)
  39. # print(g.boys)</code>

flask_sqlalchemy

  1. <code>要用就必须先安装。
  2. 所有的到导入都找 下面的db
  3. from flask_sqlalchemy import SQLAlchemy
  4. db = SQLAlchemy()</code>

flask_migrate

  1. <code>命令：manager.add_command('db1', MigrateCommand)
  2. 1 当项目第一次执行迁移的时候。
  3. python3 manage.py db1 init 只需要初始化一次
  4. 2 python3 manage.py db1 migrate # 等同于django的makemigrations
  5. 3 python3 manage.py db1 upgrade # 等同于django的migrate</code>

Flask SQLAlchemy

标签：删除数据库   and   报错   filter   str   记录   pytho   过滤   添加   

人气教程排行

本站所有资源全部来源于网络，若本站发布的内容侵害到您的隐私或者利益，请联系我们删除！