mirror of
https://github.com/yavook/kiwi-simple-metrics.git
synced 2024-11-24 16:33:01 +00:00
settings module split
This commit is contained in:
parent
f23d7f2e37
commit
f489f90a92
3 changed files with 62 additions and 53 deletions
36
kiwi_simple_metrics/settings/__init__.py
Normal file
36
kiwi_simple_metrics/settings/__init__.py
Normal file
|
@ -0,0 +1,36 @@
|
||||||
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||||
|
|
||||||
|
from . import metric, misc
|
||||||
|
from .metric import MetricSettings
|
||||||
|
|
||||||
|
|
||||||
|
class Settings(BaseSettings):
|
||||||
|
model_config = SettingsConfigDict(
|
||||||
|
env_prefix="METRIC__",
|
||||||
|
env_nested_delimiter="__",
|
||||||
|
)
|
||||||
|
|
||||||
|
# time between gathering reports
|
||||||
|
interval: float = 600
|
||||||
|
|
||||||
|
# reporting to stdout
|
||||||
|
log: misc.LogSettings = misc.LogSettings()
|
||||||
|
|
||||||
|
# separates metrics and values in reports
|
||||||
|
separator: str = ", "
|
||||||
|
|
||||||
|
# metrics settings
|
||||||
|
cpu: metric.CpuMS = metric.CpuMS()
|
||||||
|
memory: metric.MemoryMS = metric.MemoryMS()
|
||||||
|
disk: metric.DiskMS = metric.DiskMS()
|
||||||
|
|
||||||
|
# pinging webhooks
|
||||||
|
webhook: misc.WebhookSettings = misc.WebhookSettings()
|
||||||
|
|
||||||
|
|
||||||
|
SETTINGS = Settings()
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"MetricSettings",
|
||||||
|
"SETTINGS",
|
||||||
|
]
|
|
@ -1,9 +1,8 @@
|
||||||
import math
|
import math
|
||||||
from typing import Any, Literal
|
from typing import Any, Literal
|
||||||
|
|
||||||
from pydantic import (AnyUrl, BaseModel, DirectoryPath, Field,
|
from pydantic import (BaseModel, DirectoryPath, Field, FieldValidationInfo,
|
||||||
FieldValidationInfo, field_validator)
|
field_validator)
|
||||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
||||||
|
|
||||||
|
|
||||||
class MetricSettings(BaseModel):
|
class MetricSettings(BaseModel):
|
||||||
|
@ -80,53 +79,3 @@ class DiskMS(MetricSettings):
|
||||||
|
|
||||||
# paths to check for disk space
|
# paths to check for disk space
|
||||||
paths: list[DirectoryPath] = Field(default_factory=list)
|
paths: list[DirectoryPath] = Field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
class LogSettings(BaseModel):
|
|
||||||
# if True, prints reports to stdout
|
|
||||||
enabled: bool = False
|
|
||||||
|
|
||||||
# how to format reports to stdout
|
|
||||||
format: str = "[{state}] {result}"
|
|
||||||
|
|
||||||
|
|
||||||
class WebhookSettings(BaseModel):
|
|
||||||
# webhooks to ping on success/on failure
|
|
||||||
url: AnyUrl | None = None
|
|
||||||
fail: AnyUrl | None = None
|
|
||||||
|
|
||||||
# allow insecure/self-signed webhook targets
|
|
||||||
insecure: bool = False
|
|
||||||
|
|
||||||
def get_url(self, failed: bool) -> AnyUrl | None:
|
|
||||||
if failed:
|
|
||||||
return self.fail
|
|
||||||
|
|
||||||
return self.url
|
|
||||||
|
|
||||||
|
|
||||||
class Settings(BaseSettings):
|
|
||||||
model_config = SettingsConfigDict(
|
|
||||||
env_prefix="METRIC__",
|
|
||||||
env_nested_delimiter="__",
|
|
||||||
)
|
|
||||||
|
|
||||||
# time between gathering reports
|
|
||||||
interval: float = 600
|
|
||||||
|
|
||||||
# reporting to stdout
|
|
||||||
log: LogSettings = LogSettings()
|
|
||||||
|
|
||||||
# separates metrics and values in reports
|
|
||||||
separator: str = ", "
|
|
||||||
|
|
||||||
# metrics settings
|
|
||||||
cpu: CpuMS = CpuMS()
|
|
||||||
memory: MemoryMS = MemoryMS()
|
|
||||||
disk: DiskMS = DiskMS()
|
|
||||||
|
|
||||||
# pinging webhooks
|
|
||||||
webhook: WebhookSettings = WebhookSettings()
|
|
||||||
|
|
||||||
|
|
||||||
SETTINGS = Settings()
|
|
24
kiwi_simple_metrics/settings/misc.py
Normal file
24
kiwi_simple_metrics/settings/misc.py
Normal file
|
@ -0,0 +1,24 @@
|
||||||
|
from pydantic import AnyUrl, BaseModel
|
||||||
|
|
||||||
|
|
||||||
|
class LogSettings(BaseModel):
|
||||||
|
# if True, prints reports to stdout
|
||||||
|
enabled: bool = False
|
||||||
|
|
||||||
|
# how to format reports to stdout
|
||||||
|
format: str = "[{state}] {result}"
|
||||||
|
|
||||||
|
|
||||||
|
class WebhookSettings(BaseModel):
|
||||||
|
# webhooks to ping on success/on failure
|
||||||
|
url: AnyUrl | None = None
|
||||||
|
fail: AnyUrl | None = None
|
||||||
|
|
||||||
|
# allow insecure/self-signed webhook targets
|
||||||
|
insecure: bool = False
|
||||||
|
|
||||||
|
def get_url(self, failed: bool) -> AnyUrl | None:
|
||||||
|
if failed:
|
||||||
|
return self.fail
|
||||||
|
|
||||||
|
return self.url
|
Loading…
Reference in a new issue