Python库 | metaphor-connectors-0.1.5.tar.gz

# This code parses date/times, so please # # pip install python-dateutil # # To use this code, make sure you # # import json # # and then, to convert JSON from a string, do # # result = metadata_change_event_from_dict(json.loads(json_string)) from enum import Enum from dataclasses import dataclass from typing import Optional, Any, List, TypeVar, Type, cast, Callable from datetime import datetime import dateutil.parser T = TypeVar("T") EnumT = TypeVar("EnumT", bound=Enum) def from_str(x: Any) -> str: assert isinstance(x, str) return x def from_none(x: Any) -> Any: assert x is None return x def from_union(fs, x): for f in fs: try: return f(x) except: pass assert False def from_datetime(x: Any) -> datetime: return dateutil.parser.parse(x) def from_float(x: Any) -> float: assert isinstance(x, (float, int)) and not isinstance(x, bool) return float(x) def to_float(x: Any) -> float: assert isinstance(x, float) return x def from_bool(x: Any) -> bool: assert isinstance(x, bool) return x def to_class(c: Type[T], x: Any) -> dict: assert isinstance(x, c) return cast(Any, x).to_dict() def to_enum(c: Type[EnumT], x: Any) -> EnumT: assert isinstance(x, c) return x.value def from_list(f: Callable[[Any], T], x: Any) -> List[T]: assert isinstance(x, list) return [f(y) for y in x] class AspectType(Enum): DATASET_COMMENT = "DATASET_COMMENT" DATASET_DOCUMENTATION = "DATASET_DOCUMENTATION" DATASET_DOWNSTREAM = "DATASET_DOWNSTREAM" DATASET_SCHEMA = "DATASET_SCHEMA" DATASET_STATISTICS = "DATASET_STATISTICS" DATASET_UPSTREAM = "DATASET_UPSTREAM" GROUP_INFO = "GROUP_INFO" OWNERSHIP = "OWNERSHIP" PERSON_EDITABLE_INFO = "PERSON_EDITABLE_INFO" PROPERTIES = "PROPERTIES" @dataclass class AuditStamp: actor: Optional[str] = None time: Optional[datetime] = None @staticmethod def from_dict(obj: Any) -> 'AuditStamp': assert isinstance(obj, dict) actor = from_union([from_str, from_none], obj.get("actor")) time = from_union([from_datetime, from_none], obj.get("time")) return AuditStamp(actor, time) def to_dict(self) -> dict: result: dict = {} result["actor"] = from_union([from_str, from_none], self.actor) result["time"] = from_union([lambda x: x.isoformat(), from_none], self.time) return result @dataclass class ObjectID: """A class representation of the BSON ObjectId type.""" """The generation time of this ObjectId instance""" generation_time: Optional[float] = None @staticmethod def from_dict(obj: Any) -> 'ObjectID': assert isinstance(obj, dict) generation_time = from_union([from_float, from_none], obj.get("generationTime")) return ObjectID(generation_time) def to_dict(self) -> dict: result: dict = {} result["generationTime"] = from_union([to_float, from_none], self.generation_time) return result @dataclass class DatasetComment: id: Optional[ObjectID] = None aspect_type: Optional[AspectType] = None content: Optional[str] = None created: Optional[AuditStamp] = None created_at: Optional[datetime] = None deleted: Optional[AuditStamp] = None entity_id: Optional[str] = None field: Optional[str] = None dataset_comment_id: Optional[str] = None latest: Optional[bool] = None parent: Optional[str] = None @staticmethod def from_dict(obj: Any) -> 'DatasetComment': assert isinstance(obj, dict) id = from_union([ObjectID.from_dict, from_none], obj.get("_id")) aspect_type = from_union([AspectType, from_none], obj.get("aspectType")) content = from_union([from_str, from_none], obj.get("content")) created = from_union([AuditStamp.from_dict, from_none], obj.get("created")) created_at = from_union([from_datetime, from_none], obj.get("createdAt")) deleted = from_union([AuditStamp.from_dict, from_none], obj.get("deleted")) entity_id = from_union([from_str, from_none], obj.get("entityId")) field = from_union([from_str, from_none], obj.get("field")) dataset_comment_id = from_union([from_str, from_none], obj.get("id")) latest = from_union([from_bool, from_none], obj.get("latest")) parent = from_union([from_str, from_none], obj.get("parent")) return DatasetComment(id, aspect_type, content, created, created_at, deleted, entity_id, field, dataset_comment_id, latest, parent) def to_dict(self) -> dict: result: dict = {} result["_id"] = from_union([lambda x: to_class(ObjectID, x), from_none], self.id) result["aspectType"] = from_union([lambda x: to_enum(AspectType, x), from_none], self.aspect_type) result["content"] = from_union([from_str, from_none], self.content) result["created"] = from_union([lambda x: to_class(AuditStamp, x), from_none], self.created) result["createdAt"] = from_union([lambda x: x.isoformat(), from_none], self.created_at) result["deleted"] = from_union([lambda x: to_class(AuditStamp, x), from_none], self.deleted) result["entityId"] = from_union([from_str, from_none], self.entity_id) result["field"] = from_union([from_str, from_none], self.field) result["id"] = from_union([from_str, from_none], self.dataset_comment_id) result["latest"] = from_union([from_bool, from_none], self.latest) result["parent"] = from_union([from_str, from_none], self.parent) return result @dataclass class FieldDocumentation: documentation: Optional[str] = None field_path: Optional[str] = None tests: Optional[List[str]] = None @staticmethod def from_dict(obj: Any) -> 'FieldDocumentation': assert isinstance(obj, dict) documentation = from_union([from_str, from_none], obj.get("documentation")) field_path = from_union([from_str, from_none], obj.get("fieldPath")) tests = from_union([lambda x: from_list(from_str, x), from_none], obj.get("tests")) return FieldDocumentation(documentation, field_path, tests) def to_dict(self) -> dict: result: dict = {} result["documentation"] = from_union([from_str, from_none], self.documentation) result["fieldPath"] = from_union([from_str, from_none], self.field_path) result["tests"] = from_union([lambda x: from_list(from_str, x), from_none], self.tests) return result @dataclass class DatasetDocumentation: id: Optional[ObjectID] = None aspect_type: Optional[AspectType] = None created_at: Optional[datetime] = None dataset_documentations: Optional[List[str]] = None entity_id: Optional[str] = None field_documentations: Optional[List[FieldDocumentation]] = None dataset_documentation_id: Optional[str] = None latest: Optional[bool] = None @staticmethod def from_dict(obj: Any) -> 'DatasetDocumentation': assert isinstance(obj, dict) id = from_union([ObjectID.from_dict, from_none], obj.get("_id")) aspect_type = from_union([AspectType, from_none], obj.get("aspectType")) created_at = from_union([from_datetime, from_none], obj.get("createdAt")) dataset_documentations = from_union([lambda x: from_list(from_str, x), from_none], obj.get("datasetDocumentations")) entity_id = from_union([from_str, from_none], obj.get("entityId")) field_documentations = from_union([lambda x: from_list(FieldDocumentation.from_dict, x), from_none], obj.get("fieldDocumentations")) dataset_documentation_id = from_union([from_str, from_none], obj.get("id")) latest = from_union([from_bool, from_none], obj.get("latest")) return DatasetDocumentation(id, aspect_type, created_at, dataset_documentations, entity_id, field_documentations, dataset_documentation_id, latest) def to_dict(self) -> dict: