shanxia2.0/backend/app/services/db/inventory_id.py

"""出入库单 id 生成 — 租户内纯数字自增，并发安全。"""
from __future__ import annotations

from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.db_models import DbInboundOutbound, DbSequence

SEQ_NAME_INBOUND_OUTBOUND = "inbound_outbound"
_NUMERIC_ID_PATTERN = r"^[0-9]+$"


def max_numeric_inbound_outbound_id(db: Session, tenant_no: str) -> int:
    """当前租户下纯数字 id 的最大值（不含 UUID 等非数字主键）。"""
    result = db.execute(
        text(
            """
            SELECT COALESCE(MAX(CAST(id AS UNSIGNED)), 0) AS max_id
            FROM db_inbound_outbound
            WHERE tenant_no = :tenant_no
              AND del_flag = 0
              AND id REGEXP :pattern
            """
        ),
        {"tenant_no": tenant_no, "pattern": _NUMERIC_ID_PATTERN},
    ).scalar()
    return int(result or 0)


def _ensure_sequence_row(db: Session, tenant_no: str) -> None:
    exists = (
        db.query(DbSequence.id)
        .filter(
            DbSequence.tenant_no == tenant_no,
            DbSequence.seq_name == SEQ_NAME_INBOUND_OUTBOUND,
        )
        .first()
    )
    if exists:
        return

    initial_next = max_numeric_inbound_outbound_id(db, tenant_no) + 1
    db.add(
        DbSequence(
            id=generate_uuid(),
            tenant_no=tenant_no,
            seq_name=SEQ_NAME_INBOUND_OUTBOUND,
            next_val=initial_next,
        )
    )
    db.flush()


def _allocate_inbound_outbound_id(db: Session, tenant_no: str) -> str:
    _ensure_sequence_row(db, tenant_no)
    row = (
        db.query(DbSequence)
        .filter(
            DbSequence.tenant_no == tenant_no,
            DbSequence.seq_name == SEQ_NAME_INBOUND_OUTBOUND,
        )
        .with_for_update()
        .one()
    )
    assigned = int(row.next_val)
    row.next_val = assigned + 1
    db.flush()
    return str(assigned)


def next_inbound_outbound_id(db: Session, tenant_no: str) -> str:
    """
    分配下一条出入库单 id（纯数字字符串）。
    使用 db_sequence 行锁 + next_val 递增，避免并发下重复 id。
    """
    last_error: IntegrityError | None = None
    for _ in range(3):
        try:
            return _allocate_inbound_outbound_id(db, tenant_no)
        except IntegrityError as exc:
            db.rollback()
            last_error = exc
    if last_error is not None:
        raise last_error
    return _allocate_inbound_outbound_id(db, tenant_no)