Pine
c2f4c13ab3
- users.phone/wx_openid/wx_mini_openid/wx_unionid 各自唯一(非空部分唯一索引) - 创建账号必生成ID(create用new_id)；find_by_phone/find_by_wx 唯一校验 - cleanup_accounts.py 清理旧accounts已被users接管的重复账号；seed不再重复造账号、迁移后删旧行
64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
"""账号唯一性：users 的 登录名/手机号/微信 openid 各自唯一（非空值部分唯一）
|
||
|
||
账号唯一性约定（用户 ID / 登录名 / 手机号 / 微信身份 均为独立逻辑，各自唯一）：
|
||
- id 主键天然唯一（创建时由服务端生成，必存在）
|
||
- username 登录名 — 已唯一（见 0001 models unique=True）
|
||
- phone 手机号 — 本次加「非空部分唯一」索引（空串=未绑定，允许多个，互不冲突）
|
||
- wx_openid / wx_mini_openid / wx_unionid — 微信身份同样非空部分唯一，避免一个微信身份被搭到两个账号
|
||
|
||
SQLite 不支持带 WHERE 的唯一约束，但支持部分唯一索引：
|
||
CREATE UNIQUE INDEX ... ON users (phone) WHERE phone != ''
|
||
同时 DB 层保留普通索引便于按 openid 查询。
|
||
|
||
Revision ID: 0011_unique_user_identifiers
|
||
Revises: 0010_user_opc_profile
|
||
Create Date: 2026-08-25
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sqlalchemy as sa
|
||
from alembic import op
|
||
|
||
revision = "0011_unique_user_identifiers"
|
||
down_revision = "0010_user_opc_profile"
|
||
branch_labels = None
|
||
depends_on = None
|
||
|
||
|
||
def upgrade() -> None:
|
||
bind = op.get_bind()
|
||
dialect = bind.dialect.name
|
||
# 非空值部分唯一索引（SQLite 专属语法；其它方言退化为普通索引并另行约束）
|
||
# 迁移前若有重复非空值会失败——请先用 scripts/db/cleanup_accounts.py 清理。
|
||
unique_cols = ["phone", "wx_openid", "wx_mini_openid", "wx_unionid"]
|
||
for col in unique_cols:
|
||
if dialect != "sqlite":
|
||
continue
|
||
op.create_index(
|
||
f"ux_users_{col}_nonempty",
|
||
"users",
|
||
[col],
|
||
unique=True,
|
||
sqlite_where=sa.text(f"{col} != ''"),
|
||
)
|
||
# 数据库连接(非空内存)唯一：postgres 用部分唯一索引、mysql 用普通（另行靠应用层校验）
|
||
if dialect == "postgresql":
|
||
for col in unique_cols:
|
||
op.create_index(
|
||
f"ux_users_{col}_nonempty",
|
||
"users",
|
||
[col],
|
||
unique=True,
|
||
postgresql_where=sa.text(f"{col} IS NOT NULL"),
|
||
)
|
||
|
||
|
||
def downgrade() -> None:
|
||
bind = op.get_bind()
|
||
dialect = bind.dialect.name
|
||
unique_cols = ["phone", "wx_openid", "wx_mini_openid", "wx_unionid"]
|
||
for col in unique_cols:
|
||
if dialect not in ("sqlite", "postgresql"):
|
||
continue
|
||
op.drop_index(f"ux_users_{col}_nonempty", table_name="users")
|