37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from typing import Any, Dict, Optional, Generic, TypeVar
|
|
from datetime import datetime
|
|
|
|
OutputType = TypeVar("OutputType")
|
|
|
|
class ServiceOutput(Generic[OutputType]):
|
|
""" Base para cualquier salida de los pipelines"""
|
|
|
|
def __init__(self, data: OutputType, operation: str, status: str = "success"):
|
|
self.data = data
|
|
self.operation = operation
|
|
self.status = status
|
|
self.timestamp = datetime.utcnow()
|
|
self.metadata = {}
|
|
|
|
def add_metadata(self, key: str, value: Any) -> 'ServiceOutput':
|
|
self.metadata[key] = value
|
|
return self
|
|
|
|
def to_dict(self) -> Dict:
|
|
result = {
|
|
"operation": self.operation,
|
|
"status": self.status,
|
|
"timestamp": self.timestamp.isoformat(),
|
|
"data": self.data
|
|
}
|
|
if self.metadata:
|
|
result["metadata"] = self.metadata
|
|
return result
|
|
|
|
def to_json(self) -> str:
|
|
import json
|
|
return json.dumps(self.to_dict(), default=str)
|
|
|
|
def to_print(self) -> str:
|
|
"""Para salida en consola"""
|
|
return f"[{self.status.upper()}] {self.operation}: {self.data}" |