""" AUTHOR: Khushal P Soonderji DATE: Wednesday, 25th Dec., 2024. OBJECTIVE: To provide a structure to work with enlisting servers and maintaining their status. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator, AwareDatetime from typing import Optional, Literal, Union # My utils: from utils_v2.string import regex from utils_v2.date_time import date_time # To work with date and time: import datetime # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class CoreServerInfoModel(BaseModel): hostname: str = Field( description = "the hostname to identify the server", frozen = True ) os: str = Field( description = "the os the server is running", frozen = True ) cpu: str = Field( description = "the cpu that the server has in it", frozen = True ) pid: str | int = Field( description = "the process id that registered the details of this server", frozen = True ) ppid: str | int = Field( description = "the parent process id that registered the details of this server", frozen = True ) ipAddr: str | None = Field( description = "the ip address of the server", frozen = True ) portNo: str | None = Field( description = "the port no. that this service is running on", frozen = True ) project: str = Field( description = "the name of the project that this server is running", frozen = True ) service: str = Field( description = "the service in the said project that this server is running", frozen = True ) healthCheckUrl: str = Field( description = "a get request will be sent to this url to see if the server is up", frozen = True ) healthCheckInterval: int = Field( description = "the seconds after which to check for the service being up", ge = 30, frozen = True ) healthAlertUrl: str = Field( description = "a get request will be sent to this url when the server goes offline", frozen = True ) online: bool = Field( description = "whether this service is online or not", default = False, frozen = False ) batchId: str | None = Field( description = "set some value here when checking the status of this server, set it to null once done", default = None, frozen = False ) batchTs: AwareDatetime | None = Field( description = "set the time (utc) when checking the status of this server, set it to null once done", default = None, frozen = False ) firstRegTs: AwareDatetime = Field( description = "the first time (utc) this server was registered in the database", frozen = True ) lastRegTs: AwareDatetime = Field( description = "the last time (utc) this server was registered in the database", frozen = True ) lastCheckTs: AwareDatetime = Field( description = "the last time (utc) this server was checked for being online", frozen = True ) checkAfterTs: AwareDatetime = Field( description = "the time (utc) after which this server needs to be checked again for being online", frozen = True ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "ignore" # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @staticmethod def parse_date_time(value): # If a null value was given, # we can't do anything: if not value: value = None # If the input is a string: if isinstance(value, str): value = value.strip() value = date_time.parse_date_time( input_value = value, timezone = date_time.TIMEZONE_UTC, date_formats = ["%Y-%m-%d"] ) # If the input is already a date-time object, # we just normalize the timestamp: if isinstance(value, datetime.datetime): value = date_time.to_timezone( value, timezone = date_time.TIMEZONE_UTC ) # Done here: return value @field_validator( "batchTs", "firstRegTs", "lastRegTs", "lastCheckTs", "checkAfterTs", mode = "before" ) def parse_given_date_time(cls, value): return cls.parse_date_time(value) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass