SQLite+Python实现外键关联表插入与跨表数据更新

我共有3张数据表，各表结构如下所示：

三张表的基础定义：

• 第一张为product表：存储全部商品信息、商品单价以及商品利润（margin）
• 第二张为general_bill（总账单）表：存储客户信息与账单总金额相关数据
• 第三张为账单明细表：是当前需要实现功能的模块，需求逻辑如下：
  • 录入时需要关联product表中对应商品的id、记录商品购买数量
  • prix（单价）直接从product表提取，乘以购买数量得到该条明细的对应金额；margin（利润）计算逻辑一致，取商品单条利润乘以购买数量
  • 明细记录关联的general_bill_id必须和对应总账单的id保持一致（外键关联）
  • 明细数据插入完成后，需要根据同账单id下所有明细的金额、利润汇总值，更新对应general_bill记录的总金额、总利润字段

目前仅实现了最基础的功能，现有代码如下：

import sqlite3
import time, datetime
from datetime import timedelta

class Crud_db:
    def __init__(self, database = 'database.db'):
        self.database = database

    def connect(self):
        self.connection = sqlite3.connect(self.database)
        self.cursor = self.connection.cursor()
        print('connect seccesfully')

    def execute(self, query):
        self.query = query
        self.cursor.execute(self.query)

    def close(self): 
        self.connection.commit()
        self.connection.close()

    def create_tables(self):
        # create all tables

    def insert_new_bill(self):
        self.connect()
        date_f = str(datetime.date.today())
        time_f = str(datetime.datetime.now().time())
        client_name = input('client name: ')
        query01 = 'INSERT INTO general_bill (client_name, date_g, time_g) VALUES (?, ?, ?)'
        data = (client_name,date_f, time_f)
        self.cursor.execute(query01,data) 
        self.close()
        print('added to general bill ..!')

    def add_product(self):
        self.connect()
        product_name = input('product name: ')
        prix = float(input('the price : '))
        royltie = float(input('profit: '))
        product_discreption = input('discreption: ')
        product_query = 'INSERT INTO product (product_name, prix, royltie, product_descreption) VALUES (?,?,?,?)'
        data_set = [product_name,prix,royltie,product_discreption]
        self.cursor.execute(product_query,data_set) 
        self.close()
        print(f'product {product_name} added to database')
        question = input('do you wana add more products ?(yes/no): ')
        if question.lower() == 'yes':
            self.add_product()
        else:
            pass

直接在现有代码基础上补全逻辑即可，核心要做四件事：

• 补全三张表的建表语句，加好外键约束保证数据一致性
• 调整新建账单方法，返回新生成的账单ID，方便后续关联明细
• 新增账单明细录入方法，录入时自动拉取商品的单价、利润，计算单条明细的金额和利润
• 新增总账单自动汇总逻辑，每插入一条明细就重新计算对应总账单的总金额、总利润

修改后可直接运行的完整代码：

import sqlite3
import datetime

