gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

ruff

Browse source
This commit is contained in:
hajdul88 2026-01-19 12:43:46 +01:00
parent 77543dbd90
commit fb77f71c30
1 changed files with 10 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
cognee/shared/usage_logger.py
View file

@ -158,7 +158,7 @@ async def _log_usage_async(
end_time: datetime, end_time: datetime,
): ):
"""Asynchronously log function usage to Redis. """Asynchronously log function usage to Redis.
Args: Args:
function_name: Name of the function being logged. function_name: Name of the function being logged.
log_type: Type of log entry (e.g., "api_endpoint", "mcp_tool", "function"). log_type: Type of log entry (e.g., "api_endpoint", "mcp_tool", "function").
@ -167,9 +167,10 @@ async def _log_usage_async(
result: Function return value (will be sanitized). result: Function return value (will be sanitized).
success: Whether the function executed successfully. success: Whether the function executed successfully.
error: Error message if function failed, None otherwise. error: Error message if function failed, None otherwise.
duration_ms: Execution duration in milliseconds.
start_time: Function start timestamp. start_time: Function start timestamp.
end_time: Function end timestamp. end_time: Function end timestamp.
Note: Note:
This function silently handles errors to avoid disrupting the original This function silently handles errors to avoid disrupting the original
function execution. Logs are written to Redis with TTL from config. function execution. Logs are written to Redis with TTL from config.
@ -247,13 +248,13 @@ def log_usage(function_name: Optional[str] = None, log_type: str = "function"):
def decorator(func: Callable) -> Callable: def decorator(func: Callable) -> Callable:
"""Inner decorator that wraps the function with usage logging. """Inner decorator that wraps the function with usage logging.
Args: Args:
func: The async function to wrap with usage logging. func: The async function to wrap with usage logging.
Returns: Returns:
Callable: The wrapped function with usage logging enabled. Callable: The wrapped function with usage logging enabled.
Raises: Raises:
UsageLoggerError: If the function is not async. UsageLoggerError: If the function is not async.
""" """
@ -265,20 +266,20 @@ def log_usage(function_name: Optional[str] = None, log_type: str = "function"):
@wraps(func) @wraps(func)
async def async_wrapper(*args, **kwargs): async def async_wrapper(*args, **kwargs):
"""Wrapper function that executes the original function and logs usage. """Wrapper function that executes the original function and logs usage.
This wrapper: This wrapper:
- Extracts user ID and parameters from function arguments - Extracts user ID and parameters from function arguments
- Executes the original function - Executes the original function
- Captures result, success status, and any errors - Captures result, success status, and any errors
- Logs usage information asynchronously without blocking - Logs usage information asynchronously without blocking
Args: Args:
*args: Positional arguments passed to the original function. *args: Positional arguments passed to the original function.
**kwargs: Keyword arguments passed to the original function. **kwargs: Keyword arguments passed to the original function.
Returns: Returns:
Any: The return value of the original function. Any: The return value of the original function.
Raises: Raises:
Any exception raised by the original function (re-raised after logging). Any exception raised by the original function (re-raised after logging).
""" """