66import pwd
77import sys
88import time
9- from typing import Any
9+ from socket import AF_INET , SOCK_DGRAM , socket
10+ from typing import Any , NotRequired , TypedDict , cast
11+ from urllib .error import HTTPError
1012
1113from pydantic_settings import BaseSettings
14+ from requests import get
1215from uptime import boottime , uptime
1316
1417from ..utils import ( # noqa: TID252
2730)
2831from ._settings import Settings
2932
30- logger = get_logger (__name__ )
33+ log = get_logger (__name__ )
34+
35+
36+ class RuntimeDict (TypedDict , total = False ):
37+ """Type for runtime information dictionary."""
38+
39+ environment : str
40+ username : str
41+ process : dict [str , Any ]
42+ host : dict [str , Any ]
43+ python : dict [str , Any ]
44+ environ : dict [str , str ]
45+
46+
47+ class InfoDict (TypedDict , total = False ):
48+ """Type for the info dictionary."""
49+
50+ package : dict [str , Any ]
51+ runtime : RuntimeDict
52+ settings : dict [str , Any ]
53+ # Allow additional string keys with any values for service info
54+ __extra__ : NotRequired [dict [str , Any ]]
3155
3256
3357class Service (BaseService ):
@@ -74,12 +98,47 @@ def is_token_valid(self, token: str) -> bool:
7498 Returns:
7599 bool: True if the token is valid, False otherwise.
76100 """
77- logger .info (token )
101+ log .info (token )
78102 if not self ._settings .token :
79- logger .warning ("Token is not set in settings." )
103+ log .warning ("Token is not set in settings." )
80104 return False
81105 return token == self ._settings .token .get_secret_value ()
82106
107+ @staticmethod
108+ def _get_public_ipv4 (timeout : int = 5 ) -> str | None :
109+ """Get the public IPv4 address of the system.
110+
111+ Args:
112+ timeout (int): Timeout for the request in seconds.
113+
114+ Returns:
115+ str: The public IPv4 address.
116+ """
117+ try :
118+ response = get (url = "https://api.ipify.org" , timeout = timeout )
119+ response .raise_for_status ()
120+ return response .text
121+ except HTTPError as e :
122+ message = f"Failed to get public IP: { e } "
123+ log .exception (message )
124+ return None
125+
126+ @staticmethod
127+ def _get_local_ipv4 () -> str | None :
128+ """Get the local IPv4 address of the system.
129+
130+ Returns:
131+ str: The local IPv4 address.
132+ """
133+ try :
134+ with socket (AF_INET , SOCK_DGRAM ) as connection :
135+ connection .connect (("8.8.8.8" , 80 ))
136+ return str (connection .getsockname ()[0 ])
137+ except Exception as e :
138+ message = f"Failed to get local IP: { e } "
139+ log .exception (message )
140+ return None
141+
83142 @staticmethod
84143 def info (include_environ : bool = False , filter_secrets : bool = True ) -> dict [str , Any ]:
85144 """
@@ -95,7 +154,7 @@ def info(include_environ: bool = False, filter_secrets: bool = True) -> dict[str
95154 dict[str, Any]: Service configuration.
96155 """
97156 bootdatetime = boottime ()
98- rtn = {
157+ rtn : InfoDict = {
99158 "package" : {
100159 "version" : __version__ ,
101160 "name" : __project_name__ ,
@@ -104,35 +163,49 @@ def info(include_environ: bool = False, filter_secrets: bool = True) -> dict[str
104163 },
105164 "runtime" : {
106165 "environment" : __env__ ,
166+ "username" : pwd .getpwuid (os .getuid ())[0 ],
167+ "process" : {
168+ "command_line" : " " .join (sys .argv ),
169+ "entry_point" : sys .argv [0 ] if sys .argv else None ,
170+ "process_info" : json .loads (get_process_info ().model_dump_json ()),
171+ },
172+ "host" : {
173+ "os" : {
174+ "platform" : platform .platform (),
175+ "system" : platform .system (),
176+ "release" : platform .release (),
177+ "version" : platform .version (),
178+ },
179+ "machine" : {
180+ "arch" : platform .machine (),
181+ "processor" : platform .processor (),
182+ "cpu_count" : os .cpu_count (),
183+ },
184+ "network" : {
185+ "hostname" : platform .node (),
186+ "local_ipv4" : Service ._get_local_ipv4 (),
187+ "public_ipv4" : Service ._get_public_ipv4 (),
188+ },
189+ "uptime" : {
190+ "seconds" : uptime (),
191+ "boottime" : bootdatetime .isoformat () if bootdatetime else None ,
192+ },
193+ },
107194 "python" : {
108195 "version" : platform .python_version (),
109196 "compiler" : platform .python_compiler (),
110197 "implementation" : platform .python_implementation (),
111198 "sys.path" : sys .path ,
112- },
113- "interpreter_path" : sys .executable ,
114- "command_line" : " " .join (sys .argv ),
115- "entry_point" : sys .argv [0 ] if sys .argv else None ,
116- "process_info" : json .loads (get_process_info ().model_dump_json ()),
117- "username" : pwd .getpwuid (os .getuid ())[0 ],
118- "host" : {
119- "system" : platform .system (),
120- "release" : platform .release (),
121- "version" : platform .version (),
122- "machine" : platform .machine (),
123- "processor" : platform .processor (),
124- "hostname" : platform .node (),
125- "ip_address" : platform .uname ().node ,
126- "cpu_count" : os .cpu_count (),
127- "uptime" : uptime (),
128- "boottime" : bootdatetime .isoformat () if bootdatetime else None ,
199+ "interpreter_path" : sys .executable ,
129200 },
130201 },
202+ "settings" : {},
131203 }
132204
205+ runtime = cast ("RuntimeDict" , rtn ["runtime" ])
133206 if include_environ :
134207 if filter_secrets :
135- rtn [ " runtime" ] ["environ" ] = {
208+ runtime ["environ" ] = {
136209 k : v
137210 for k , v in os .environ .items ()
138211 if not (
@@ -144,9 +217,9 @@ def info(include_environ: bool = False, filter_secrets: bool = True) -> dict[str
144217 )
145218 }
146219 else :
147- rtn [ " runtime" ] ["environ" ] = dict (os .environ )
220+ runtime ["environ" ] = dict (os .environ )
148221
149- settings = {}
222+ settings : dict [ str , Any ] = {}
150223 for settings_class in locate_subclasses (BaseSettings ):
151224 settings_instance = load_settings (settings_class )
152225 env_prefix = settings_instance .model_config .get ("env_prefix" , "" )
@@ -158,12 +231,16 @@ def info(include_environ: bool = False, filter_secrets: bool = True) -> dict[str
158231 settings [flat_key ] = value
159232 rtn ["settings" ] = settings
160233
234+ # Convert the TypedDict to a regular dict before adding dynamic service keys
235+ result_dict : dict [str , Any ] = dict (rtn )
236+
161237 for service_class in locate_subclasses (BaseService ):
162238 if service_class is not Service :
163239 service = service_class ()
164- rtn [service .key ()] = service .info ()
240+ result_dict [service .key ()] = service .info ()
165241
166- return rtn
242+ log .info ("Service info: %s" , result_dict )
243+ return result_dict
167244
168245 @staticmethod
169246 def div_by_zero () -> float :
0 commit comments