Python SQLAlchemy快速入門教程
阿新 • • 發佈:2020-11-07
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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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/