sqlalchemy

Posted 2021-02-09 玩蛇少年7372

create_table

# 1.导入SQLAlchemy
from sqlalchemy.ext.declarative import declarative_base
# 3.导入ORM对应数据库数据类型的字段
from sqlalchemy import Column, Integer, String
# 5.创建数据库连接
from sqlalchemy import create_engine
# 2.创建ORM模型基类
Base = declarative_base()


# 4.创建ORM对象
class User(Base):
    __tablename__ = ‘user‘
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), index=True)


# 数据库连接创建完成
engine = create_engine(‘mysql+pymysql://root:[email protected]地址:端口/数据库名称?charset=uft8‘, echo=True)

print(engine)
# 6.去数据库中创建与User所对应的数据表
# 去engine数据库中创建所有继承Base类的 ORM对象
Base.metadata.create_all(engine)

crud_insert

# 通过SQLAlchemy 对数据表进行增删改查
# crud == 增删改查
# 想要操控数据库 首先打开数据库连接
from sqlalchemy.orm import sessionmaker
# 创建回话窗口
from sqlalchemy import create_engine
engine = create_engine(‘mysql+pymysql://root:[email protected]地址:端口/数据库名称?charset=uft8‘, echo=True)


# 创建回话 - 打开数据库连接

Session = sessionmaker(engine)
# 打开会话窗口
db_session = Session()

# 1.增加数据 == insert
# insert into user(name)  values (‘zzy‘)
from create_table import User
db_session.add_all([
    User(name="施瓦辛格"),
    User(name="蒙娜丽莎"),
    User(name="杰森斯坦森")
])

db_session.commit()
db_session.close()
除了查询 其余增删改都需要commit and close


extend:
from crteate_table import User
user1 = User(name=‘111‘)
user2 = User(name=‘222‘)
user3 = User(name=‘333‘)

db_session.add(user1)
db_session.add(user2)
db_session.add(user3)

db.session.commit()
db.session.close()

crud_search

查询
会话窗口
from sqlalchemy.orm import sessionmaker
from create_table import engine, User

Sesssion = sessionmaker(engine)
db_session = Session()

1.简单查询
sql: select * from table;

user_list = db_session.query(User).all()
for row in user_list:
    print(row.id, row.name)

2.带条件的查询
user_list = db_session.query(User).filter(User.id == 4).all()
print(user_list[0].id,user_list[0].name)

user = db_session.query(User).filter_by(id=4).first()
print(user.id,user.name)

user_list = db_session.query(User).filter(User.id >= 4).all()
for row in user_list:
     print(row.name,row.id)

扩展 查看sql语句
sql = db_session.query(User).filter(User.id >= 4)
print(sql)

crud_update

# 更新修改数据
from sqlalchemy.orm import sessionmaker
from create_table import engine, User

Session = sessionmaker(engine)
db_session = Session()

# 1.修改数据 - 先查询预修改的数据
# update `user` set `name` = "666" where id = 1
# res = db_session.query(User).filter(User.name=="Ywb").update({"name":"武配齐"})
# print(res)
# db_session.commit()
# db_session.close()
# res = db_session.query(User).filter(User.id >= 5).update({"name":123456})
# print(res)
# db_session.commit()
# db_session.close()

crud_delete

# 删除数据
from sqlalchemy.orm import sessionmaker
from create_table import engine,User

Session = sessionmaker(engine)
db_session = Session()

# 1.删除 - 先查询预删除的数据
# delete from user where id=5
# res = db_session.query(User).filter(User.id==5).delete()
# print(res)
# db_session.commit()
# db_session.close()

# 2.删除多条数据
# res = db_session.query(User).filter(User.name=="123456").delete()
# print(res)
# db_session.commit()
# db_session.close()

create_table_Foreign Key

# 一对多建表操作
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")


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/day127?charset=utf8")

Base.metadata.create_all(engine)

crud_insert_foreign_key

# 一对多建表操作
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")


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/day127?charset=utf8")

Base.metadata.create_all(engine)

crud_search_foreign_key

# 添加数据
from sqlalchemy.orm import sessionmaker
from create_table_ForeignKey import engine,Student,School

Session = sessionmaker(engine)
db_session = Session()

# 1.查询数据 - relationship 版 正向
# stu = db_session.query(Student).all()
# for row in stu:
#     print(row.id,row.name,row.school_id,row.stu2sch.name)

# 2.查询数据 - relationship 版 反向
# sch = db_session.query(School).all()
# for school in sch:
#     for student in school.sch2stu:
#         print(school.id,school.name,student.name)





# 高级版查询操作,厉害了哦
# 老规矩
from create_table import User, engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

# 查询数据表操作
# and or
# res = db_session.query(User).filter(User.id == 4 or User.name == "DragonFire").all()
# print(res[0].name,res)
#
# if False and True:
#     print("lihaile")

# from sqlalchemy.sql import and_ , or_
# ret = db_session.query(User).filter(and_(User.id == 3, User.name == ‘DragonFire‘)).all()
# print(ret)
# ret = db_session.query(User).filter(or_(User.id == 4, User.name == ‘DragonFire‘)).all()
# print(ret)
# ret = db_session.query(User).filter_by(id=4,name="DragonFire").all()
# 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) # 15 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进行排序
# from sqlalchemy.sql import text
# r6 = db_session.query(User).filter(text("id=:value and name=:name")).params(value=3, name=‘DragonFire‘).first()
# print(r6.name)

# 原生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=3,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=3,name="DragonFire")

# 排序 :
# 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_by(name=‘DragonFire‘).all()
ret = session.query(User).filter(User.id > 1, User.name == ‘DragonFire‘).all()
ret = session.query(User).filter(User.id.between(1, 3), User.name == ‘DragonFire‘).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=‘DragonFire‘))).all() 子查询
from sqlalchemy import and_, or_
ret = session.query(User).filter(and_(User.id > 3, User.name == ‘DragonFire‘)).all()
ret = session.query(User).filter(or_(User.id < 2, User.name == ‘DragonFire‘)).all()
ret = session.query(User).filter(
    or_(
        User.id < 2,
        and_(User.name == ‘eric‘, User.id > 3),
        User.extra != ""
    )).all()
# select * from User where id<2 or (name="eric" and id>3) 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)[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(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()
"""

# ret = db_session.query(User).filter(User.id.between(1, 3)).all() # between 大于1小于3的
# print(ret)
# ret = db_session.query(User).filter(User.id.in_([1,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
# ret = db_session.query(User).filter(~User.id.in_([1,2,3])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
# print(ret)

# from sqlalchemy.sql import and_, or_
#
# ret = db_session.query(User).filter(
#     or_(
#         User.id < 2,
#         and_(User.name == ‘eric‘, User.id > 3),
#         User.name != ""
#     ))
# print(ret)

# 关闭连接
db_session.close()

crud_update_foreign_key

# 添加数据
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=="OldBoyShanghai").first()
db_session.query(Student).filter(Student.name == "小笼包").update({"school_id":sch.id})
db_session.commit()
db_session.close()


# 高级版更新操作
from create_table import User, engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

# 直接修改
# 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()

crud_delete_foreign_key

# 添加数据
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=="OldBoyShanghai").first()
db_session.query(Student).filter(Student.school_id == sch.id).delete()
db_session.commit()
db_session.close()