Compare commits

..

2 Commits

2 changed files with 33 additions and 21 deletions
+16 -19
View File
@@ -27,7 +27,7 @@ def encode_base62(num: int, length: int = None) -> str:
return res return res
def encode_tablename_to_prefix(tablename: str, length: int = 3) -> str: def encode_tablename_to_prefix(tablename: str, length: int) -> str:
""" """
将表名 encode 成固定长度的前缀,使用 MD5 拿到表名的唯一哈希值,转为整数后用 Base62 截取前 N 位 将表名 encode 成固定长度的前缀,使用 MD5 拿到表名的唯一哈希值,转为整数后用 Base62 截取前 N 位
""" """
@@ -39,18 +39,17 @@ def encode_tablename_to_prefix(tablename: str, length: int = 3) -> str:
num = int(hex_digest, 16) num = int(hex_digest, 16)
# 转为 Base62 字符串 # 转为 Base62 字符串
b62_str = encode_base62(num, length=length) return encode_base62(num, length=length)
# 截取固定长度(例如 3 位)作为该表的专属代码
return b62_str[:length]
def gen_convex_payload() -> str: def gen_convex_payload(length: int) -> str:
"""生成 128位 ULID 转 Base62 的趋势递增 Payload""" """生成固定长度的 ULID Payload"""
u = ulid.new() u = ulid.new()
return encode_base62(u.int) return encode_base62(u.int, length=length)
# --- 注入器类 ---
class AutoID: class AutoID:
""" """
全自动 Convex 风格 ID 注入器 全自动 Convex 风格 ID 注入器
@@ -61,6 +60,11 @@ class AutoID:
_id_prefix_len = 3 # 前缀长度 _id_prefix_len = 3 # 前缀长度
_id_payload_len = 22 # 主体长度(ULID 转 Base62 最大约 22 位) _id_payload_len = 22 # 主体长度(ULID 转 Base62 最大约 22 位)
@classmethod
def id_length(cls) -> int:
"""返回当前类 ID 的总长度"""
return cls._id_prefix_len + cls._id_payload_len
@declared_attr @declared_attr
def id(cls): def id(cls):
# 运行时动态获取子类的 __tablename__ # 运行时动态获取子类的 __tablename__
@@ -68,18 +72,11 @@ class AutoID:
# 获取子类配置的长度 # 获取子类配置的长度
p_len = cls._id_prefix_len p_len = cls._id_prefix_len
s_len = cls._id_payload_len s_len = cls._id_payload_len
prefix = encode_tablename_to_prefix(tablename, length=p_len)
# 将表名编码为固定前缀(例如 "users" -> "7X1"
hasher = hashlib.md5(tablename.encode("utf-8"))
table_num = int(hasher.hexdigest(), 16)
table_prefix = encode_base62(table_num, length=p_len)
# 核心工厂函数:拼接【编码后的表前缀】和【递增Payload】 # 核心工厂函数:拼接【编码后的表前缀】和【递增Payload】
def id_factory(): def id_factory():
# 生成固定长度的 Payload payload = gen_convex_payload(length=s_len)
u = ulid.new() return f"{prefix}{payload}"
payload = encode_base62(u.int, length=s_len)
# 直接拼接,不带下划线
return f"{table_prefix}{payload}"
return Column(String(p_len + s_len), primary_key=True, default=id_factory) return Column(String(cls.id_length()), primary_key=True, default=id_factory)
+17 -2
View File
@@ -1,13 +1,28 @@
from typing import TYPE_CHECKING
from sqlalchemy import Column, DateTime, String, ForeignKey, UniqueConstraint from sqlalchemy import Column, DateTime, String, ForeignKey, UniqueConstraint
from sqlalchemy.sql import func from sqlalchemy.sql import func
from .base import AutoID, Base from .base import AutoID, Base
# 只有在静态检查时才导入
# 运行阶段这段代码会被跳过,彻底避免循环导入
# 不确定是否用得上,但感觉这是个好习惯,尤其在模型之间有外键关系时
if TYPE_CHECKING:
from .user import User
class Follow(Base, AutoID): class Follow(Base, AutoID):
"""用户关注关系表""" """用户关注关系表"""
__tablename__ = "follows" __tablename__ = "follows"
follower_id = Column(String(32), ForeignKey("users.id", ondelete="CASCADE"), index=True) follower_id = Column(
followee_id = Column(String(32), ForeignKey("users.id", ondelete="CASCADE"), index=True) String(User.id_length()),
ForeignKey("users.id", ondelete="CASCADE"),
index=True
)
followee_id = Column(
String(User.id_length()),
ForeignKey("users.id", ondelete="CASCADE"),
index=True
)
created_at = Column(DateTime(timezone=True), server_default=func.now()) created_at = Column(DateTime(timezone=True), server_default=func.now())