| from __future__ import annotations |
|
|
| import dataclasses |
| import json |
| import os |
| import secrets |
| from datetime import datetime |
| from typing import Any, Dict |
|
|
|
|
| def create_id(length: int = 4) -> str: |
| date_str = datetime.utcnow().strftime("%Y%m%d-%H%M%S") |
| ran_str = secrets.token_hex(length) |
| return f"{date_str}-{ran_str}" |
|
|
|
|
| def read_yaml(path: str) -> Dict[str, Any]: |
| import yaml |
|
|
| try: |
| with open(path, "r") as file: |
| return yaml.safe_load(file) |
| except Exception as e: |
| raise ValueError(f"Yaml loading failed due to: {e}") |
|
|
|
|
| def validate_yaml(content: Any, schema: str) -> bool: |
| import jsonschema |
|
|
| |
| |
| |
| assert schema in ["example_schema", "plugin_schema"] |
| if schema == "example_schema": |
| schema_path = os.path.join(os.path.dirname(__file__), "../plugin/taskweaver.conversation-v1.schema.json") |
| else: |
| schema_path = os.path.join(os.path.dirname(__file__), "../plugin/taskweaver.plugin-v1.schema.json") |
|
|
| with open(schema_path) as file: |
| schema_object: Any = json.load(file) |
| try: |
| jsonschema.validate(content, schema=schema_object) |
| return True |
| except jsonschema.ValidationError as e: |
| raise ValueError(f"Yaml validation failed due to: {e}") |
|
|
|
|
| class EnhancedJSONEncoder(json.JSONEncoder): |
| def default(self, o: Any): |
| if dataclasses.is_dataclass(o): |
| return dataclasses.asdict(o) |
| return super().default(o) |
|
|
|
|
| def json_dumps(obj: Any) -> str: |
| return json.dumps(obj, cls=EnhancedJSONEncoder) |
|
|
|
|
| def json_dump(obj: Any, fp: Any): |
| json.dump(obj, fp, cls=EnhancedJSONEncoder) |
|
|
