Python SQLAlchemy快速入门教程

0、模块的安装

pip install SQLAlchemy

一、单表的操作

1、单表的创建

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/5
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine import create_engine

# 声明所有的ORM类对象继承的基类
BaseModel = declarative_base()

# Column 定义列的数据
# Integer、String数据类型
class User(BaseModel):
    __tablename__ = "t_user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, index=True, unique=True)

if __name__ == '__main__':
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

    # 创建数据库表
    BaseModel.metadata.create_all(engine)

2、单表的增删改查

2.1、单表插入数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/5

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

# 声明所有的ORM类对象继承的基类
BaseModel = declarative_base()

# Column 定义列的数据
# Integer、String数据类型
class User(BaseModel):
    __tablename__ = "t_user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, index=True, unique=True)

if __name__ == '__main__':
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

    # 创建sql会话
    db_sessoin = sessionmaker(engine)()

    # 插入数据
    user = User(name='单条插入，你好1')
    user_list = [
        User(name='批量插入，你好_list_1'),
        User(name='批量插入，你好_list_2'),
        User(name='批量插入，你好_list_3')
    ]
    db_sessoin.add(user)  # 单条插入
    db_sessoin.add_all(user_list)  # 批量插入

    db_sessoin.commit()  # 提交数据
    db_sessoin.close()  # 关闭会话

插入数据代码

2.2、单表查询数据

2.2.1、单表无条件查询数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/5
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

# 声明所有的ORM类对象继承的基类
BaseModel = declarative_base()


# Column 定义列的数据
# Integer、String数据类型
class User(BaseModel):
    __tablename__ = "t_user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, index=True, unique=True)


if __name__ == '__main__':
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

    # 创建sql会话
    db_sessoin = sessionmaker(engine)()

    # 查询所有数据
    user_list = db_sessoin.query(User).all()
    print('查询所有的数据')
    for user_obj in user_list:
        print(user_obj.id, user_obj.name)

    # 查询第一条数据
    first_user = db_sessoin.query(User).first()
    print('查询第一条数据')
    print(first_user.id, first_user.name)

    db_sessoin.close()  # 关闭会话

单表无条件查询数据代码

2.2.2、单表有条件查询数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/5
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

# 声明所有的ORM类对象继承的基类
BaseModel = declarative_base()

# Column 定义列的数据
# Integer、String数据类型
class User(BaseModel):
    __tablename__ = "t_user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, index=True, unique=True)

if __name__ == '__main__':
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

    # 创建sql会话
    db_sessoin = sessionmaker(engine)()

    # 查询指定的数据
    user_list1 = db_sessoin.query(User).filter(User.id == 3).all()
    user_list2 = db_sessoin.query(User).filter_by(id=3).all()

    # and
    ret_list = db_sessoin.query(User).filter(User.id == 2, User.name == '批量插入，你好_list_2').all()

    # 使用and链接起来，其实是or的关系
    # ret_list = db_sessoin.query(User).filter(User.id == 2 and User.name == '批量插入，你好_list_2').all()

    db_sessoin.close()  # 关闭会话

单表有条件查询数据的代码

2.3、单表修改数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/5
# @Author  : suk
# @File    : first_orm.py
# @Software: PyCharm
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

# 声明所有的ORM类对象继承的基类
BaseModel = declarative_base()


# Column 定义列的数据
# Integer、String数据类型
class User(BaseModel):
    __tablename__ = "t_user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, index=True, unique=True)


if __name__ == '__main__':
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

    # 创建sql会话
    db_sessoin = sessionmaker(engine)()

    ret = db_sessoin.query(User).filter(User.id == 1).update(
        {
            'name': '张三'
        }
    )

    db_sessoin.commit()
    db_sessoin.close()  # 关闭会话

单表修改数据代码

2.4、单表删除数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/5
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

# 声明所有的ORM类对象继承的基类
BaseModel = declarative_base()

# Column 定义列的数据
# Integer、String数据类型
class User(BaseModel):
    __tablename__ = "t_user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False, index=True, unique=True)

if __name__ == '__main__':
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

    # 创建sql会话
    db_sessoin = sessionmaker(engine)()

    ret = db_sessoin.query(User).filter(User.id == 2).delete()

    db_sessoin.commit()
    db_sessoin.close()  # 关闭会话

单表删除数据代码

二、一对一的操作

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

Base_Model = declarative_base()

class Wife(Base_Model):
    __tablename__ = 'wife'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    husband = relationship("Husband", uselist=False, back_populates="wife")  # uselist=False,不能使用列表批量增加数据

