Python通过SQLAlchemy操作sqlite3

定义表结构时, 可选参数如下

• primary_key=True 是否为主键
• nullable=False 是否可为空
• unique=True 是否唯一
• default=None 默认值
• index=True 是否创建索引
• foreign_key=True 是否为外键
• autoincrement=True 是否自增
• ondelete=”CASCADE” 删除时的行为

eg:

# database_util.py
# 可以正常使用非空/重复约束等. 但不能体现外键约束
# 外键约束需要如下
# with engine.connect() as conn:
#    conn.execute(text("PRAGMA foreign_keys = ON;"))

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

# 创建数据库引擎
DATABASE_URL = 'sqlite:///example.db'
engine = create_engine(DATABASE_URL)

# 创建一个基础类
Base = declarative_base()

# 定义 Student 表
class Student(Base):
    __tablename__ = 'students'
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False, unique=True)
    age = Column(Integer, nullable=False)

    records = relationship("Record", back_populates="student", cascade="all, delete-orphan")
    
	def __repr__(self):
	    return f"Student(id={self.id}, name={self.name}, age={self.age}"


# 定义 Record 表
class Record(Base):
    __tablename__ = 'records'
    
    id = Column(Integer, primary_key=True)
    subject = Column(String, nullable=False)
    score = Column(Integer, nullable=False)
    student_id = Column(Integer, ForeignKey('students.id'), nullable=False)

    student = relationship("Student", back_populates="records")

	def __repr__(self):
	    return f"Record(id={self.id}, score={self.score}, stu_id={self.student_id})"
# 创建会话

class DatabaseUtil:
    def __init__(self):
        # 创建表
        self.Session = sessionmaker(bind=engine)

    def init_db(self):
        Base.metadata.create_all(engine)

    def add_student(self, name, age):
        with self.Session() as session:
            if session.query(Student).filter(Student.name == name).first():
                print(f"Student {name} already exists")
                return
            new_student = Student(name=name, age=age)
            session.add(new_student)
            session.commit()
        print(f"Added student: {name}, Age: {age}")

    def get_students(self):
        with self.Session() as session:
            students = session.query(Student).all()
            for student in students:
                print(f"Student ID: {student.id}, Name: {student.name}, Age: {student.age}")

    def update_student(self, name, new_name=None, new_age=None):
        with self.Session() as session:
            student = session.query(Student).filter(Student.name == name).first()
            if student:
                if new_name:
                    student.name = new_name
                if new_age is not None:
                    student.age = new_age
                session.commit()
                print(f"Updated student: {name}")
            else:
                print("Student not found")

    def delete_student(self, name):
        with self.Session() as session:
            student = session.query(Student).filter(Student.name == name).first()
            if student:
                session.delete(student)
                session.commit()
                print(f"Deleted student: {name}")
            else:
                print("Student not found")
            session.close()

    def add_record(self, student_name, subject, score):
        with self.Session() as session:
            student = session.query(Student).filter(Student.name == student_name).first()
            if student:
                new_record = Record(subject=subject, score=score, student_id=student.id)
                session.add(new_record)
                session.commit()
                print(f"Added record for student: {student_name}, Subject: {subject}, Score: {score}")
            else:
                print("Student not found")

    def get_records(self, student_name):
        with self.Session() as session:
            student = session.query(Student).filter(Student.name == student_name).first()
            if student:
                records = session.query(Record).filter(Record.student_id == student.id).all()
                for record in records:
                    print(f"Record ID: {record.id}, Subject: {record.subject}, Score: {record.score}")
            else:
                print("Student not found")

    def update_record(self, student_name, subject, new_subject=None, new_score=None):
        with self.Session() as session:
            student = session.query(Student).filter(Student.name == student_name).first()
            if student:
                record = session.query(Record).filter(Record.student_id == student.id, Record.subject == subject).first()
                if record:
                    if new_subject:
                        record.subject = new_subject
                    if new_score is not None:
                        record.score = new_score
                    session.commit()
                    print(f"Updated record for student: {student_name}, Subject: {subject}")
                else:
                    print("Record not found")
            else:
                print("Student not found")

    def delete_records(self, student_name, subject):
        with self.Session() as session:
            student = session.query(Student).filter(Student.name == student_name).first()
            if student:
                records = session.query(Record).filter(Record.student_id == student.id, Record.subject == subject).all()
                for record in records:
                    session.delete(record)
                    session.commit()
                    print(f"Deleted record for student: {student_name}, Subject: {subject}")
            else:
                print("Student not found")

    def clear_orphan_records(self):
        with self.Session() as session:
            records = session.query(Record).filter(Record.student_id.not_in(session.query(Student.id))).all()
            for record in records:
                session.delete(record)
                session.commit()
                print(f"Deleted orphan record: {record.id}")


def main():
    db_util = DatabaseUtil()
    # db_util.init_db()

    # db_util.add_student("Alice", 20)
    # db_util.add_student("Bob", 22)

    # db_util.get_students()

    # db_util.update_student("Alice", new_age=21)

    # db_util.add_record("Alice", "Math", 95)
    # db_util.add_record("Alice", "English", 88)
    # db_util.add_record("Bob", "English", 88)

    # db_util.delete_student("Bob")

    # db_util.get_records("Alice")

    # db_util.update_record("Alice", "Math", new_score=98)

    # db_util.delete_records("Alice", "English")

    db_util.clear_orphan_records()

if __name__ == "__main__":
    main()