1. 程式人生 > 資料庫 >Python SQLAlchemy快速入門教程

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:[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/