SQLAlchemy基本使用，创建表，增删改查

Posted 2022-03-09 robertx

基础语法

• 创建连接

from sqlalchemy import create_engine
# 写法1
engine = create_engine("postgresql://scott:[email protected]/test?charset=utf8")

# 写法2
engine = create_engine("mysql+pymysql://root:[email protected]/test",encoding='latin1', echo=True")

URL的字符串形式是 dialect[+driver]://user:[email protected]/dbname[?key=value..]，在这里 dialect是一个数据库的名称，如mysql，oracle， postgresql等等，和driver一个DBAPI的名称，诸如 psycopg2，pyodbc，cx_oracle等。或者，该URL可以是一个实例。

• 创建回话,

  用来与数据库交谈,ORM对数据库的“处理”是Session。当我们第一次设置应用程序时，在与create_engine()语句相同的级别上，我们定义一个Session类，它将作为新Session 对象的工厂

from sqlalchemy import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

---

SQLAlchemy创建表,及简单的CRUD

1.创建表

# 1. 创建连接
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

# 2. 创建基类, 相当于 django orm中的model.model
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column,Integer,String
# 3. 创建表模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

Base.metadata.create_all(engine)

2.单表的CRUD

增加数据

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

# 1. 添加单条数据
# u = User(name='用户1')
# db_session.add(u)

# 2. 添加多条数据
db_session.add_all([
    User(name='用户2'),
    User(name='用户3'),
])

# 结束记得提交,数据才能保存在数据库中
db_session.commit()
# 关闭会话
db_session.close()

查询数据

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

# 查询用户表中所有用户
user_all = db_session.query(User).all()

# where条件查询
user = db_session.query(User).filter(User.id>=2).all()
# 取出一条数据
user = db_session.query(User).filter(User.id>=2).first()

db_session.close()

修改数据

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()


# 更新单条数据
res = db_session.query(User).filter(User.id==20).update("name":"新用户")
db_session.commit()
db_session.close()

# 更新多条数据
res2 = db_session.query(User).filter(User.id<=20).update("name":"修改用户")
print(res)  # 这里结果的数字是更新的行数

db_session.commit()
db_session.close()

删除数据

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

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

db_session.commit()
db_session.close()

复杂条件查询

# 1.and_ or_ 条件查询
from sqlalchemy.sql import and_,or_
ret = db_session.query(User).filter(and_(User.id>3,User.name=='xxx')).all()
ret2 = db_session.query(User).filter(or_(User.id<2,User.name=='xxx')).all()

# 2. 查询所有数据
res = db_session.query(User).all()

# 3. 查询数据,指定查询数据列加入别名
r3 = db_session.query(User.name.label('username'),User.id).firset()
print(r2.id,r2.username)

# 4. 筛选条件格式
r4 = db_session.query(User).filte(User.name='xxx').all()
r5 = db_session.query(User).filter_by(name='xxx').first()

# 6. 原生sql查询
r6 = db_session.query(User).from_statement(text("select * from User where name=:name")).params(name='xxx').all()

# 7. 筛选查询列, query的时候不再使用User ORM对象,而是使用User.name来对内容进行选取
user_list = db_session.query(User.name).all()
print(user_list)

for row in user_list:
    print(row.name)

# 8. 复杂查询
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='xxx')

# 9. 排序
user_list = db_session.query(User).order_by(User.id.desc()).all()

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

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

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

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

# 14. 通配符
ret = db_session.query(User).filter(~User.name.like('e%')).all()

# 15. 限制
ret = db_session.query(User)[1:2]

# 16. 分组
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).having(func.min(User.id) >2).all()

复杂修改数据

#高级版更新操作
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 > 0).update("name" : "xxx")

#在原有值基础上添加 方式1, 名字后面加后缀
db_session.query(User).filter(User.id > 0).update(User.name: User.name + "后缀", synchronize_session=False)

#在原有值基础上添加 方式2, 年龄加一岁
db_session.query(User).filter(User.id > 0).update("age": User.age + 1, synchronize_session="evaluate")
db_session.commit()

3.一对多的操作 外键 Foreignkey

1.创建数据表及外键关系

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship



