airbyte-protocol-models-dataclasses

import json from dataclasses import asdict from datetime import datetime from airbyte_protocol_models_dataclasses.v0.models import AirbyteMessage, AirbyteRecordMessage, Type # Create an AirbyteRecordMessage representing a data record record_data = { "id": 1, "name": "Test User", "email": "test.user@example.com", "created_at": datetime.now().isoformat() } record_message = AirbyteRecordMessage( stream="users", data=record_data, emitted_at=int(datetime.now().timestamp() * 1000), ) # Wrap the record in a top-level AirbyteMessage airbyte_message = AirbyteMessage( type=Type.RECORD, record=record_message, ) # Convert the dataclass instance to a dictionary and then to a JSON string json_output = json.dumps(asdict(airbyte_message), indent=2) print("\n--- Airbyte Record Message ---") print(json_output) # Example of creating an AirbyteStateMessage state_data = {"users_sync_progress": {"last_id": 100, "updated_at": datetime.now().isoformat()}} state_message = AirbyteMessage( type=Type.STATE, state=state_data, ) # Serialize the state message json_state_output = json.dumps(asdict(state_message), indent=2) print("\n--- Airbyte State Message ---") print(json_state_output)