mareza/python_utils
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
M. A. Reza 2026-02-18 15:13:57 +03:30
parent 89669f8a15
commit 4f8b67a794
1 changed files with 21 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
utils/logger.py
View File

@ -5,6 +5,7 @@ import logging.handlers
import os import os
from pathlib import Path from pathlib import Path
import sys import sys
from zoneinfo import ZoneInfo
def setup_logging( def setup_logging(
@ -25,6 +26,7 @@ def setup_logging(
microseconds: int = 3, # 0, 3, or 6 microseconds: int = 3, # 0, 3, or 6
include_pid: bool = True, include_pid: bool = True,
include_thread: bool = True, include_thread: bool = True,
time_zone_name: str = "UTC",
) -> None: ) -> None:
""" """
Script-friendly logging: Script-friendly logging:
@ -106,6 +108,7 @@ def setup_logging(
include_name=include_name, include_name=include_name,
include_pid=include_pid, include_pid=include_pid,
include_thread=include_thread, include_thread=include_thread,
time_zone_name=time_zone_name,
) )
else: else:
formatter = _TextFormatter( formatter = _TextFormatter(
@ -116,14 +119,15 @@ def setup_logging(
include_name=include_name, include_name=include_name,
include_pid=include_pid, include_pid=include_pid,
include_thread=include_thread, include_thread=include_thread,
time_zone_name=time_zone_name,
) )
# ---- Console handler ---- # ---- Console handler ----
if use_rich and not json_logs: if use_rich and not json_logs:
console = RichHandler( console = RichHandler(
level=console_level, level=console_level,
rich_tracebacks=True, rich_tracebacks=False,
tracebacks_show_locals=True, tracebacks_show_locals=False,
show_time=False, # YOU already handle time show_time=False, # YOU already handle time
show_level=False, # YOU already handle level show_level=False, # YOU already handle level
show_path=False, # avoid noise show_path=False, # avoid noise
@ -181,7 +185,7 @@ def _level_from_env(val: str | None, *, default: int) -> int:
return getattr(logging, name, default) return getattr(logging, name, default)
def _format_timestamp(*, calendar: str, microseconds: int) -> str: def _format_timestamp(*, calendar: str, microseconds: int, time_zone_name: str) -> str:
""" """
microseconds: microseconds:
- 0: YYYY-mm-dd HH:MM:SS - 0: YYYY-mm-dd HH:MM:SS
@ -191,7 +195,7 @@ def _format_timestamp(*, calendar: str, microseconds: int) -> str:
from datetime import datetime from datetime import datetime
if calendar == "gregorian": if calendar == "gregorian":
now = datetime.now() now = datetime.now(tz=ZoneInfo(time_zone_name))
base = now.strftime("%Y-%m-%d %H:%M:%S") base = now.strftime("%Y-%m-%d %H:%M:%S")
if microseconds == 0: if microseconds == 0:
return base return base
@ -205,7 +209,7 @@ def _format_timestamp(*, calendar: str, microseconds: int) -> str:
import jdatetime import jdatetime
except Exception: except Exception:
# fallback # fallback
now = datetime.now() now = datetime.now(tz=ZoneInfo(time_zone_name))
base = now.strftime("%Y-%m-%d %H:%M:%S") base = now.strftime("%Y-%m-%d %H:%M:%S")
if microseconds == 0: if microseconds == 0:
return base + " (gregorian-fallback)" return base + " (gregorian-fallback)"
@ -214,7 +218,7 @@ def _format_timestamp(*, calendar: str, microseconds: int) -> str:
return f"{base}.{us // 1000:03d} (gregorian-fallback)" return f"{base}.{us // 1000:03d} (gregorian-fallback)"
return f"{base}.{us:06d} (gregorian-fallback)" return f"{base}.{us:06d} (gregorian-fallback)"
jnow = jdatetime.datetime.now() jnow = jdatetime.datetime.now(tz=ZoneInfo(time_zone_name))
base = jnow.strftime("%Y-%m-%d %H:%M:%S") base = jnow.strftime("%Y-%m-%d %H:%M:%S")
if microseconds == 0: if microseconds == 0:
return base return base
@ -246,6 +250,7 @@ class _TextFormatter(logging.Formatter):
include_name: bool, include_name: bool,
include_pid: bool, include_pid: bool,
include_thread: bool, include_thread: bool,
time_zone_name: str,
): ):
super().__init__() super().__init__()
self.app_name = app_name self.app_name = app_name
@ -255,9 +260,10 @@ class _TextFormatter(logging.Formatter):
self.include_name = include_name self.include_name = include_name
self.include_pid = include_pid self.include_pid = include_pid
self.include_thread = include_thread self.include_thread = include_thread
self.time_zone_name = time_zone_name
def format(self, record: logging.LogRecord) -> str: def format(self, record: logging.LogRecord) -> str:
ts = _format_timestamp(calendar=self.calendar, microseconds=self.microseconds) ts = _format_timestamp(calendar=self.calendar, microseconds=self.microseconds, time_zone_name=self.time_zone_name)
# best practice: keep a stable, grep-friendly prefix # best practice: keep a stable, grep-friendly prefix
prefix_parts = [ts, record.levelname] prefix_parts = [ts, record.levelname]
@ -289,6 +295,7 @@ class _JsonFormatter(logging.Formatter):
include_name: bool, include_name: bool,
include_pid: bool, include_pid: bool,
include_thread: bool, include_thread: bool,
time_zone_name: str,
): ):
super().__init__() super().__init__()
self.app_name = app_name self.app_name = app_name
@ -298,12 +305,15 @@ class _JsonFormatter(logging.Formatter):
self.include_name = include_name self.include_name = include_name
self.include_pid = include_pid self.include_pid = include_pid
self.include_thread = include_thread self.include_thread = include_thread
self.time_zone_name = time_zone_name
def format(self, record: logging.LogRecord) -> str: def format(self, record: logging.LogRecord) -> str:
import json import json
payload: dict[str, object] = { payload: dict[str, object] = {
"ts": _format_timestamp(calendar=self.calendar, microseconds=self.microseconds), "ts": _format_timestamp(
calendar=self.calendar, microseconds=self.microseconds, time_zone_name=self.time_zone_name
),
"level": record.levelname, "level": record.levelname,
"msg": record.getMessage(), "msg": record.getMessage(),
} }
@ -329,8 +339,9 @@ if __name__ == "__main__":
include_name=True, include_name=True,
include_pid=False, include_pid=False,
include_thread=False, include_thread=False,
console_level=logging.DEBUG, console_level=logging.INFO,
use_rich=True, use_rich=True,
time_zone_name="Asia/Tehran",
) )
logging.getLogger("elasticsearch").setLevel(logging.WARNING) logging.getLogger("elasticsearch").setLevel(logging.WARNING)
logging.getLogger("elastic_transport").setLevel(logging.WARNING) logging.getLogger("elastic_transport").setLevel(logging.WARNING)
@ -338,6 +349,7 @@ if __name__ == "__main__":
logging.getLogger("tortoise.db_client").setLevel(logging.WARNING) logging.getLogger("tortoise.db_client").setLevel(logging.WARNING)
logging.getLogger("asyncio").setLevel(logging.WARNING) logging.getLogger("asyncio").setLevel(logging.WARNING)
logging.getLogger("tortoise").setLevel(logging.WARNING) logging.getLogger("tortoise").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
log.info("script_started") log.info("script_started")
# time.sleep(2) # time.sleep(2)