class Crud_db:
    def __init__(self, database = 'database.db'):
        self.database = database

    def connect(self):
        self.connection = sqlite3.connect(self.database)
        # 手动开启SQLite外键约束（默认关闭）
        self.connection.execute("PRAGMA foreign_keys = ON")
        self.cursor = self.connection.cursor()

    def close(self): 
        self.connection.commit()
        self.connection.close()

    def create_tables(self):
        self.connect()
        # 商品表
        create_product = """
        CREATE TABLE IF NOT EXISTS product (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_name TEXT NOT NULL,
            prix REAL NOT NULL,
            royltie REAL NOT NULL,
            product_descreption TEXT
        )
        """
        # 总账单表，新增总金额、总利润字段，默认值为0
        create_general_bill = """
        CREATE TABLE IF NOT EXISTS general_bill (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            client_name TEXT NOT NULL,
            date_g TEXT NOT NULL,
            time_g TEXT NOT NULL,
            total_amount REAL DEFAULT 0,
            total_margin REAL DEFAULT 0
        )
        """
        # 账单明细表，加外键关联
        create_bill_detail = """
        CREATE TABLE IF NOT EXISTS bill_detail (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            general_bill_id INTEGER NOT NULL,
            product_id INTEGER NOT NULL,
            quantity INTEGER NOT NULL,
            detail_prix REAL NOT NULL,
            detail_amount REAL NOT NULL,
            detail_margin REAL NOT NULL,
            FOREIGN KEY (general_bill_id) REFERENCES general_bill(id) ON DELETE CASCADE,
            FOREIGN KEY (product_id) REFERENCES product(id)
        )
        """
        self.cursor.execute(create_product)
        self.cursor.execute(create_general_bill)
        self.cursor.execute(create_bill_detail)
        self.close()
        print("所有表创建完成")

    def insert_new_bill(self):
        self.connect()
        date_f = str(datetime.date.today())
        time_f = str(datetime.datetime.now().time())
        client_name = input('请输入客户名称: ')
        query01 = 'INSERT INTO general_bill (client_name, date_g, time_g) VALUES (?, ?, ?)'
        data = (client_name,date_f, time_f)
        self.cursor.execute(query01,data)
        # 返回新建账单的自增ID，用于后续录入明细
        new_bill_id = self.cursor.lastrowid
        self.close()
        print(f'新账单创建成功，账单ID为：{new_bill_id}')
        return new_bill_id

    def add_product(self):
        while True:
            self.connect()
            product_name = input('请输入商品名称: ')
            prix = float(input('请输入商品单价: '))
            royltie = float(input('请输入商品单利润: '))
            product_discreption = input('请输入商品描述: ')
            product_query = 'INSERT INTO product (product_name, prix, royltie, product_descreption) VALUES (?,?,?,?)'
            data_set = [product_name,prix,royltie,product_discreption]
            self.cursor.execute(product_query,data_set)
            new_prod_id = self.cursor.lastrowid
            self.close()
            print(f'商品 {product_name} 添加成功，商品ID为：{new_prod_id}')
            question = input('是否继续添加商品？(yes/no): ').lower()
            if question != 'yes':
                break

    def _update_bill_total(self, bill_id):
        """内部方法：汇总指定账单下所有明细，更新总账单的金额和利润"""
        sum_query = """
        SELECT SUM(detail_amount), SUM(detail_margin) FROM bill_detail 
        WHERE general_bill_id = ?
        """
        self.cursor.execute(sum_query, (bill_id,))
        total_amount, total_margin = self.cursor.fetchone()
        # 处理账单无明细时的空值
        total_amount = total_amount if total_amount else 0
        total_margin = total_margin if total_margin else 0
        update_query = """
        UPDATE general_bill SET total_amount = ?, total_margin = ? WHERE id = ?
        """
        self.cursor.execute(update_query, (total_amount, total_margin, bill_id))

    def add_bill_detail(self, bill_id):
        """给指定账单录入商品明细"""
        self.connect()
        # 先校验账单ID是否存在
        self.cursor.execute("SELECT id FROM general_bill WHERE id = ?", (bill_id,))
        if not self.cursor.fetchone():
            self.close()
            print("对应账单ID不存在，请检查后重试")
            return
        
        while True:
            try:
                product_id = int(input("请输入购买商品ID: "))
                quantity = int(input("请输入购买数量: "))
                # 查询对应商品的单价和利润
                self.cursor.execute("SELECT prix, royltie FROM product WHERE id = ?", (product_id,))
                prod_data = self.cursor.fetchone()
                if not prod_data:
                    print("商品ID不存在，请重新输入")
                    continue
                prod_prix, prod_margin = prod_data
                # 计算单条明细的金额和利润
                detail_amount = prod_prix * quantity
                detail_margin = prod_margin * quantity
                # 插入明细记录
                insert_detail = """
                INSERT INTO bill_detail (general_bill_id, product_id, quantity, detail_prix, detail_amount, detail_margin)
                VALUES (?, ?, ?, ?, ?, ?)
                """
                self.cursor.execute(insert_detail, (bill_id, product_id, quantity, prod_prix, detail_amount, detail_margin))
                print(f"该条明细添加成功，明细金额：{detail_amount:.2f}，明细利润：{detail_margin:.2f}")
                # 自动更新总账单汇总数据
                self._update_bill_total(bill_id)
                self.connection.commit()

                add_more = input("是否继续添加该账单下的其他商品？(yes/no): ").lower()
                if add_more != 'yes':
                    break
            except ValueError:
                print("输入格式错误，请输入数字类型的ID和数量")
        self.close()
        print("账单明细录入完成，总账单金额和利润已自动更新")

使用示例：

if __name__ == "__main__":
    db = Crud_db()
    # 第一次运行先执行建表
    # db.create_tables()
    # 提前录入商品基础信息
    # db.add_product()
    # 创建新账单，拿到返回的账单ID
    # current_bill_id = db.insert_new_bill()
    # 给对应账单录入明细
    # db.add_bill_detail(current_bill_id)

注意事项：

• 所有SQL操作都用参数化传入值，不要直接拼接字符串，避免SQL注入风险
• 汇总计算直接在数据库层用SUM聚合完成，不需要拉取全量明细到Python内存计算，效率更高
• 每次插入明细后立刻触发总账单更新，保证总账单和明细数据始终一致

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