Skip to content

Commit

Permalink
Using types.CoroutineType in coroutine decorators
Browse files Browse the repository at this point in the history
Required since pyright 1.1.394, see microsoft/pyright#9849
  • Loading branch information
mmmarcos committed Feb 21, 2025
1 parent c57186e commit f0a3e31
Showing 1 changed file with 8 additions and 7 deletions.
15 changes: 8 additions & 7 deletions server/parsec/components/postgresql/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@
import re
import traceback
from functools import wraps
from typing import Any, Callable, Coroutine, Iterable, Protocol, TypeVar, cast
from types import CoroutineType
from typing import Any, Callable, Iterable, Protocol, TypeVar, cast

from typing_extensions import Concatenate, ParamSpec

Expand Down Expand Up @@ -282,8 +283,8 @@ class WithPool(Protocol):


def transaction[**P, T, S: WithPool](
func: Callable[Concatenate[S, AsyncpgConnection, P], Coroutine[Any, Any, T]],
) -> Callable[Concatenate[S, P], Coroutine[Any, Any, T]]:
func: Callable[Concatenate[S, AsyncpgConnection, P], CoroutineType[Any, Any, T]],
) -> Callable[Concatenate[S, P], CoroutineType[Any, Any, T]]:
"""
This is used to decorate an API method that needs to be executed in a transaction.
Expand Down Expand Up @@ -312,8 +313,8 @@ async def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> T:


def no_transaction[**P, T, S: WithPool](
func: Callable[Concatenate[S, AsyncpgConnection, P], Coroutine[Any, Any, T]],
) -> Callable[Concatenate[S, P], Coroutine[Any, Any, T]]:
func: Callable[Concatenate[S, AsyncpgConnection, P], CoroutineType[Any, Any, T]],
) -> Callable[Concatenate[S, P], CoroutineType[Any, Any, T]]:
"""
This is used to decorate an API method that needs to request the database
without transaction (i.e. for query that doesn't do write operations).
Expand All @@ -333,8 +334,8 @@ class RetryNeeded(BaseException):


def retryable[S, **P, T](
func: Callable[Concatenate[S, P], Coroutine[Any, Any, T]],
) -> Callable[Concatenate[S, P], Coroutine[Any, Any, T]]:
func: Callable[Concatenate[S, P], CoroutineType[Any, Any, T]],
) -> Callable[Concatenate[S, P], CoroutineType[Any, Any, T]]:
@wraps(func)
async def wrapper(self: S, *args: P.args, **kwargs: P.kwargs) -> T:
while True:
Expand Down

0 comments on commit f0a3e31

Please sign in to comment.