数据契约（Data Contract）SLA 制定与生产者

"""数据契约 SLA 管理演示- 模拟生产者发布数据、消费者订阅契约、自动校验与违约处置全流程"""import datetimeimport jsonfrom typing import Dict, List, Optional  class DataContract:    """数据契约类：保存 SLA 条款并支持自动校验"""     def __init__(self, name: str):        self.name = name        self.schema: Dict[str, str] = {}        self.quality_rate: float = 99.0        self.freshness_seconds: int = 3600        self.breach_count: int = 0        self.status: str = "active"        self.version: int = 1     def set_sla(self, schema: Dict[str, str],                quality: float, freshness: int) -> None:        """设置契约 SLA 条款"""        self.schema = schema        self.quality_rate = quality        self.freshness_seconds = freshness     def check_schema(self, data_row: Dict) -> bool:        """检查单行数据是否符合 Schema 定义"""        for field, ftype in self.schema.items():            if field not in data_row:                return False            if ftype == "int" and not isinstance(data_row[field], int):                return False            if ftype == "str" and not isinstance(data_row[field], str):                return False        return True     def check_freshness(self, row_time: datetime.datetime) -> bool:        """检查数据是否在新鲜度窗口内"""        delay = (datetime.datetime.now() - row_time).total_seconds()        return delay <= self.freshness_seconds     def record_breach(self) -> None:        """记录违约并升级契约状态"""        self.breach_count += 1        if self.breach_count > 3:            self.status = "suspended"     def to_json(self) -> str:        """输出契约 JSON 描述"""        return json.dumps({            "name": self.name,            "version": self.version,            "schema": self.schema,            "quality": self.quality_rate,            "freshness": self.freshness_seconds,            "status": self.status        }, ensure_ascii=False)  class Producer:    """数据生产者：按契约发布数据"""     def __init__(self, name: str):        self.name = name        self.published_count: int = 0     def publish(self, contract: DataContract) -> Optional[Dict]:        """按契约发布一行模拟数据"""        if contract.status != "active":            print("契约已暂停，停止发布")            return None        row = {            "order_id": self.published_count + 1,            "user_name": "user_" + str(self.published_count),            "amount": self.published_count * 10        }        self.published_count += 1        return row  class Consumer:    """数据消费者：订阅契约并验证数据"""     def __init__(self, name: str):        self.name = name        self.accepted_count: int = 0        self.rejected_count: int = 0     def consume(self, contract: DataContract,                row: Dict, row_time: datetime.datetime) -> None:        """消费数据前自动执行契约校验"""        if not contract.check_schema(row):            contract.record_breach()            self.rejected_count += 1            print(f"[消费者] 拒绝：Schema 不符 {row}")            return        if not contract.check_freshness(row_time):            contract.record_breach()            self.rejected_count += 1            print(f"[消费者] 拒绝：数据过期 {row_time}")            return        self.accepted_count += 1        print(f"[消费者] 接受：{row}")  def demo() -> None:    """运行完整演示"""    contract = DataContract("订单数据契约")    contract.set_sla(        schema={"order_id": "int", "user_name": "str", "amount": "int"},        quality=99.5,        freshness=300    )     producer = Producer("订单系统")    consumer = Consumer("分析平台")     print("=== 数据契约 SLA 演示 ===")    print(contract.to_json())    print()     for i in range(5):        print(f"--- 第 {i + 1} 轮 ---")        row = producer.publish(contract)        if row is None:            break        row_time = datetime.datetime.now() - datetime.timedelta(            seconds=60 * i)        consumer.consume(contract, row, row_time)     print()    print(f"接受数: {consumer.accepted_count}, "          f"拒绝数: {consumer.rejected_count}, "          f"违约数: {contract.breach_count}")  if __name__ == "__main__":    demo()