SQLAlchemy 增删改查 一对一 多对多

首先要导入SQLAIchemy模块

from sqlalchemy.ect.declaative import  declarative_base

创建orm基类

Base = declarative_base()

创建orm对象

class User(Base):

    __tablename__ = "user"

    id = Column(Integer,primary_key=True,autoincrement=True)

    name = Column(String(),index=True)

创建数据库链接

from sqlalchemy import create_engine

ngine = create_engine("mysql+pymysql://root:[email protected]:3306/user?charset=utf8")

备注：pwd表示密码，如果数据库属性中没有密码的话可以忽略，user是数据库名

# 数据库链接创建完成
# 去数据库中创建与User所对应的数据库表
# 去engine数据库中创建继承Base类所对应的数据表

Base.metadata.create_all(engine)

#############################################################################################################################

添加数据

首先打开数据库链接

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:[email protected]:3306/user?charset=utf8")

创建会话窗口---打开数据库链接

from sqlalchemy.orm import  sessionmaker

创建会话

Session = sessionmaker(engine)

打开会话窗口

db_session = Session()

添加单条数据

user_obj=  User(name="ywb")

db_session.add(user_obj)

添加多条数据

db_session.add_all([

    User(name="Alex"),

    User(name="Wusir"),

    User(name="Yuan")

])

开始执行操作

   db_session.commit()

执行完成之后关闭执行命令

db_session.close()

查询数据

 # 创建会话窗口

 from sqlalchemy.orm import sessionmaker

 from creat_table import engine,User

 Session = sessionmaker(engine)

 db_session = Session()

 # 简单的查询语句

 user_list = db_session.query(User).all()

 for row in user_list:

     print(row.id,row.name)

 # 单条查询

 user = db_session.query(User).first()

 print(user.id,user.name)

 # 带条件的查询

 user_list = db_session.query(User).filter(User.id==).all()

 print(user_list[].id,user_list[].name)

 user = db_session.query(User).filter_by(id=).first()

 print(user.id,user.name)

 user_list = db_session.query(User).filter(User.id>=).all()

 for row in user_list:

     print(row.id,row.name)

 # 查询sql语句(拓展)

 sql = db_session.query(User).filter(User.id>=)

 print(sql)

查询数据

修改更新数据

 #修改数据

 # 建立会话

 from sqlalchemy.orm import sessionmaker

 # 导入链接数据库的模块

 from creat_table import engine,User

 # 开始创建会话

 Session = sessionmaker(engine)

 # 开启会话

 db_session = Session()

 # 开始修改数据

 res = db_session.query(User).filter(User.name=="ybw").update({"name":"哈哈哈"})

 print(res)

 db_session.commit()

 db_session.close()

修改更新数据

删除数据

 #删除数据

 from sqlalchemy.orm import sessionmaker

 from creat_table import engine,User

 Session = sessionmaker(engine)

 db_session = Session()

 # 删除

 res = db_session.query(User).filter(User.id == ).delete()

 print(res)

 db_session.commit()

 db_session.close()

删除数据

建立一对多：的ForeignKey关系实现的增删改查

创建数据库

 #一对多操作

 from sqlalchemy.ext.declarative import declarative_base

 Base = declarative_base()

 from sqlalchemy import Column,String,Integer,ForeignKey

 from sqlalchemy.orm import relationship

 class Student(Base):

     __tablename__ = "student"

     id = Column(Integer,primary_key=True)

     name = Column(String())

     school_id = Column(Integer,ForeignKey("school.id"))

     stu2sch = relationship("School",backref = "sch2stu")

 class School(Base):

     __tablename__ = "school"

     id = Column(Integer,primary_key=True)

     name = Column(String())

 from sqlalchemy import create_engine

 engine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/day127?charset=utf8")

 Base.metadata.create_all(engine)

创建数据库

添加数据

 #   添加数据

 from sqlalchemy.orm import sessionmaker

 from creat_table_ForeignKey import engine,Student,School

 Session = sessionmaker(engine)

 db_session = Session()

 ## 添加数据   原始方法

 sch_obj = School(name="OldBoyBeijing")

 db_session.add(sch_obj)

 db_session.commit()

 sch = db_session.query(School).filter(School.name =="OldBoyBeijing").first()

 stu_obj = Student(name="Yuan",school_id=sch.id)

 db_session.add(stu_obj)

 db_session.commit()

 db_session.close()

 # 添加数据  反向添加 relationship

 sch_obj = School(name="OldBoyShanghai")

 sch_obj.sch2stu = [Student(name="江疏影"),Student(name="周冬雨"),]

 db_session.add(sch_obj)

 db_session.commit()

 db_session.close()

 # 添加数据 正向添加   relationship

 stu_obj = Student(name="大黄鸭",stu2sch = School(name="OldBoyShenzhen"))

 db_session.add(stu_obj)

 db_session.commit()

 db_session.close()

添加数据

查询数据

 #   查询数据

 from sqlalchemy.orm import sessionmaker

 from creat_table_ForeignKey import engine,Student,School

 Session = sessionmaker(engine)

 db_session = Session()

 # #.查询数据  正向查询数据  relationship

 stu = db_session.query(Student).all()

 for row in stu:

     print(row.id,row.name,row.school_id,row.stu2sch.name)

 # #.查询数据  反向查询数据  relationship

 sch = db_session.query(School).all()

 for school in sch:

     for student in school.sch2stu:

         print(school.id,school.name,student.name)

查询数据

修改数据

 #   更新数据

 from sqlalchemy.orm import sessionmaker

 from creat_table_ForeignKey import engine,Student,School

 Session = sessionmaker(engine)

 db_session = Session()

 # 修改数据

 sch = db_session.query(School).filter(School.name=="OldBoyShanghai").first()

 db_session.query(Student).filter(Student.name == "大黄鸭").update({"school_id":sch.id})

 db_session.commit()

 db_session.close()

修改数据

删除数据

 #   删除数据

 from sqlalchemy.orm import sessionmaker

 from creat_table_ForeignKey import engine,Student,School

 Session = sessionmaker(engine)

 db_session = Session()

 # 删除数据

 sch = db_session.query(School).filter(School.name=="OldBoyShanghai").first()

 db_session.query(Student).filter(Student.school_id == sch.id).delete()

 db_session.commit()

 db_session.close()

删除数据

高级版的查询操作

 # 高级版查询操作,厉害了哦

 #老规矩

 from my_create_table import User,engine

 from sqlalchemy.orm import sessionmaker

 Session = sessionmaker(engine)

 db_session = Session()

 # 查询数据表操作

 # and or

 from sqlalchemy.sql import and_ , or_

 ret = db_session.query(User).filter(and_(User.id > , User.name == 'DragonFire')).all()

 ret = db_session.query(User).filter(or_(User.id < , User.name == 'DragonFire')).all()

 # 查询所有数据

 r1 = db_session.query(User).all()

 # 查询数据 指定查询数据列 加入别名

 r2 = db_session.query(User.name.label('username'), User.id).first()

 print(r2.id,r2.username) #  NBDragon

 # 表达式筛选条件

 r3 = db_session.query(User).filter(User.name == "DragonFire").all()

 # 原生SQL筛选条件

 r4 = db_session.query(User).filter_by(name='DragonFire').all()

 r5 = db_session.query(User).filter_by(name='DragonFire').first()

 # 字符串匹配方式筛选条件 并使用 order_by进行排序

 r6 = db_session.query(User).filter(text("id<:value and name=:name")).params(value=, name='DragonFire').order_by(User.id).all()

 #原生SQL查询

 r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='DragonFire').all()

 # 筛选查询列

 # query的时候我们不在使用User ORM对象,而是使用User.name来对内容进行选取

 user_list = db_session.query(User.name).all()

 print(user_list)

 for row in user_list:

     print(row.name)

 # 别名映射  name as nick

 user_list = db_session.query(User.name.label("nick")).all()

 print(user_list)

 for row in user_list:

     print(row.nick) # 这里要写别名了

 # 筛选条件格式

 user_list = db_session.query(User).filter(User.name == "DragonFire").all()

 user_list = db_session.query(User).filter(User.name == "DragonFire").first()

 user_list = db_session.query(User).filter_by(name="DragonFire").first()

 for row in user_list:

     print(row.nick)

 # 复杂查询

 from sqlalchemy.sql import text

 user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=,name="DragonFire")

 # 查询语句

 from sqlalchemy.sql import text

 user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=,name="DragonFire")

 # 排序 :

 user_list = db_session.query(User).order_by(User.id).all()

 user_list = db_session.query(User).order_by(User.id.desc()).all()

 for row in user_list:

     print(row.name,row.id)

 #其他查询条件

 """

 ret = session.query(User).filter_by(name='DragonFire').all()

 ret = session.query(User).filter(User.id > , User.name == 'DragonFire').all()

 ret = session.query(User).filter(User.id.between(, ), User.name == 'DragonFire').all() # between 大于1小于3的

 ret = session.query(User).filter(User.id.in_([,,])).all() # in_([,,]) 只查询id等于1,,4的

 ret = session.query(User).filter(~User.id.in_([,,])).all() # ~xxxx.in_([,,]) 查询不等于1,,4的

 ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='DragonFire'))).all() 子查询

 from sqlalchemy import and_, or_

 ret = session.query(User).filter(and_(User.id > , User.name == 'DragonFire')).all()

 ret = session.query(User).filter(or_(User.id < , User.name == 'DragonFire')).all()

 ret = session.query(User).filter(

     or_(

         User.id < ,

         and_(User.name == 'eric', User.id > ),

         User.extra != ""

     )).all()

 # select * from User where id< or (name="eric" and id>) or extra != ""

 # 通配符

 ret = db_session.query(User).filter(User.name.like('e%')).all()

 ret = db_session.query(User).filter(~User.name.like('e%')).all()

 # 限制

 ret = db_session.query(User)[:]

 # 排序

 ret = db_session.query(User).order_by(User.name.desc()).all()

 ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all()

 # 分组

 from sqlalchemy.sql import func

 ret = db_session.query(User).group_by(User.extra).all()

 ret = db_session.query(

     func.max(User.id),

     func.sum(User.id),

     func.min(User.id)).group_by(User.name).all()

 ret = db_session.query(

     func.max(User.id),

     func.sum(User.id),

     func.min(User.id)).group_by(User.name).having(func.min(User.id) >).all()

 """

 # 关闭连接

 db_session.close()

 orm_select_more

