如何用SQLAlchemy判断数据库为空并仅在空时插入示例数据？

问题描述 #

我想实现一个函数，只有当数据库完全为空时才插入示例数据。目前我通过检查是否存在名为"Ilya"的用户来判断，但如果这个用户被删除，函数会重新插入所有数据。请帮忙修改代码，实现仅在数据库为空时执行插入操作。

当前函数代码 #

from sqlalchemy.orm import sessionmaker


def db_load_example_data(app, db,session):
    customer_name = "Ilya"
    customer_details = session.query(db.Customers).filter(
    db.Customers.name == customer_name).first()
    if customer_details:
        pass
    else:
        #books
        book_01 = db.Books(name="The Teachings Of Don Juan: A Yaqui Way of Knowledge", author="Carlos Castaneda", year_published=1968, type=1)
        book_02 = db.Books(name="Journeys out of the body", author="Robert Monroe", year_published=1971, type=2)
        book_03 = db.Books(name="Lucid Dreaming The power of being aware and awake in your dreams", author="Stephen LaBerge", year_published=1985, type=3)

        # customers
        custormer_00 = db.Customers(name="Ilya", city="TelAviv", age=22)
        custormer_01 = db.Customers(name="Jack", city="RoshHain", age=35)
        custormer_02 = db.Customers(name="Michael", city="Haifa", age=38)
        
        # loans
        loan_01 = db.Loans(customer_id="1", book_id="1", loan_date="2022-10-06", return_date="2022-10-08")

        with app.app_context():
            db.base.metadata.create_all(db.engine)
            Session = sessionmaker(bind=db.engine)  # initialize sessionmaker
            session = Session()  # make Session object
            session.add(custormer_00)
            session.add(custormer_01)
            session.add(custormer_02)
            session.add(book_01)
            session.add(book_02)
            session.add(book_03)
            session.add(loan_01)
            session.commit()

数据库定义 #

# import statements
from sqlalchemy import create_engine
from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.ext.declarative import declarative_base

# create engine
engine = create_engine('sqlite:///library.sqlite', echo=True,
                       connect_args={'check_same_thread': False})
base = declarative_base()

# Books Table


class Books(base):

    __tablename__ = 'Books'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    author = Column(String)
    year_published = Column(String)
    type = Column(Integer)

    def __init__(self, name, author, year_published, type):
        self.name = name
        self.author = author
        self.year_published = year_published
        self.type = int(type)

# Customers Table


class Customers(base):
    __tablename__ = 'Customers'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    city = Column(String)
    age = Column(Integer)

    def __init__(self, name, city, age):
        self.name = name
        self.city = city
        self.age = int(age)

# Loans Table


class Loans(base):
    __tablename__ = 'Loans'
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey(
        "Customers.id", ondelete="CASCADE"), nullable=False)
    book_id = Column(Integer, ForeignKey(
        "Books.id", ondelete="CASCADE"), nullable=False)
    loan_date = Column(String)
    return_date = Column(String)

    def __init__(self, customer_id, book_id, loan_date, return_date):
        self.customer_id = int(customer_id)
        self.book_id = int(book_id)
        self.loan_date = loan_date
        self.return_date = return_date

解决方案 #

核心问题是你依赖特定用户的存在判断数据库状态，这显然不可靠。正确的做法是统计所有业务表的总记录数，确认数据库是否完全为空。同时优化代码中重复创建session的冗余问题，复用传入的session资源。

修改后的函数代码 #

from sqlalchemy.orm import sessionmaker
from sqlalchemy import func


def db_load_example_data(app, db, session):
    # 统计所有业务表的记录数，判断数据库是否为空
    total_customers = session.query(func.count(db.Customers.id)).scalar()
    total_books = session.query(func.count(db.Books.id)).scalar()
    total_loans = session.query(func.count(db.Loans.id)).scalar()

    if total_customers + total_books + total_loans > 0:
        return  # 数据库已有数据，直接返回

    # 数据库为空，插入示例数据
    # books
    book_01 = db.Books(name="The Teachings Of Don Juan: A Yaqui Way of Knowledge", author="Carlos Castaneda", year_published=1968, type=1)
    book_02 = db.Books(name="Journeys out of the body", author="Robert Monroe", year_published=1971, type=2)
    book_03 = db.Books(name="Lucid Dreaming The power of being aware and awake in your dreams", author="Stephen LaBerge", year_published=1985, type=3)

    # customers
    customer_00 = db.Customers(name="Ilya", city="TelAviv", age=22)
    customer_01 = db.Customers(name="Jack", city="RoshHain", age=35)
    customer_02 = db.Customers(name="Michael", city="Haifa", age=38)
    
    # loans
    loan_01 = db.Loans(customer_id=1, book_id=1, loan_date="2022-10-06", return_date="2022-10-08")

    with app.app_context():
        db.base.metadata.create_all(db.engine)
        # 复用传入的session，无需重新创建
        session.add(customer_00)
        session.add(customer_01)
        session.add(customer_02)
        session.add(book_01)
        session.add(book_02)
        session.add(book_03)
        session.add(loan_01)
        session.commit()

额外优化点 #

• 修正了原代码中custormer的拼写错误，改为customer
• 将Loan中的customer_id和book_id直接传入整数，匹配字段类型
• 使用func.count()统计记录数，比first()更高效准确
• 只要任意业务表存在数据就直接返回，避免不必要的操作

内容的提问来源于stack exchange，提问作者Ilya