Skip to content

update wrapper #741

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Bionic711 wants to merge 1 commit into
base: Development
Choose a base branch
from
Open

update wrapper #741

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
220 changes: 141 additions & 79 deletions application/single_app/semantic_kernel_plugins/plugin_invocation_logger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import time
import logging
import functools
import inspect
from typing import Any, Dict, List, Optional, Callable
from datetime import datetime
from dataclasses import dataclass, asdict
Expand Down Expand Up @@ -330,93 +331,154 @@ def decorator(func: Callable) -> Callable:
extra={"function_name": func.__name__, "plugin_name": plugin_name},
level=logging.DEBUG)

@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
function_name = func.__name__

log_event(f"[Plugin Function Logger] Function call started",
extra={"plugin_name": plugin_name, "function_name": function_name},
level=logging.DEBUG)

# Prepare parameters (combine args and kwargs)
def _build_parameters(args, kwargs):
parameters = {}
if args:
# Handle 'self' parameter for methods
if hasattr(args[0], '__class__'):
parameters.update({f"arg_{i}": arg for i, arg in enumerate(args[1:])})
else:
parameters.update({f"arg_{i}": arg for i, arg in enumerate(args)})
parameters.update(kwargs)

# Enhanced logging: Show parameters
return parameters

def _log_start(function_name: str):
log_event(
f"[Plugin Function Logger] Function call started",
extra={"plugin_name": plugin_name, "function_name": function_name},
level=logging.DEBUG
)

def _log_parameters(function_name: str, parameters: Dict[str, Any]):
param_str = ", ".join([f"{k}={v}" for k, v in parameters.items()]) if parameters else "no parameters"
log_event(f"[Plugin Function Logger] Function parameters",
extra={
"plugin_name": plugin_name,
"function_name": function_name,
"parameters": parameters,
"param_string": param_str
},
level=logging.DEBUG)

try:
result = func(*args, **kwargs)
end_time = time.time()
duration_ms = (end_time - start_time) * 1000

# Enhanced logging: Show result and timing
result_preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result)
log_event(f"[Plugin Function Logger] Function completed successfully",
extra={
"plugin_name": plugin_name,
"function_name": function_name,
"result_preview": result_preview,
"duration_ms": duration_ms,
"full_function_name": f"{plugin_name}.{function_name}"
},
level=logging.INFO)

log_plugin_invocation(
plugin_name=plugin_name,
function_name=function_name,
parameters=parameters,
result=result,
start_time=start_time,
end_time=end_time,
success=True
)

return result

except Exception as e:
end_time = time.time()
duration_ms = (end_time - start_time) * 1000

# Enhanced logging: Show error and timing
log_event(f"[Plugin Function Logger] Function failed with error",
extra={
"plugin_name": plugin_name,
"function_name": function_name,
"duration_ms": duration_ms,
"error_message": str(e),
"full_function_name": f"{plugin_name}.{function_name}"
},
level=logging.ERROR)

log_plugin_invocation(
plugin_name=plugin_name,
function_name=function_name,
parameters=parameters,
result=None,
start_time=start_time,
end_time=end_time,
success=False,
error_message=str(e)
)

raise # Re-raise the exception

log_event(
f"[Plugin Function Logger] Function parameters",
extra={
"plugin_name": plugin_name,
"function_name": function_name,
"parameters": parameters,
"param_string": param_str
},
level=logging.DEBUG
)

def _log_success(function_name: str, result: Any, duration_ms: float):
result_preview = str(result)[:200] + "..." if len(str(result)) > 200 else str(result)
log_event(
f"[Plugin Function Logger] Function completed successfully",
extra={
"plugin_name": plugin_name,
"function_name": function_name,
"result_preview": result_preview,
"duration_ms": duration_ms,
"full_function_name": f"{plugin_name}.{function_name}"
},
level=logging.INFO
)

def _log_failure(function_name: str, error: Exception, duration_ms: float):
log_event(
f"[Plugin Function Logger] Function failed with error",
extra={
"plugin_name": plugin_name,
"function_name": function_name,
"duration_ms": duration_ms,
"error_message": str(error),
"full_function_name": f"{plugin_name}.{function_name}"
},
level=logging.ERROR
)

if inspect.iscoroutinefunction(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
start_time = time.time()
function_name = func.__name__
_log_start(function_name)
parameters = _build_parameters(args, kwargs)
_log_parameters(function_name, parameters)

try:
result = await func(*args, **kwargs)
end_time = time.time()
duration_ms = (end_time - start_time) * 1000
_log_success(function_name, result, duration_ms)

log_plugin_invocation(
plugin_name=plugin_name,
function_name=function_name,
parameters=parameters,
result=result,
start_time=start_time,
end_time=end_time,
success=True
)

return result

except Exception as e:
end_time = time.time()
duration_ms = (end_time - start_time) * 1000
_log_failure(function_name, e, duration_ms)

log_plugin_invocation(
plugin_name=plugin_name,
function_name=function_name,
parameters=parameters,
result=None,
start_time=start_time,
end_time=end_time,
success=False,
error_message=str(e)
)

raise
else:
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
function_name = func.__name__
_log_start(function_name)
parameters = _build_parameters(args, kwargs)
_log_parameters(function_name, parameters)

try:
result = func(*args, **kwargs)
if inspect.iscoroutine(result):
raise RuntimeError("Async coroutine returned from sync function wrapper")
end_time = time.time()
duration_ms = (end_time - start_time) * 1000
_log_success(function_name, result, duration_ms)

log_plugin_invocation(
plugin_name=plugin_name,
function_name=function_name,
parameters=parameters,
result=result,
start_time=start_time,
end_time=end_time,
success=True
)

return result

except Exception as e:
end_time = time.time()
duration_ms = (end_time - start_time) * 1000
_log_failure(function_name, e, duration_ms)

log_plugin_invocation(
plugin_name=plugin_name,
function_name=function_name,
parameters=parameters,
result=None,
start_time=start_time,
end_time=end_time,
success=False,
error_message=str(e)
)

raise

return wrapper
return decorator

Expand Down