mirror of
https://github.com/tiqi-group/pydase_service_base.git
synced 2025-04-22 17:20:03 +02:00
49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
from datetime import datetime
|
|
from enum import Enum
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
# TODO: we are not using all possible severity levels from loguru here (see
|
|
# https://loguru.readthedocs.io/en/stable/api/logger.html). Either add them to this enum
|
|
# or prevent people from using the other severity levels or tell them not to use them.
|
|
class LogBroadcastMessageLevel(Enum):
|
|
# low level messages that help tracking down issues
|
|
DEBUG = "DEBUG"
|
|
# generic messages
|
|
INFO = "INFO"
|
|
# something is suspicious, but program can continue normally
|
|
WARNING = "WARNING"
|
|
# something is wrong, correct program execution cannot be guaranteed
|
|
ERROR = "ERROR"
|
|
# program cannot continue to run any more
|
|
CRITICAL = "CRITICAL"
|
|
|
|
|
|
class LogBroadcastMessage(BaseModel):
|
|
originator: str = Field(
|
|
..., description="The script or service where this message comes from."
|
|
)
|
|
|
|
level: LogBroadcastMessageLevel = Field(
|
|
...,
|
|
description="Log level of this message. Possible values: DEBUG, INFO, WARNING, CRITICAL",
|
|
)
|
|
message: str = Field(..., description="Actual content of the log message.")
|
|
package: str = Field(
|
|
..., description="The python package where this message came from."
|
|
)
|
|
line_number: str = Field(
|
|
..., description="Line number where this message originated from."
|
|
)
|
|
function: str = Field(
|
|
..., description="The function that embeds the log message call."
|
|
)
|
|
timestamp: datetime = Field(..., description="When the log message was created.")
|
|
|
|
@property
|
|
def routing_key(self) -> str:
|
|
"""Routing key based on contents of this message.
|
|
Constructed as: {self.__class__.__name__}.{self.level}.{self.package}"""
|
|
return f"{self.originator}.{self.level}.{self.package}.{self.function}"
|