class Scholl(Base):
    __tablename__ = 'school'
    id = Column(Integer,primary_key=True)
    name = Column(String(32))

class Student(Base):
    __tablename__ = 'student'
    id = Column(Integer,primary_key=True)
    name = Column(String(32))

    # 关联字段,让class_id 与 class 的 id 进行关联,主外键关系(这里的ForeignKey一定要是表名.id不是对象, 这是数据库层面的
    school_id = Column(Integer,ForeignKey('school.id'))
    
    # orm层面的关系,数据库中不存在这个字段
    to_school = relationship("School",backref='to_student')

Base.metadata.create_all(engine)

添加数据

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship



class School(Base):
    __tablename__ = 'school'
    id = Column(Integer,primary_key=True)
    name = Column(String(32))

class Student(Base):
    __tablename__ = 'student'
    id = Column(Integer,primary_key=True)
    name = Column(String(32))

    # 关联字段,让class_id 与 class 的 id 进行关联,主外键关系(这里的ForeignKey一定要是表名.id不是对象
    school_id = Column(Integer,ForeignKey('school.id'))

    to_school = relationship("School",backref='to_student')

# Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine)
db_session = Session()

# 正向添加字段
s = Student(name='小明',to_school=School(name='太阳小学'))
s2 = Student(name='小亮',to_school=db_session.query(School).filter(School.name=='太阳小学').first())
db_session.add(s)

# 反向添加字段
school = School(name='月亮小学')
school.to_student=[Student(name='xxx1'),Student(name='xxx2')]
db_session.add(school)
db_session.commit()

查询数据

# 正向查询
student_list = db_session.query(Student).all()
for stu in student_list:
    print(stu.name,stu.to_school.name)

# 反向查询
school_list = db_session.query(School).all()
for school in school_list:
    print(school.name,end='')
    for s in school.to_student:
        print(s.name)

修改数据

school = db_session.query(School).filter(School.name=='太阳小学').first()
db_session.query(Student).filter(Student.school_id==school.id).update('name':'太阳小学生')

db_session.commit()

删除数据

school = db_session.query(School).filter(School.name=='太阳小学').first()
db_session.query(Student).filter(Student.id==school.id).delete()
 
db_session.commit()
db_session.close()

3. 多对多关系

创建表及关系

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test',encoding='utf8')

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from sqlalchemy import Integer,Column,String,ForeignKey
from sqlalchemy.orm import relationship

class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer,primary_key=True)
    name = Column(String(16))

    to_girl = relationship('Boy',secondary='hotel',backref='to_boy')

class Girl(Base):
    __tablename__ = 'girl'
    id = Column(Integer,primary_key=True)
    name = Column(String(16))

class Hotel(Base):
    __tablename__ = 'hotel'
    id = Column(Integer,primary_key=True)
    boy_id = Column(Integer,ForeignKey('boy.id'))
    girl_id = Column(Integer,ForeignKey('girl.id'))

Base.metadata.create_all(engine)

基于relationship增加数据 多对多

boy = Boy(name='男孩1')
boy.b_to_g = [Girl(name='女1'),Girl(name='女2')]

db_session.add(boy)
db_session.commit()

基于relationship查询数据 多对多

# 创建连接
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.ext.declarative import declarative_base
# 创建基类
Base = declarative_base()

from sqlalchemy import Integer,Column,String,ForeignKey
from sqlalchemy.orm import relationship
# 创建表关系
class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer,primary_key=True)
    name = Column(String(16))

    b_to_g = relationship('Girl',secondary='hotel',backref='g_to_b')

class Girl(Base):
    __tablename__ = 'girl'
    id = Column(Integer,primary_key=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

class Hotel(Base):
    __tablename__ = 'hotel'
    id = Column(Integer,primary_key=True)
    boy_id = Column(Integer,ForeignKey('boy.id'))
    girl_id = Column(Integer,ForeignKey('girl.id'))

# Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine)
db_session = Session()

# 正向查询
res = db_session.query(Boy).all()
for boy in res:
    print(boy.id,boy.name,boy.b_to_g)

# 反向查询
res = db_session.query(Girl).first()
for boy in res.g_to_b:
    print(boy.id,boy.name)