class Husband(Base_Model):
    __tablename__ = 'husband'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    wife_id = Column(Integer, ForeignKey('wife.id'))
    wife = relationship("Wife", back_populates="husband")

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    # 创建数据表
    Base_Model.metadata.create_all(engine)
    db_session = sessionmaker(engine)()

    # 增加数据
    wife_obj = Wife(name='女1', husband=Husband(name='男1'))
    db_session.add(wife_obj)

    # 删除数据
    db_session.query(Husband).filter(Husband.id == '1').delete()

    # 修改数据
    ret = db_session.query(Wife).filter(Wife.name == '女1').one()
    ret.husband = db_session.query(Husband).filter(Husband.name == '男3').one()

    # 查询数据
    # 正向查询
    ret = db_session.query(Husband).filter(Husband.name == '男3').one()
    print(ret.name, ret.wife.name)

    # 反向查询
    ret = db_session.query(Wife).filter(Wife.name == '女1').one()
    print(ret.name, ret.husband.name)

    db_session.commit()
    db_session.close()

一对一创建表和CRUD代码

三、一对多的操作

1、一对多表的创建

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

BaseModel = declarative_base()

engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')

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

class Student(BaseModel):
    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    sch_id = Column(Integer, ForeignKey('school.id'))  # 这个是数据库层面关联

    # ORM层面关联
    stu2sch = relationship('School', backref='sch2stu')

if __name__ == '__main__':
    # 创建表
    BaseModel.metadata.create_all(engine)

2、一对多的增删改查

2.1、一对多插入数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

BaseModel = declarative_base()

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

class Student(BaseModel):
    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    sch_id = Column(Integer, ForeignKey('school.id'))  # 这个是数据库层面关联

    # ORM层面关联
    stu2sch = relationship('School', backref='sch2stu')

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    db_session = sessionmaker(engine)()
    # 正向插入
    stu = Student(name='张三', stu2sch=School(name='清华大学'))
    db_session.add(stu)

    # 反向插入
    sch = School(name='北京大学')
    sch.sch2stu = [
        Student(name='王五'),
        Student(name='赵龙')
    ]
    db_session.add(sch)
    db_session.commit()
    db_session.close()

一对多插入数据

2.2、一对多查询数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
# @Author  : suk
# @File    : fk.py
# @Software: PyCharm
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

BaseModel = declarative_base()

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

class Student(BaseModel):
    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    sch_id = Column(Integer, ForeignKey('school.id'))  # 这个是数据库层面关联

    # ORM层面关联
    stu2sch = relationship('School', backref='sch2stu')

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    db_session = sessionmaker(engine)()

    # 正向查询
    students = db_session.query(Student).all()
    print('正向查询')
    for student in students:
        print(student.name, student.stu2sch.name)

    # 反向查询
    schools = db_session.query(School).all()
    print('反向查询')
    for school in schools:
        for stu in school.sch2stu:
            print(school.name, stu.name)

    db_session.close()

一对多查询数据代码

2.3、一对多修改数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6

from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

BaseModel = declarative_base()

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

class Student(BaseModel):
    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    sch_id = Column(Integer, ForeignKey('school.id'))  # 这个是数据库层面关联

    # ORM层面关联
    stu2sch = relationship('School', backref='sch2stu')

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    db_session = sessionmaker(engine)()

    # 正向修改，思路：先查询学生，再修改学校
    school_obj = db_session.query(School).filter(School.name == '清华大学').one()
    student_obj = db_session.query(Student).filter(Student.id == 1).one()
    student_obj.stu2sch = school_obj

    # 反向修改，思路：先查询学校，再将学校下面的学生转移到新的学校
    school_obj = db_session.query(School).filter(School.name == '北京大学').one()
    student_list = school_obj.sch2stu

    school_obj = db_session.query(School).filter(School.name == '清华大学').one()
    school_obj.sch2stu = school_obj.sch2stu + student_list

    db_session.commit()
    db_session.close()

一对多修改数据代码

2.4、一对多删除数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker

BaseModel = declarative_base()

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

class Student(BaseModel):
    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    sch_id = Column(Integer, ForeignKey('school.id'))  # 这个是数据库层面关联

    # ORM层面关联
    stu2sch = relationship('School', backref='sch2stu')

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    db_session = sessionmaker(engine)()

    # 反向修改，思路：先查询学校，再将学校下面的学生删除
    school_obj = db_session.query(School).filter(School.name == '北京大学').one()

    # school_obj.sch2stu.clear()  # 清除关联
    for sch in school_obj.sch2stu:
        db_session.query(Student).filter(Student.id == sch.id).delete()

    db_session.commit()
    db_session.close()

一对多删除数据代码

四、多对多的操作

1、多对多表的创建

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

Base_Model = declarative_base()

class Girl(Base_Model):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # secondary="hotel",数据表中的数据才能证明两者关系
    g2b = relationship('Boy', backref='b2g', secondary='hotel')

