一、ORM 与 SQLAlchemy 简介
ORM 全称 Object Relational Mapping, 翻译过来叫对象关系映射。简单的说，ORM 将数据库中的表与面向对象语言中的类建立了一种对应关系。这样，我们要操作数据库，数据库中的表或者表中的一条记录就可以直接通过操作类或者类实例来完成。
SQLAlchemy 是Python 社区最知名的 ORM 工具之一，为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。
安装 SQLAlchemy：
pip3 install sqlalchemy
二、创建数据表和数据库连接
create_table.py
#导入 from sqlalchemy import Column,Integer, String, create_engine from sqlalchemy.orm import sessionmaker #创建操纵数据库的会话窗口 from sqlalchemy.ext.declarative import declarative_bas #创建对象的基类 Base = declarative_base() # 定义User类: class User(Base): # 表的名字: __tablename__ = 'user' # 表的结构: id = Column(Integer, primary_key=True) name = Column(String(32),index=True) #初始化数据库连接： engine = create_engine("mysql+pymysql://root:口令@127.0.0.1:3306/数据库名?charset=utf8") #创建会话对象 Session = sessionmaker(engine)#自动检索所有继承Base的ORM 对象 并且创建所有的数据表 Base.metadata.create_all(engine)
三、单表的增删改查操作
增加数据
#mysql操作 insert into user(name) values ("fengchong") #sqlalchemy操作 #1.导入数据库连接，表，会话对象 from create_table import engine,User,Session #2.打开会话对象 db_session = Session() #3.添加一条数据 db_session.add( User(name="fengchong") ) #执行会话窗口中的所有操作 db_session.commit() #4.添加多条数据 db_session.add_all( [ User(name="zhaomingyang"), User(name="haha"), User(name="xiaoxiao") ] ) db_session.commit() #5.关闭会话对象 db_session.close()
查询数据
#mysql操作： select * from user where 条件 #sqlalchemy操作： #1.导入 from create_table import User,engine,Session #2.打开会话对象 db_session = Session() #3.简单查询 user_list = db_session.query(User).all() for user in user_list: print(user.id,user.name) user = db_session.query(User).first() #4.带条件的查询 user_list = db_session.query(User).filter(User.id == 3).all() #取出符合条件的全部结果，结果为一个列表，里面存的是一个个的对象 print(user_list) user = db_session.query(User).filter_by(id=4).first() print(user) #取出符合条件的第一个，user为符合条件的对象 user_list = db_session.query(User).filter(User.id >= 2).all() print(user_list)
修改数据
mysql操作： UPDATE user SET name="fengchong123" WHERE name=fengchong sqlalchemy操作： #1.导入 from create_table import User,engine,Session #2.打开会话对象 db_session = Session() #3.先查询在修改 res = db_session.query(User).filter(User.name == "fengchong").update({"name":"fengchong123"}) #4.提交 db_session.commit() #5.关闭会话 db_session.close()
删除数据
#mysql操作： delete from user where id=5 #sqlalchemy操作： #1.导入 from create_table import User,engine,Session #2.打开会话对象 db_session = Session() #3.先查询再删除 res = db_session.query(User).filter(User.id==5).delete() #4.提交 db_session.commit() #5.关闭会话 db_session.close()
高级版查询操作
# 查询数据表操作 # and or from sqlalchemy.sql import and_ , or_ ret = db_session.query(User).filter(and_(User.id == 3, User.name == 'fengchong')).all() #and print(ret) ret = db_session.query(User).filter(or_(User.id == 4, User.name == 'fengchong123')).all() #or print(ret) ret = db_session.query(User).filter_by(id=4,name="fengchong123").all() #and print(ret) # 查询所有数据 r1 = db_session.query(User).all() # 查询数据 指定查询数据列 加入别名 r2 = db_session.query(User.id.label("uid"),User.name.label("username")).first() print(r2.uid,r2.username) # 表达式筛选条件 r3 = db_session.query(User).filter(User.name == "fengchong").all() # 原生SQL筛选条件 r4 = db_session.query(User).filter_by(name='fengchong').all() r5 = db_session.query(User).filter_by(name='fengchong').first() # 字符串匹配方式筛选条件 并使用 order_by进行排序 from sqlalchemy.sql import text r6 = db_session.query(User).filter(text("id=:value and name=:name")).params(value=3, name='fengchong').first() print(r6.name) # 原生SQL查询 r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='fengchong').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 == "fengchong").all() user_list = db_session.query(User).filter(User.name == "fengchong").first() user_list = db_session.query(User).filter_by(name="fengchong").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=3,name="fengchong") # 查询语句 from sqlalchemy.sql import text user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="fengchong") # 排序 : user_list = db_session.query(User).order_by(User.id.asc()).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(User.id.between(1, 3), User.name == 'fengchong').all() # between 大于1小于3的 ret = session.query(User).filter(User.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的 ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的 ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='fengchong'))).all() 子查询 # 通配符 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)[1:2] # 排序 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( 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) >2).all() """
高级版修改操作
#直接修改 # db_session.query(User).filter(User.id > 0).update({"name" : "099"}) #在原有值基础上添加 - 1 db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False) db_session.commit() #在原有值基础上添加 - 2 # db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate") # db_session.commit()
四、一对多表的增删改查
创建数据表及关系relationship
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import relationship class Student(Base): __tablename__ = "student" id = Column(Integer,primary_key=True) name = Column(String(32)) school_id = Column(Integer,ForeignKey("school.id")) stu2sch = relationship("School",backref="sch2stu") #backref用于反向查询 class School(Base): __tablename__= "school" id = Column(Integer,primary_key=True) name = Column(String(32)) from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:口令@127.0.0.1:3306/数据库名?charset=utf8") Base.metadata.create_all(engine)
基于relationship增加数据
from create_table_ForeignKey import engine,Student,School #创建会话窗口 from sqlalchemy.orm import sessionmaker #创建会话对象 Session = sessionmaker(engine) #打开会话对象 db_session = Session() ##############1.正向添加数据################# stu_obj = Student(name="baozi",stu2sch=School(name="北京大学")) db_session.add(stu_obj) db_session.commit() ##############2.反向添加数据第一种写法################## sch_obj = School(name="天津大学") sch_obj.sch2stu = [Student(name="xiena"),Student(name="hejiong")] db_session.add(sch_obj) db_session.commit() ##############2.反向添加数据第二种写法 sch_obj = School(name="清华大学",sch2stu = [Student(name="维嘉"),Student(name="吴昕")]) db_session.add(sch_obj) db_session.commit()
基于relationship查询数据
#查询数据 from create_table_ForeignKey import engine,Student,School #创建会话窗口 from sqlalchemy.orm import sessionmaker #创建会话对象 Session = sessionmaker(engine) #打开会话窗口 db_session = Session() ##############1.正向数据查询############### stu_list = db_session.query(Student).all() for row in stu_list: print(row.id,row.name,row.school_id,row.stu2sch.name) ##############2.反向数据查询################ sch_list = db_session.query(School).all() for school in sch_list: for student in school.sch2stu: print(school.id,school.name,student.name)
更新数据
from sqlalchemy.orm import sessionmaker from create_table_ForeignKey import engine,Student,School Session = sessionmaker(engine) db_session = Session() # 修改数据 sch = db_session.query(School).filter(School.name=="清华大学").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 create_table_ForeignKey import engine,Student,School Session = sessionmaker(engine) db_session = Session() # 删除数据 sch = db_session.query(School).filter(School.name=="北京大学").first() db_session.query(Student).filter(Student.school_id == sch.id).delete() db_session.commit() db_session.close()
五、多对多表的增删改查
创建表及表关系
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() from sqlalchemy import Column,Integer,String,ForeignKey from sqlalchemy.orm import relationship class Girls(Base): __tablename__ = "girl" id = Column(Integer,primary_key=True) name = Column(String(32)) #创建关系 girl2boy = relationship("Boys",secondary="hotel",backref="boy2girl") class Boys(Base): __tablename__ = "boy" id = Column(Integer, primary_key=True) name = Column(String(32)) 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")) from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:口令@127.0.0.1:3306/数据库名?charset=utf8") Base.metadata.create_all(engine)
基于relationship增加数据
from sqlalchemy.orm import sessionmaker from M2M import engine,Boys,Girls Session = sessionmaker(engine) db_session = Session() #1.反向添加 boy_obj = Boys(name="haha",boy2girl = [Girls(name="赵丽颖"),Girls(name="陈妍希")]) db_session.add(boy_obj) db_session.commit() #2.正向添加 girl_obj = Girls(name="笑笑",girl2boy=[Boys(name="小土狗"),Boys(name="大狼狗")]) db_session.add(girl_obj) db_session.commit()
基于relationship查询数据
from sqlalchemy.orm import sessionmaker from M2M import engine,Boys,Girls Session = sessionmaker(engine) db_session = Session() # 1.查询 relationship 正向 g_list = db_session.query(Girls).all() for girl in g_list: for boy in girl.girl2boy: print(girl.name,boy.name) # 2.查询 relationship 反向 b_list = db_session.query(Boys).all() for boy in b_list: for girl in boy.boy2girl: print(boy.name,girl.name)