Python数据库操作模式

什么是ORM？
对象关系映射（Object-Relational Mapping，ORM）是一种编程技术，它允许开发者使用面向对象的方式来操作数据库。ORM库将数据库表映射为Python类，将表中的行映射为对象，将字段映射为对象的属性，将SQL查询映射为方法调用。这样，开发者可以使用Python代码来操作数据库，而不需要直接编写SQL语句。
为什么使用ORM？

提高开发效率：ORM可以减少编写SQL语句的时间，使开发者可以专注于业务逻辑。
提高代码可维护性：ORM代码通常更易于阅读和维护。
跨数据库支持：大多数ORM库支持多种数据库，可以轻松切换数据库而无需修改代码。
事务管理：ORM通常提供事务管理功能，简化事务处理。

常见的Python ORM库

SQLAlchemy：功能强大且灵活，支持多种数据库，提供了ORM和SQL表达式语言。
Django ORM：Django框架自带的ORM，简单易用，适合快速开发。
Peewee：轻量级ORM，易于学习和使用。
Tortoise ORM：支持异步操作的ORM，适合异步应用。

使用SQLAlchemy进行ORM操作

from contextlib import contextmanager

from sqlalchemy import create_engine, and_, update, select
from sqlalchemy.orm import sessionmaker, registry, relationship
from sqlalchemy import Column, Integer, String, Float, ForeignKey

# 创建一个映射注册表
mapperregistry = registry()

Base = mapperregistry.generatebase()


class Invoice(Base):
    tablename = 'invoice'
    id = Column(Integer, primarykey=True)
    ordersn = Column(String(50))
    # 其他字段...


class InvoiceInfo(Base):
    tablename = 'invoiceinfo'
    id = Column(Integer, primarykey=True)
    invoiceid = Column(Integer, ForeignKey('invoice.id'))
    recid = Column(Integer)
    name = Column(String(100))
    units = Column(String(50))
    class = Column('class', String(100))  # 修改为 class
    classcode = Column(String(50))
    taxrate = Column(Float)
    invoice = relationship("Invoice", backpopulates="invoiceinfo")


Invoice.invoiceinfo = relationship("InvoiceInfo", orderby=InvoiceInfo.id, backpopulates="invoice")


class MilitaryOrderGoods(Base):
    tablename = 'militaryordergoods'
    id = Column(Integer, primarykey=True)
    recid = Column(Integer)
    goodsid = Column(Integer)
    # 其他字段...


class Goods(Base):
    tablename = 'goods'
    id = Column(Integer, primarykey=True)
    goodsid = Column(Integer)
    invoicename = Column(String(100))
    unit = Column(String(50))
    classname = Column(String(100))
    classcode = Column(String(50))
    # 其他字段...


class TaxCategories(Base):
    tablename = 'taxcategories'
    classid = Column(Integer, primarykey=True)
    classname = Column(String(100))
    classcode = Column(String(50))
    taxrate = Column(Float)
    flag = Column(Integer)  # 假设 flag 是一个整数字段
    # 其他字段...


@contextmanager
def getdbsession():
    engine = createengine(
        'mysql+pymysql://test?charset=utf8')
    Session = sessionmaker(bind=engine)
    session = Session()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()


def updateinvoicefields(ordersn, name=None, units=None, classname=None, taxrate=None):
    with getdbsession() as session:
        # 构造更新条件
        conditions = [Invoice.ordersn == ordersn]

        # 打印条件过滤结果
        print(f"Conditions: {conditions}")
        subquerytax = select(TaxCategories.classname, TaxCategories.classcode).where(
            TaxCategories.classname == classname,
            TaxCategories.flag == 1
        ).limit(1)
        result = session.execute(subquerytax).fetchone()

        if result:
            classnamevalue = result[0]
            classcodevalue = result[1]
            print('result-------------', result)
        else:
            print('No matching record found for classname and classcode.')
            return 0

        # 构造连接
        subquery = (
            select(
                InvoiceInfo.id,
                Goods.goodsid,
                Goods.invoicename,
                Goods.unit
            )
                .join(Invoice, Invoice.id == InvoiceInfo.invoiceid)
                .join(MilitaryOrderGoods, InvoiceInfo.recid == MilitaryOrderGoods.recid)
                .join(Goods, MilitaryOrderGoods.goodsid == Goods.goodsid)
                .filter(and(*conditions))
        )

        # 获取子查询结果
        results = session.execute(subquery).fetchall()
        print('results-------------', results)
        if not results:
            print("No records to update.")
            return 0
            # 构造更新字典
        updatevalues = {
            'class': classnamevalue,
            'name': name,
            'classcode': classcodevalue,
            'taxrate': taxrate
        }
        # 只有在 units 不为空时才添加到更新字段中
        if units and units.strip():
            updatevalues['units'] = units
        # 构造更新语句
        stmtinvoiceinfo = (
            update(InvoiceInfo)
                .where(InvoiceInfo.id.in([row[0] for row in results]))
                .values(**updatevalues
                        )
        )

        stmtgoods = (
            update(Goods)
                .where(Goods.goodsid.in([row[1] for row in results]))
                .values(
                invoicename=name,
                unit=units
            )
        )

        # 执行更新操作
        updatedcountinvoiceinfo = session.execute(stmtinvoiceinfo).rowcount
        updatedcountgoods = session.execute(stmtgoods).rowcount

        # 打印更新的记录数
        print(f"Updated Records Count in InvoiceInfo: {updatedcountinvoiceinfo}")
        print(f"Updated Records Count in Goods: {updatedcountgoods}")

        # 返回更新的行数
        return updatedcountinvoiceinfo + updatedcountgoods


# 调用示例
print(updateinvoicefields("20220110134064", name="测试事务回滚", units='', classname="日用杂品", taxrate=13))