元类

元类注册子类:

import json

registry = {}


def register_class(target_class):
    registry[target_class.__name__] = target_class


class Meta(type):
    def __new__(cls, *args, **kwargs):
        """
        1. cls 为当前准备创建对象的类
        2. args 中第一个参数为类的名字
        3. args 中第二个参数为类继承的父类集合
        4. args 中第三个参数为类的方法集合
        :param args:
        :param kwargs:
        :return:
        """
        cls = type.__new__(cls, *args, **kwargs)
        register_class(cls)
        return cls


class Serializable(object, metaclass=Meta):
    def __init__(self, *args):
        self.args = args

    def serialize(self):
        return json.dumps(
            {
                "class": self.__class__.__name__,
                "args": self.args
            }
        )


def deserialize(data):
    params = json.loads(data)
    name = params["class"]
    target_class = registry[name]
    return target_class(*params["args"])


class Point2D(Serializable):
    def __init__(self, x, y):
        super().__init__(x, y)
        self.x = x
        self.y = y

    def __repr__(self):
        return f"Point2D:{self.x, self.y}"


if __name__ == "__main__":
    point = Point2D(5, 3)
    serializePoint2D = (point.serialize())
    print(point)
    print(serializePoint2D)
    obj = deserialize(serializePoint2D)
    print(obj)

元类注解类的属性:

class Field(object):
    def __init__(self, name, internal_name):
        self.name = name
        self.internal_name = internal_name


class Meta(type):
    def __new__(meta, name, bases, class_dict):
        for key, value in class_dict.items():
            if isinstance(value, Field):
                value.name = key
                value.internal_name = "_" + key
        cls = type.__new__(meta, name, bases, class_dict)
        return cls


class DataBase(object, metaclass=Meta):
    pass


# apply
class UserTable(DataBase):
    name = Field(name="user_name", internal_name="user_name")
    age = Field(name="user_age", internal_name="user_gae")


if __name__ == "__main__":
    user = UserTable()
    print(user)
    print(user.name)
    user.name = "zhangsan"
    print(user.__dict__)

元类强制子类实现某方法

class Meta(type):
    def __new__(cls, name, bases, class_dict, **kwargs):
        if name != "Base" and "foo" not in class_dict:
            raise TypeError("需要重写 foo 方法!")
        cls = type.__new__(cls, name, bases, class_dict, **kwargs)
        return cls


class Base(object, metaclass=Meta):
    # 继承 Base 的子类需要重写此方法
    def foo(self):
        return self.foo()


class Apply(Base):
    pass

    # def foo(self):
    #     pass


if __name__ == "__main__":
    apply = Apply()
    print(apply)

统计一个类的所有子类

class Meta(type):
    def __init__(cls, name, bases, namespace, **kwargs):
        super().__init__(name, bases, namespace, **kwargs)
        if not hasattr(cls, 'registory'):
            # this is the base class
            cls.registory = {}
        else:
            # this is the subclass
            cls.registory[name.lower()] = cls


class Fruit(object, metaclass=Meta):
    pass


class Apple(Fruit):
    pass


class Orange(Fruit):
    pass


print(Fruit.registory)

 上一篇
K8s 应用发布方案 - 金丝雀 K8s 应用发布方案 - 金丝雀
K8s 应用发布方案项目背景假设目标场景为需要发布名称为 DataAPI 的应用,此应用在 k8s 集群上共有 5 个 Pod,均为无状态的应用,分布在 2 个节点上。假设目前生产使用版本为 v1,待发布版本为 v2。 准备镜像已推送镜像
2019-06-28
下一篇 
StatefulSet StatefulSet
StatefulSet 用于管理有状态 Pod 的控制器,例如 Redis,ZK 等,与 Deployment 相反。在 K8s 上定义和管理有状态的应用是比较困难的,每个有状态的服务都有自己的特性,你要考虑它们出现问题之后的重启方式,数据
2019-06-22
  目录