class Boy(Base_Model):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

class Hotel(Base_Model):
    __tablename__ = 'hotel'
    id = Column(Integer, primary_key=True)
    gid = Column(Integer, ForeignKey('girl.id'))
    bid = Column(Integer, ForeignKey('boy.id'))

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    Base_Model.metadata.create_all(engine)

2、多对多的增删改查

2.1、多对多插入数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

Base_Model = declarative_base()

class Girl(Base_Model):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # secondary="hotel",数据表中的数据才能证明两者关系
    g2b = relationship('Boy', backref='b2g', secondary='hotel')

class Boy(Base_Model):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

class Hotel(Base_Model):
    __tablename__ = 'hotel'
    id = Column(Integer, primary_key=True)
    gid = Column(Integer, ForeignKey('girl.id'))
    bid = Column(Integer, ForeignKey('boy.id'))


if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    # 创建数据表
    # Base_Model.metadata.create_all(engine)
    db_session = sessionmaker(engine)()

    # 正向增加
    g = Girl(name='小红', g2b=[Boy(name='张三'), Boy(name='李四')])
    db_session.add(g)

    # 反向增加
    b = Boy(name='张某')
    b.b2g = [
        Girl(name='女某1'),
        Girl(name='女某2'),
        Girl(name='女某3'),
    ]

    db_session.add(b)
    db_session.commit()
    db_session.close()

多对多插入数据

2.2、多对多查询数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

Base_Model = declarative_base()

class Girl(Base_Model):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # secondary="hotel",数据表中的数据才能证明两者关系
    g2b = relationship('Boy', backref='b2g', secondary='hotel')

class Boy(Base_Model):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

class Hotel(Base_Model):
    __tablename__ = 'hotel'
    id = Column(Integer, primary_key=True)
    gid = Column(Integer, ForeignKey('girl.id'))
    bid = Column(Integer, ForeignKey('boy.id'))

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    # 创建数据表
    # Base_Model.metadata.create_all(engine)
    db_session = sessionmaker(engine)()

    # 正向查询
    girl_list = db_session.query(Girl).filter(Girl.name == '小红').all()
    for girl in girl_list:
        boy_list = girl.g2b
        for boy in boy_list:
            print(girl.name, boy.name)
    # 反向查询
    boy_list = db_session.query(Boy).filter(Boy.name == '张三').all()

    for boy in boy_list:
        girl_list = boy.b2g
        for girl in girl_list:
            print(boy.name, girl.name)

    db_session.close()

多对多查询数据代码

2.3、多对多修改数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

Base_Model = declarative_base()

class Girl(Base_Model):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # secondary="hotel",数据表中的数据才能证明两者关系
    g2b = relationship('Boy', backref='b2g', secondary='hotel')

class Boy(Base_Model):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

class Hotel(Base_Model):
    __tablename__ = 'hotel'
    id = Column(Integer, primary_key=True)
    gid = Column(Integer, ForeignKey('girl.id'))
    bid = Column(Integer, ForeignKey('boy.id'))

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    # 创建数据表
    # Base_Model.metadata.create_all(engine)
    db_session = sessionmaker(engine)()

    girl_obj = db_session.query(Girl).filter(Girl.name == '女某1').first()
    boy_obj = db_session.query(Boy).filter(Boy.name == '李四').first()

    # 将关联清除，再重新绑定
    girl_obj.g2b.clear()
    girl_obj.g2b.append(boy_obj)

    db_session.commit()
    db_session.close()

多对多修改数据代码

2.4、多对多删除数据

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/6
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.engine import create_engine

Base_Model = declarative_base()

class Girl(Base_Model):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    # secondary="hotel",数据表中的数据才能证明两者关系
    g2b = relationship('Boy', backref='b2g', secondary='hotel')

class Boy(Base_Model):
    __tablename__ = 'boy'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)

class Hotel(Base_Model):
    __tablename__ = 'hotel'
    id = Column(Integer, primary_key=True)
    gid = Column(Integer, ForeignKey('girl.id'))
    bid = Column(Integer, ForeignKey('boy.id'))

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://test:test@127.0.0.1:3306/celery_db?charset=utf8')
    # 创建数据表
    # Base_Model.metadata.create_all(engine)
    db_session = sessionmaker(engine)()

    girl_obj = db_session.query(Girl).filter(Girl.name == '小红').first()
    boy_obj = db_session.query(Boy).filter(Boy.name == '张某').first()

    girl_obj.g2b.clear()

    # 删除所有的数据
    girl_obj.g2b.remove(boy_obj)

    # 删除指定的数据
    db_session.add(girl_obj)
    db_session.commit()
    db_session.close()

多对多删除数据代码

五、总结

更新请参考官方文档：
https://docs.sqlalchemy.org/en/13/