Ordnungsgemäß ein Python -Dekorateur mit Paramspec und verkettet, das eine willkürliche Argumentpositionierung ermöglich ⇐ Python

[Anonymous]

Ich habe einen vorhandenen Python -Dekorateur, der sicherstellt, dass eine Methode eine Psycopg Asyncconnection -Instanz erhält. Ich versuche, die Eingabe zu aktualisieren, um ParamSpec und verkettet , da die aktuelle Implementierung nicht typeSafe ist, aber ich bleibe stecken.
Hier ist die aktuelle Implementierung:

Code: Select all

def ensure_conn(func: Callable[..., Coroutine[Any, Any, R]]) -> Callable[..., Coroutine[Any, Any, R]]:
"""Ensure the function has a conn argument. If conn is not provided, generate a new connection and pass it to the function."""

async def wrapper(*args: Any, **kwargs: Any) -> R:
# Get named keyword argument conn, or find an AsyncConnection in the args
kwargs_conn = kwargs.get("conn")
conn_arg: AsyncConnection[Any] | None = None
if isinstance(kwargs_conn, AsyncConnection):
conn_arg = kwargs_conn
elif not conn_arg:
for arg in args:
if isinstance(arg, AsyncConnection):
conn_arg = arg
break
if conn_arg:
# If conn is provided, call the method as is
return await func(*args, **kwargs)
else:
# If conn is not provided, generate a new connection and pass it to the method
db_driver = DbDriver()
async with db_driver.connection() as conn:
return await func(*args, **kwargs, conn=conn)

return wrapper
< /code>
Aktuelle Verwendung: < /p>
@ensure_conn
async def get_user(user_id: UUID, conn: AsyncConnection):
async with conn.cursor() as cursor:
// do stuff
< /code>
... Aber ich kann das nennen und es fällt nicht fehl, typeCing: < /p>
get_user('519766c5-af86-47ea-9fa9-cee0c0de66b1', conn, arg_that_should_fail_typing)

Hier ist die nächstgelegene aktuelle Implementierung, die ich mit ParamSpec und verkettet :
erhalten habe

Code: Select all

def ensure_conn_decorator[**P, R](func: Callable[Concatenate[AsyncConnection[Any], P], R]) -> Coroutine[Any, Any, R]:
"""Ensure the function has a conn argument. If conn is not provided, generate a new connection and pass it to the function."""
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
# Get named keyword argument conn, or find an AsyncConnection in the args
kwargs_conn = kwargs.get("conn")
conn_arg: AsyncConnection[Any] | None = None
if isinstance(kwargs_conn, AsyncConnection):
conn_arg = kwargs_conn
elif not conn_arg:
for arg in args:
if isinstance(arg, AsyncConnection):
conn_arg = arg
break
if conn_arg:
# If conn is provided, call the method as is
return await func(*args, **kwargs)
else:
# If conn is not provided, generate a new connection and pass it to the method
db_driver = DbDriver()
async with db_driver.connection() as conn:
return await func(*args, **kwargs, conn=conn)

return wrapper
< /code>
Probleme sind < /p>

 conn muss das erste Argument der Methode sein, anstatt eine willkürliche Position zu haben - es war normalerweise das letzte Argument, nachdem eine beliebige x -Anzahl von Args < /li>
 den richtigen Rückgabetyp < /li> < /< /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /ul> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /> < /ul> sein kann, sein.Expression of type "(**P@ensure_conn_decorator) -> Coroutine[Any, Any, R@ensure_conn_decorator]" is incompatible with return type "Coroutine[Any, Any, R@ensure_conn_decorator]"
"function" is incompatible with "Coroutine[Any, Any, R@ensure_conn_decorator]"