SQLAlchemy 基础
普通操作:
创建表
from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Integer, String, ColumnBase = declarative_base()class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True, index=True) name = Column(String(32), nullable=False)from sqlalchemy.engine import create_engineengine = create_engine('mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy123?charset=utf8')Base.metadata.create_all(engine)CRUD (增删改查)
from sqlalchemy.orm import sessionmakerfrom sqlalchemy import create_engineengine = create_engine('mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy123?charset=utf8')# 连接数据库Session = sessionmaker(engine)# 实例化连接db_session = Session()# 1. 增加数据from CreateTable import User# 插入单挑user = User(name="张飞")db_session.add(user)db_session.commit()# 插入多条db_session.add_all([User(name="李林"), User(name='悦悦')])db_session.commit()db_session.close()# 2.查询 select * from tableres = db_session.query(User).all()for user in res: print(user.id,user.name)res = db_session.query(User).first()print(res.id,res.name)res = db_session.query(User).filter(User.id == 3).first()print(res.name)res = db_session.query(User).filter(User.id <= 2).all()for u in res: print(u.id,u.name)res = db_session.query(User).filter(User.id == 1 , User.name=="张飞").first()print(res.name)# 3.更改数据# update user set name="" where id=1res = db_session.query(Users).filter(Users.id == 1).update({"name":"DragonFire"})print(res)db_session.commit()res = db_session.query(Users).update({"name":"DragonFire"})print(res)db_session.commit()# 4.删除# delete from table where id=1res = db_session.query(Users).filter(Users.id == 1).delete()print(res)db_session.commit()res = db_session.query(Users).delete()print(res)db_session.commit()
外键 :
创建表
from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import relationshipBase = declarative_base()from sqlalchemy import Column, Integer, String, ForeignKeyclass Student(Base): __tablename__ = "student" id = Column(Integer, primary_key=True, autoincrement=True, index=True) name = Column(String(32), nullable=False) # 外键 sch_id = Column(Integer, ForeignKey('school.id')) # 用于 正向 插入 查询 backref='sch2stu' 用于反向操作 stu2sch = relationship('School', backref='sch2stu')class School(Base): __tablename__ = 'school' id = Column(Integer, primary_key=True, autoincrement=True, index=True) name = Column(String(32), nullable=False)from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/SQLAlchemy123?charset=utf8")Base.metadata.create_all(engine) - 操作
from sqlalchemy.orm import sessionmakerfrom CreateTable_ForeignKey import engineSession = sessionmaker(engine)db_session = Session()from CreateTable_ForeignKey import Student, School# 1.添加数据# 普通添加sch = School(name="Beijing")db_session.add(sch)db_session.commit()sch_info = db_session.query(School).filter(School.name == "Beijing").first()print(sch_info.name)stu = Student(name="张飞", sch_id=sch_info.id)db_session.add(stu)db_session.commit()db_session.close()# # 使用 relationship 正向添加stu = Student(name="李林", stu2sch=School(name="Shanghai"))db_session.add(stu)db_session.commit()# relationship 反向添加sch = School(name="Tianjin")sch.sch2stu = [Student(name="娃娃"), Student(name="阿道夫")]db_session.add(sch)db_session.commit()# # 2.查询res = db_session.query(Student).all()for stu in res: print(stu.name, stu.stu2sch.name)res = db_session.query(School).all()for sch in res: for stu in sch.sch2stu: print(sch.name, stu.name)
多对多:
- 创建表
from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()from sqlalchemy import String,Integer,Column,ForeignKey,create_engineengine = create_engine("mysql+pymysql://root:@127.0.0.1:3306/sqlalchemy666?charset=utf8")from sqlalchemy.orm import relationshipclass Girl(Base): __tablename__ = "girls" id=Column(Integer,primary_key=True) name=Column(String(32),nullable=False) g2b = relationship("Boy",backref="b2g",secondary="hotels")class Boy(Base): __tablename__ = "boys" id=Column(Integer,primary_key=True) name=Column(String(32),nullable=False)class Hotel(Base): __tablename__ = "hotels" id=Column(Integer,primary_key=True) boy_id = Column(Integer,ForeignKey("boys.id"),nullable=False) girl_id = Column(Integer,ForeignKey("girls.id"),nullable=False)Base.metadata.create_all(engine) - 操作表
from sqlalchemy.orm import sessionmakerfrom CreateTableM2M import engineSession = sessionmaker(engine)db_session = Session()from CreateTableM2M import Girl,Boy# 1.增加数据# relationship 正向添加g = Girl(name="赵丽颖",g2b=[Boy(name="DragonFire"),Boy(name="冯绍峰")])db_session.add(g)db_session.commit()# relationship 反向添加b = Boy(name="AlexDSB")b.b2g = [Girl(name="罗玉凤"),Girl(name="娟儿"),Girl(name="芙蓉姐姐")]db_session.add(b)db_session.commit()# 2.查询# relationship 正向res = db_session.query(Girl).all()for g in res: for b in g.g2b: print(g.name,b.name)# relationship 反向res = db_session.query(Boy).all()for b in res: for g in b.b2g: print(b.name,g.name)高级操作
高级操作
# 高级版查询操作,厉害了哦# 老规矩from CreateTable import User, enginefrom sqlalchemy.orm import sessionmakerSession = sessionmaker(engine)db_session = Session()# 查询数据表操作# and orfrom sqlalchemy.sql import and_, or_, descret = db_session.query(User).filter(and_(User.id > 3, User.name == '张飞')).all()ret = db_session.query(User).filter(or_(User.id < 2, User.name == '李林')).all()ret = db_session.query(User).filter( or_( User.id < 2, and_( User.name == 'eric', User.id > 3 ), User.name != "" ))print(ret)# select * from User where id<2 or (name="eric" and id>3) or extra != ""# 查询所有数据r1 = db_session.query(User).all()# 查询数据 指定查询数据列 加入别名r2 = db_session.query(User.name.label('username'), User.id).first()print(r2.id, r2.username) # 15 NBDragon# 表达式筛选条件r3 = db_session.query(User).filter(User.name == "张飞").all()# 原生SQL筛选条件r4 = db_session.query(User).filter_by(name='张飞').all()r5 = db_session.query(User).filter_by(name='张飞').first()# 字符串匹配方式筛选条件 并使用 order_by进行排序r6 = db_session.query(User).order_by(User.name.desc()).all()for i in r6: print(i.id, i.name)# 原生SQL查询from sqlalchemy.sql import textr7 = 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 nickuser_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 == "张飞").all()user_list = db_session.query(User).filter(User.name == "张飞").first()user_list = db_session.query(User).filter_by(name="张飞").first()for row in user_list: print(row.nick)# 复杂查询from sqlalchemy.sql import textuser_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3, name="张飞")# 查询语句from sqlalchemy.sql import textuser_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3, name="张飞")# 排序 :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='张飞').all()ret = session.query(User).filter(User.id > 1, User.name == '张飞').all()ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='张飞'))).all() 子查询from sqlalchemy import and_, or_ret = session.query(User).filter(and_(User.id > 3, User.name == '张飞')).all()ret = session.query(User).filter(or_(User.id < 2, User.name == '张飞')).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 funcret = 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) >2).all()"""# 关闭连接db_session.close()ret = db_session.query(User).filter(User.id.between(1, 3)).all() # between 大于1小于3的for i in ret: print(i.id, i.name)ret = db_session.query(User).filter(~User.id.in_([1, 4])).all() # in_([1,3,4]) 只查询id等于1,3,4的for i in ret: print(i.id, i.name)ret = db_session.query(User).filter(~User.id.in_([1, 3, 4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的for i in ret: print(i.id, i.name)# 通配符ret = db_session.query(User).filter(User.name.like('%e%')).all()for i in ret: print(i.id, i.name)ret = db_session.query(User).filter(~User.name.like('Z%')).all()for i in ret: print(i.id, i.name)# 高级版更新操作from CreateTable import User, enginefrom sqlalchemy.orm import sessionmakerSession = sessionmaker(engine)db_session = Session()# 直接修改db_session.query(User).filter(User.id > 3).update({User.name: User.name + "099"}, synchronize_session=False)db_session.commit()db_session.query(User).filter(User.id > 3).update({"name": User.name + "123"}, synchronize_session=False)db_session.commit()# 在原有值基础上添加 - 1db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)# 在原有值基础上添加 - 2db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")db_session.commit()