如何在Python中以值而非类型作为泛型参数实现多纪元Snowflake？

你想要用Snowflake[DISCORD_EPOCH]这种类似泛型的语法，创建绑定不同纪元的雪花算法类——Python本身的泛型系统主要针对类型参数，但可以通过重载类的__class_getitem__方法实现这个需求，因为Cls[X]的语法本质是调用Cls.__class_getitem__(X)。

下面提供两种实用的实现方式：

方法一：利用类方法__class_getitem__实现类绑定 #

先定义纪元常量：

DISCORD_EPOCH = 1420070400000  # Discord官方纪元（毫秒级）
MY_EPOCH = 1609459200000      # 自定义纪元（比如2021年元旦）

然后实现Snowflake类，通过__class_getitem__返回绑定特定纪元的子类：

import time

class Snowflake:
    # 缓存已创建的绑定类，避免重复生成
    _bound_cache = {}

    def __init__(self, epoch):
        self.epoch = epoch

    @classmethod
    def __class_getitem__(cls, epoch):
        # 先检查缓存，存在则直接返回
        if epoch not in cls._bound_cache:
            # 创建绑定纪元的子类
            class BoundSnowflake(cls):
                _FIXED_EPOCH = epoch

                def __init__(self):
                    super().__init__(self._FIXED_EPOCH)

            # 给子类设置有意义的名字，方便调试
            BoundSnowflake.__name__ = f"{cls.__name__}[{epoch}]"
            cls._bound_cache[epoch] = BoundSnowflake
        return cls._bound_cache[epoch]

    # 简化的雪花ID生成方法
    def generate(self, worker_id: int, sequence: int) -> int:
        current_ts = int(time.time() * 1000)
        timestamp = current_ts - self.epoch
        return (timestamp << 22) | (worker_id << 12) | sequence

    # 雪花ID解析方法
    def parse(self, snowflake_id: int) -> dict:
        timestamp = (snowflake_id >> 22) + self.epoch
        worker_id = (snowflake_id >> 12) & 0x3FF
        sequence = snowflake_id & 0xFFF
        return {
            "timestamp": timestamp,
            "worker_id": worker_id,
            "sequence": sequence
        }

使用方式完全符合你的需求：

# 创建绑定不同纪元的类
DiscordSnowflake = Snowflake[DISCORD_EPOCH]
MySnowflake = Snowflake[MY_EPOCH]

# 实例化并使用
discord_snow = DiscordSnowflake()
print(discord_snow.generate(worker_id=1, sequence=0))
print(discord_snow.parse(1234567890123456789))

my_snow = MySnowflake()
print(my_snow.generate(worker_id=2, sequence=1))

方法二：用元类实现更简洁的逻辑 #

如果想把绑定逻辑和业务逻辑分离，可以用元类来处理：

import time

class SnowflakeMeta(type):
    _bound_cache = {}

    def __getitem__(cls, epoch):
        if epoch not in cls._bound_cache:
            class BoundSnowflake(metaclass=SnowflakeMeta):
                _FIXED_EPOCH = epoch

                def generate(self, worker_id: int, sequence: int) -> int:
                    current_ts = int(time.time() * 1000)
                    timestamp = current_ts - self._FIXED_EPOCH
                    return (timestamp << 22) | (worker_id << 12) | sequence

                def parse(self, snowflake_id: int) -> dict:
                    timestamp = (snowflake_id >> 22) + self._FIXED_EPOCH
                    worker_id = (snowflake_id >> 12) & 0x3FF
                    sequence = snowflake_id & 0xFFF
                    return {
                        "timestamp": timestamp,
                        "worker_id": worker_id,
                        "sequence": sequence
                    }

            BoundSnowflake.__name__ = f"{cls.__name__}[{epoch}]"
            cls._bound_cache[epoch] = BoundSnowflake
        return cls._bound_cache[epoch]

class Snowflake(metaclass=SnowflakeMeta):
    pass

使用方式和之前完全一致：

DiscordSnowflake = Snowflake[DISCORD_EPOCH]
MySnowflake = Snowflake[MY_EPOCH]

核心原理 #

两种方法的本质都是：当你调用Snowflake[epoch]时，Python会触发对应的__class_getitem__（类方法）或元类的__getitem__方法，返回一个预先绑定了指定纪元的子类。缓存机制则避免了重复创建相同纪元的类，提升性能。

内容的提问来源于stack exchange，提问作者anton