39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
import inspect
|
|
|
|
|
|
def error_not_async(func):
|
|
if not inspect.iscoroutinefunction(func):
|
|
lines, start = inspect.getsourcelines(func)
|
|
|
|
fdef_index = 0
|
|
while fdef_index < len(lines) and not lines[fdef_index].strip().startswith(
|
|
"def"
|
|
):
|
|
fdef_index += 1
|
|
|
|
if fdef_index < len(lines):
|
|
line_num = start + fdef_index
|
|
|
|
fdef_content = lines[fdef_index].strip()
|
|
|
|
ERROR_HEADER = "\033[1m\033[91m[FATAL_EXECUTION_ERROR]: Non-Asynchronous Route Detected\033[0m"
|
|
|
|
ERROR_BODY = (
|
|
f"\n"
|
|
f'--> File "{inspect.getsourcefile(func)}", line {line_num}, in {func.__name__}\n\n'
|
|
f"\033[1m[CONSTRAINT_VIOLATION]\033[0m\n"
|
|
f" Synchronous function used where an async coroutine was required.\n"
|
|
f" \033[93mCode Traceback:\033[0m\n"
|
|
f" {line_num}: \033[93m{fdef_content}\033[0m\n\n"
|
|
f"\033[1m[SUGGESTED_PATCH]\033[0m\n"
|
|
f" Apply the 'async' keyword to the function signature:\n"
|
|
f" {line_num}: \033[92masync {fdef_content}\033[0m"
|
|
)
|
|
|
|
raise RuntimeError(ERROR_HEADER + ERROR_BODY)
|
|
|
|
else:
|
|
raise RuntimeError(
|
|
"\033[1m\033[91m[FATAL_EXECUTION_ERROR]: Non-Asynchronous Route Detected\033[0m"
|
|
)
|