高级版查询操作

高级修改数据操作

 #高级版更新操作

 from my_create_table import User,engine

 from sqlalchemy.orm import sessionmaker

 Session = sessionmaker(engine)

 db_session = Session()

 #直接修改

 db_session.query(User).filter(User.id > ).update({"name" : ""})

 #在原有值基础上添加 -

 db_session.query(User).filter(User.id > ).update({User.name: User.name + ""}, synchronize_session=False)

 #在原有值基础上添加 -

 db_session.query(User).filter(User.id > ).update({"age": User.age + }, synchronize_session="evaluate")

 db_session.commit()

 orm_update_more

高级修改操作

建立多对多：的ManyToMany

创建表关系：

 from sqlalchemy.ext.declarative import declarative_base

 Base = declarative_base()

 from sqlalchemy import Column,Integer,String,ForeignKey

 from sqlalchemy.orm import relationship

 class Hotel(Base):

     __tablename__="hotel"

     id=Column(Integer,primary_key=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(),index=True)

     #创建关系

     boys = relationship("Boy",secondary="hotel",backref="girl2boy")

 class Boy(Base):

     __tablename__="boy"

     id=Column(Integer,primary_key=True)

     name = Column(String(),index=True)

 from sqlalchemy import create_engine

 engine = create_engine("mysql+pymysql://root:[email protected]:3306/dragon?charset=utf8")

 Base.metadata.create_all(engine)

 my_M2M.py

 my_M2M.py

创建表关系

基于relationship增加数据：

 from my_M2M import Girl,Boy,Hotel,engine

 # 创建连接

 from sqlalchemy.orm import sessionmaker

 # 创建数据表操作对象 sessionmaker

 DB_session = sessionmaker(engine)

 db_session = DB_session()

 # .通过Boy添加Girl和Hotel数据

 boy = Boy(name="DragonFire")

 boy.girl2boy = [Girl(name="赵丽颖"),Girl(name="Angelababy")]

 db_session.add(boy)

 db_session.commit()

 # .通过Girl添加Boy和Hotel数据

 girl = Girl(name="珊珊")

 girl.boys = [Boy(name="Dragon")]

 db_session.add(girl)

 db_session.commit()

 orm_M2M_insert.py

 orm_M2M_insert.py

基于relationship增加数据

基于relationship查询数据：

 from my_M2M import Girl,Boy,Hotel,engine

 # 创建连接

 from sqlalchemy.orm import sessionmaker

 # 创建数据表操作对象 sessionmaker

 DB_session = sessionmaker(engine)

 db_session = DB_session()

 # .通过Boy查询约会过的所有Girl

 hotel = db_session.query(Boy).all()

 for row in hotel:

     for row2 in row.girl2boy:

         print(row.name,row2.name)

 # .通过Girl查询约会过的所有Boy

 hotel = db_session.query(Girl).all()

 for row in hotel:

     for row2 in row.boys:

         print(row.name,row2.name)

 orm_M2M_select.py

 orm_M2M_select.py

基于relationship查询数据