93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
from enum import StrEnum
|
|
from dataclasses import field
|
|
from pydantic import BaseModel
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
class Status(StrEnum):
|
|
GOOD = "GOOD"
|
|
BAD = "BAD"
|
|
|
|
|
|
class LogTiming(BaseModel):
|
|
start_date_timestamp: int = 0
|
|
start_date: str = ""
|
|
end_date: str = ""
|
|
duration_sec: float = 0.0
|
|
|
|
|
|
class BaseLog(BaseModel):
|
|
name: str
|
|
timing: LogTiming = field(default=LogTiming())
|
|
status: Status
|
|
log_file_path: str = "" # If set we have logged this particular step to a file
|
|
|
|
|
|
class ActLog(BaseLog):
|
|
error: str = ""
|
|
msg: str = ""
|
|
logs: list[StepLogModel]
|
|
|
|
@property
|
|
def failed(self) -> bool:
|
|
""" Checks if the step failed """
|
|
return len(self.error) > 0 or self.status == Status.BAD
|
|
|
|
@classmethod
|
|
def ok(cls, name: str, logs: list[StepLogModel], msg: str = "success") -> ActLog:
|
|
return cls(name=name, status=Status.GOOD, msg=msg, logs=logs)
|
|
|
|
@classmethod
|
|
def fail(cls, name: str, err: str, logs: list[StepLogModel]) -> ActLog:
|
|
return cls(name=name, status=Status.BAD, error=err, logs=logs)
|
|
|
|
|
|
class PlaybookLog(BaseLog):
|
|
errors: str = ""
|
|
summary: str = ""
|
|
act_logs: list[ActLog]
|
|
|
|
@property
|
|
def failed(self) -> bool:
|
|
return len(self.errors) > 0 or self.status == Status.BAD
|
|
|
|
@classmethod
|
|
def ok(cls, name: str, logs: list[ActLog], msg: str = "success") -> PlaybookLog:
|
|
return cls(name=name, status=Status.GOOD, summary=msg, act_logs=logs)
|
|
|
|
@classmethod
|
|
def fail(cls, name: str, logs: list[ActLog], err: str) -> PlaybookLog:
|
|
return cls(name=name, status=Status.BAD, errors=err, act_logs=logs)
|
|
|
|
|
|
class StepLogModel(BaseLog):
|
|
error: str = ""
|
|
msg: str = ""
|
|
substeps_logs: list[StepLogModel] = []
|
|
pipe_ctx: dict[str, object] = field(default_factory=dict)
|
|
|
|
@property
|
|
def failed(self) -> bool:
|
|
""" Checks if the step failed """
|
|
return len(self.error) > 0 or self.status == Status.BAD
|
|
|
|
@classmethod
|
|
def ok(cls, name: str, msg: str = "success", pipe_ctx: dict[str, object] = {},
|
|
substeps: list[StepLogModel] = []) -> StepLogModel:
|
|
return cls(name=name, status=Status.GOOD, msg=msg, pipe_ctx=pipe_ctx,
|
|
substeps_logs=substeps)
|
|
|
|
@classmethod
|
|
def fail(cls, name: str, err: str) -> StepLogModel:
|
|
return cls(name=name, status=Status.BAD, error=err)
|
|
|
|
|
|
def log_file_name(name: str, start_date_ns: int) -> str:
|
|
""" Format to ISO 8601. """
|
|
seconds = start_date_ns // 1_000_000_000
|
|
ms = (start_date_ns % 1_000_000_000) // 1_000_000
|
|
|
|
dt = datetime.fromtimestamp(seconds, tz=timezone.utc).astimezone()
|
|
|
|
return dt.strftime(f"{name}_%Y-%m-%d_%H-%M-%S.{ms:03d}.log")
|