public/fastapi-best-architecture
1
0
Fork 0
mirror of https://github.com/fastapi-practices/fastapi-best-architecture.git synced 2026-09-21 21:15:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Optimize the server information retrieval (#595)

Browse Source 
* Optimize the server information retrieval

* Update import

* Fix typing
This commit is contained in:
Wu Clan
2025-04-28 15:41:53 +08:00
committed by GitHub
parent 9b3425d801
commit bc5d142920
1 changed files with 60 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/utils/server_info.py
+60 -43
View File
@@ -61,55 +61,62 @@ class ServerInfo:
:param td: 时间差对象
:return:
"""
total_seconds = round(td.total_seconds())
return ServerInfo.fmt_seconds(total_seconds)
return ServerInfo.fmt_seconds(round(td.total_seconds()))
@staticmethod
def get_cpu_info() -> dict[str, float | int]:
"""获取 CPU 信息"""
cpu_info = {'usage': round(psutil.cpu_percent(percpu=False), 2)} # %
cpu_info = {
'usage': round(psutil.cpu_percent(interval=0.1), 2), # %
'logical_num': psutil.cpu_count(logical=True) or 0,
'physical_num': psutil.cpu_count(logical=False) or 0,
'max_freq': 0.0,
'min_freq': 0.0,
'current_freq': 0.0,
}
try:
# CPU 频率信息，最大、最小和当前频率
cpu_freq = psutil.cpu_freq()
cpu_info.update({
'max_freq': round(cpu_freq.max, 2), # MHz
'min_freq': round(cpu_freq.min, 2), # MHz
'current_freq': round(cpu_freq.current, 2), # MHz
})
if hasattr(psutil, 'cpu_freq'):
cpu_freq = psutil.cpu_freq()
if cpu_freq: # Some systems return None
cpu_info.update({
'max_freq': round(cpu_freq.max, 2),
'min_freq': round(cpu_freq.min, 2),
'current_freq': round(cpu_freq.current, 2),
})
except Exception:
cpu_info.update({'max_freq': 0, 'min_freq': 0, 'current_freq': 0})
pass
# CPU 逻辑核心数，物理核心数
cpu_info.update({
'logical_num': psutil.cpu_count(logical=True),
'physical_num': psutil.cpu_count(logical=False),
})
return cpu_info
@staticmethod
def get_mem_info() -> dict[str, float]:
"""获取内存信息"""
mem = psutil.virtual_memory()
gb_factor = 1024**3
return {
'total': round(mem.total / 1024 / 1024 / 1024, 2), # GB
'used': round(mem.used / 1024 / 1024 / 1024, 2), # GB
'free': round(mem.available / 1024 / 1024 / 1024, 2), # GB
'usage': round(mem.percent, 2), # %
'total': round(mem.total / gb_factor, 2),
'used': round(mem.used / gb_factor, 2),
'free': round(mem.available / gb_factor, 2),
'usage': round(mem.percent, 2),
}
@staticmethod
def get_sys_info() -> dict[str, str]:
"""获取服务器信息"""
hostname = socket.gethostname()
ip = '127.0.0.1'
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sk:
sk.connect(('8.8.8.8', 80))
ip = sk.getsockname()[0]
except socket.gaierror:
ip = '127.0.0.1'
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.settimeout(0.5)
s.connect(('8.8.8.8', 80))
ip = s.getsockname()[0]
except (socket.gaierror, socket.timeout, OSError):
pass
return {
'name': socket.gethostname(),
'name': hostname,
'ip': ip,
'os': platform.system(),
'arch': platform.machine(),
@@ -119,17 +126,20 @@ class ServerInfo:
def get_disk_info() -> list[dict[str, str]]:
"""获取磁盘信息"""
disk_info = []
for disk in psutil.disk_partitions():
usage = psutil.disk_usage(disk.mountpoint)
disk_info.append({
'dir': disk.mountpoint,
'type': disk.fstype,
'device': disk.device,
'total': ServerInfo.format_bytes(usage.total),
'free': ServerInfo.format_bytes(usage.free),
'used': ServerInfo.format_bytes(usage.used),
'usage': f'{round(usage.percent, 2)} %',
})
for partition in psutil.disk_partitions(all=False):
try:
usage = psutil.disk_usage(partition.mountpoint)
disk_info.append({
'dir': partition.mountpoint,
'type': partition.fstype,
'device': partition.device,
'total': ServerInfo.format_bytes(usage.total),
'free': ServerInfo.format_bytes(usage.free),
'used': ServerInfo.format_bytes(usage.used),
'usage': f'{usage.percent:.2f}%',
})
except (PermissionError, psutil.AccessDenied):
continue
return disk_info
@staticmethod
@@ -137,18 +147,25 @@ class ServerInfo:
"""获取服务信息"""
process = psutil.Process(os.getpid())
mem_info = process.memory_info()
start_time = timezone.f_datetime(datetime.utcfromtimestamp(process.create_time()).replace(tzinfo=tz.utc))
try:
create_time = datetime.fromtimestamp(process.create_time(), tz=tz.utc)
start_time = timezone.f_datetime(create_time)
except (psutil.NoSuchProcess, OSError):
start_time = timezone.now()
elapsed = ServerInfo.fmt_timedelta(timezone.now() - start_time)
return {
'name': 'Python3',
'version': platform.python_version(),
'home': sys.executable,
'cpu_usage': f'{round(process.cpu_percent(interval=1), 2)} %',
'mem_vms': ServerInfo.format_bytes(mem_info.vms), # 虚拟内存, 即当前进程申请的虚拟内存
'mem_rss': ServerInfo.format_bytes(mem_info.rss), # 常驻内存, 即当前进程实际使用的物理内存
'mem_free': ServerInfo.format_bytes(mem_info.vms - mem_info.rss), # 空闲内存
'cpu_usage': f'{process.cpu_percent(interval=0.1):.2f}%',
'mem_vms': ServerInfo.format_bytes(mem_info.vms),
'mem_rss': ServerInfo.format_bytes(mem_info.rss),
'mem_free': ServerInfo.format_bytes(mem_info.vms - mem_info.rss),
'startup': start_time,
'elapsed': ServerInfo.fmt_timedelta(timezone.now() - start_time),
'elapsed': elapsed,
}