27 lines
831 B
Python
27 lines
831 B
Python
|
"""16进制自增计数器 — 基于 PostgreSQL Sequence，生成 16 位 HEX 唯一 ID"""
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
SEQUENCE_NAME = "product_hex_counter"
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def ensure_sequence(db: AsyncSession) -> None:
|
|||
|
|
"""确保 counter sequence 存在（幂等）"""
|
|||
|
|
await db.execute(
|
|||
|
|
text(f"CREATE SEQUENCE IF NOT EXISTS {SEQUENCE_NAME} START 1;")
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def next_hex_id(db: AsyncSession, length: int = 16) -> str:
|
|||
|
|
"""
|
|||
|
|
生成下一个 hex ID。
|
|||
|
|
|
|||
|
|
示例: 1 → "0000000000000001"
|
|||
|
|
15 → "000000000000000F"
|
|||
|
|
16 → "0000000000000010"
|
|||
|
|
255 → "00000000000000FF"
|
|||
|
|
"""
|
|||
|
|
result = await db.execute(text(f"SELECT nextval('{SEQUENCE_NAME}');"))
|
|||
|
|
counter: int = result.scalar_one()
|
|||
|
|
return format(counter, f"0{length}X")
|