SQLAlchemy

Posted 2021-02-09 amberLIU

Python 的 ORM 框架 SQLAlchemy

Object Relation Mapping
Object - Table 通过 Object 去操纵数据表

 

新建单表 create_table.py

from sqlalchemy.ext.declarative import declarative_base  # orm 官宣基类
from sqlalchemy import Column, Integer, String  # orm 数据类型字段
from sqlalchemy import create_engine  # 创建数据库连接

# 常见ORM 模型基类
Base = declarative_base()


class User(Base):
    __tablename__ = ‘user‘
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(20))


engine = create_engine(‘mysql+pymysql://root:@127.0.0.1:3306/day127sqlalchemy?charset=utf8‘)

# 去engine 数据库中创建所有继承Base 的orm 对象
Base.metadata.create_all(engine)

 

单表的增删改查 curd_one_table.py

from sqlalchemy.orm import sessionmaker

from create_table import engine, User

# crud 增删改查 : create/ retrieve/ update/ delete


# 创建会话, 打开数据库连接(通过engine)
Session = sessionmaker(engine)
# 打开会话窗口
db_session = Session()


# 增 insert into 表名(字段1, 字段2) values(值1, 值2)
user_1 = User(name=‘May‘)
db_session.add(user_1)
# 增 删 改 都要 commit
db_session.commit()
db_session.close()
# 批量增加
db_session.add_all([User(name=‘Sandy‘), User(name=‘Sheldon‘)])
db_session.commit()
db_session.close()


# 查 select * from 表名 where ....
res1 = db_session.query(User).all()
print(res1)  # 列表套对象
# 带条件的查询 filter
res2 = db_session.query(User).filter(User.id > 2).first()
print(res2.name)
res3 = db_session.query(User).filter_by(id = 3).first()
print(res3.name)


# 改 update user set name=Ryan where id=1
res4 = db_session.query(User).filter(User.id == 1).update({‘name‘: ‘ryan‘})
db_session.commit()
db_session.close()


# 删 delete from user where ...
res5 = db_session.query(User).filter(User.id == 5).delete()
db_session.commit()
db_session.close()

 

ForeignKey的创建 create_ForeignKey_table.py

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

Base = declarative_base()


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


class Student(Base):
    __tablename__ = ‘student‘
    id = Column(Integer, primary_key=True)
    name = Column(String(20))
    # 关联字段, 让school_id 和school的id 字段关联, 注意一定是foreignkey一定是表名
    school_id = Column(Integer, ForeignKey(‘school.id‘))
    # 将school和student 创建关系, backref 是反向关联字段
    stu2sch = relationship(‘School‘, backref=‘sch2stu‘)


engine = create_engine(‘mysql+pymysql://root:@127.0.0.1:3306/day127sqlalchemy?charset=utf8‘)
Base.metadata.create_all(engine)

 

ForeignKey的增删改查 curd_foreign_key_table.py

from sqlalchemy.orm import sessionmaker

from create_ForeignKey_table import engine, School, Student

Session = sessionmaker(engine)
db_session = Session()


# 增
# 1.笨方法
school_obj = School(name=‘Beijing_college‘)
db_session.add(school_obj)
db_session.commit()
school_id = db_session.query(School).filter(School.name == ‘Beijing_college‘).first().id
student_obj = Student(name=‘001‘, school_id=school_obj.id)
db_session.add(student_obj)
db_session.commit()
db_session.close()
# 2. 正向 relationship
student2 = Student(name=‘002‘, stu2sch=School(name=‘Shanghai_college‘))
db_session.add(student2)
db_session.commit()
db_session.close()
# 3. 反向 relationship
school_3 = School(name=‘Guangzhou_college‘)
school_3.sch2stu = [Student(name=‘003‘), Student(name=‘004‘)]
db_session.add(school_3)
db_session.commit()
db_session.close()


# 查
# 正向查询
# res1 = db_session.query(Student).filter(Student.stu2sch.id > 2).all()
# 报错. 类 没有 stu2sch 这个属性, 但是 实例化出来的对象可以
student_list = db_session.query(Student).all()
for student in student_list:
    print(student.id, student.name, student.stu2sch.name)
# 反向查询
school_list = db_session.query(School).all()
for school in school_list:
    print(school.sch2stu)  # [<create_ForeignKey_table.Student object at 0x0000009742531B70>]
    for student in school.sch2stu:
        print(school.name, student.name)


# 改
school_obj = db_session.query(School).filter(School.id == 3).first()
res = db_session.query(Student).filter(Student.name == ‘001‘).update({Student.school_id: school_obj.id})
db_session.commit()
student_obj = db_session.query(Student).filter(Student.name == ‘001‘).first()
print(student_obj.school_id, student_obj.stu2sch.name)  # 3 Shanghai_college
db_session.close()


# 删
sch = db_session.query(School).filter(School.id == 3).first()
db_session.query(Student).filter(Student.school_id == sch.id).delete()
db_session.commit()
db_session.close()

 

更多查询操作, 待补充, 先整理下再贴