Skip to content

Fix: make python 3.12 type X = ... syntax work with dependency injection #2951

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
StefanBRas wants to merge 5 commits into
base: main
Choose a base branch
from
+138 −37
Open

Fix: make python 3.12 type X = ... syntax work with dependency injection #2951

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2c9703d
fix: work with type syntax
StefanBRas Feb 5, 2026
9eb262f
Merge branch 'main' into sras/dependecy-error-type-syntax
StefanBRas Feb 9, 2026
7e50413
fix: one more instance of asyncio.iscoroutinefunction
StefanBRas Feb 11, 2026
80de12a
Merge branch 'main' into sras/dependecy-error-type-syntax
StefanBRas Feb 11, 2026
7846971
test: add tests for type alias with type syntax
StefanBRas Feb 17, 2026
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
7 changes: 5 additions & 2 deletions sdks/python/examples/dependency_injection/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from contextlib import asynccontextmanager, contextmanager
from typing import Annotated, AsyncGenerator, Generator
from typing import Annotated, AsyncGenerator, Generator, TypeAlias

from pydantic import BaseModel

Expand Down Expand Up @@ -88,12 +88,15 @@ class Output(BaseModel):
chained_async_dep: str


type AsyncDep = Annotated[str, Depends(async_dep)]
Copy link

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using PEP 695 type AsyncDep = ... syntax makes this example file a SyntaxError on Python < 3.12. If the SDK/examples are intended to run on older supported Python versions, switch to a backwards-compatible alias (e.g., assignment-based alias, optionally with TypeAlias) instead of the type statement, or gate the example so it’s only executed on 3.12+.

Suggested change
type AsyncDep = Annotated[str, Depends(async_dep)]
AsyncDep: TypeAlias = Annotated[str, Depends(async_dep)]

Copilot uses AI. Check for mistakes.
Copy link
Contributor

@mrkaye97 mrkaye97 Feb 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@StefanBRas it'd be worth testing this on 3.10 and 3.11 to confirm, because we'll need to handle this case if this is correct

Copy link
Contributor Author

@StefanBRas StefanBRas Feb 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can handle TypeAlias on any version as well, but I wasn't sure how you want to organize the tests since the type X = ... is a syntax error on <3.12 so it is not even possible to import a module with the syntax.

I'm thinking to just add a module with the type that is then conditionally imported depending on the version.
Do you want a test that is just skipped with pytest skipped if version is <3.12 or just that the test only uses type syntax if >=3.12?

Copy link
Contributor

@mrkaye97 mrkaye97 Feb 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we have this in utils/typing.py - something similar might help here too?

if sys.version_info >= (3, 12):
    AwaitableLike: TypeAlias = Awaitable[_T_co]
    CoroutineLike: TypeAlias = Coroutine[Any, Any, _T_co]
else:
    AwaitableLike: TypeAlias = Generator[Any, None, _T_co] | Awaitable[_T_co]
    CoroutineLike: TypeAlias = Generator[Any, None, _T_co] | Coroutine[Any, Any, _T_co]

Copy link
Contributor Author

@StefanBRas StefanBRas Feb 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunately that's not enough:

~/tmp/test_type$ cat test.py
import sys
from typing import Any, Awaitable, Coroutine, TypeAlias, Generator

if sys.version_info >= (3, 12):
    type AwaitableLike = Awaitable[Any]
    type CoroutineLike = Coroutine[Any, Any, Any]
else:
    AwaitableLike: TypeAlias = Generator[Any, None, Any] | Awaitable[Any]
    CoroutineLike: TypeAlias = Generator[Any, None, Any] | Coroutine[Any, Any, Any]

print("No errors")
~/tmp/test_type$ uv run --python 3.12 python test.py
No errors
~/tmp/test_type$ uv run --python 3.10 python test.py
  File "/home/sras/tmp/test_type/test.py", line 5
    type AwaitableLike = Awaitable[Any]
         ^^^^^^^^^^^^^
SyntaxError: invalid syntax

The syntax can't be in the file at all, not even in a class scope.

Maybe it's possible to construct the TypeAliasType object itself without the syntax, but that feels more brittle.

My idea is to have something like this:

~/tmp/test_type$ cat test_split.py
import sys
from typing import Any, Awaitable, Coroutine

if sys.version_info >= (3, 12):
    from test_312 import AwaitableLike, CoroutineLike
else:
    from test_310 import AwaitableLike, CoroutineLike

print("No errors")
~/tmp/test_type$ cat test_310.py
from typing import Any, Awaitable, Coroutine, Generator, TypeAlias

AwaitableLike: TypeAlias = Generator[Any, None, Any] | Awaitable[Any]
CoroutineLike: TypeAlias = Generator[Any, None, Any] | Coroutine[Any, Any, Any]
~/tmp/test_type$ cat test_312.py
from typing import Any, Awaitable, Coroutine

type AwaitableLike = Awaitable[Any]
type CoroutineLike = Coroutine[Any, Any, Any]
~/tmp/test_type$ uv run --python 3.12 python test_split.py
No errors
~/tmp/test_type$ uv run --python 3.10 python test_split.py
No errors
~/tmp/test_type$

Copy link
Contributor Author

@StefanBRas StefanBRas Feb 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mrkaye97 I put in a test now that does that split.
I saw that the example file was used in the docs too, so I put it such that it didn't interfere with that.
That also means that it probably could be refactored a bit to be nicer, but I think it's okay as is



# > Inject dependencies
@hatchet.task()
async def async_task_with_dependencies(
_i: EmptyModel,
ctx: Context,
async_dep: Annotated[str, Depends(async_dep)],
async_dep: AsyncDep,
Copy link

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using PEP 695 type AsyncDep = ... syntax makes this example file a SyntaxError on Python < 3.12. If the SDK/examples are intended to run on older supported Python versions, switch to a backwards-compatible alias (e.g., assignment-based alias, optionally with TypeAlias) instead of the type statement, or gate the example so it’s only executed on 3.12+.

Copilot uses AI. Check for mistakes.
sync_dep: Annotated[str, Depends(sync_dep)],
async_cm_dep: Annotated[str, Depends(async_cm_dep)],
sync_cm_dep: Annotated[str, Depends(sync_cm_dep)],
Expand Down
3 changes: 3 additions & 0 deletions sdks/python/hatchet_sdk/runnables/task.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
get_origin,
get_type_hints,
)
from typing_extensions import TypeAliasType
Copy link

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The runtime class for type X = ... in Python 3.12 is typing.TypeAliasType. Depending on how typing_extensions.TypeAliasType is implemented in the installed version, isinstance(annotation, TypeAliasType) can be brittle across environments (e.g., if it’s not the same class object). Prefer importing TypeAliasType from typing when available with a fallback to typing_extensions, or use a small compatibility check that handles both stdlib and typing_extensions implementations.

Suggested change
from typing_extensions import TypeAliasType
try:
from typing import TypeAliasType # type: ignore[attr-defined]
except ImportError: # pragma: no cover - fallback for older Python versions
from typing_extensions import TypeAliasType

Copilot uses AI. Check for mistakes.

from pydantic import BaseModel, ConfigDict, TypeAdapter

Expand Down Expand Up @@ -269,6 +270,8 @@ async def _parse_parameter(
) = None,
) -> DependencyToInject | None:
annotation = param.annotation
if isinstance(annotation, TypeAliasType):
annotation = annotation.__value__
Comment on lines +273 to +274
Copy link

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The runtime class for type X = ... in Python 3.12 is typing.TypeAliasType. Depending on how typing_extensions.TypeAliasType is implemented in the installed version, isinstance(annotation, TypeAliasType) can be brittle across environments (e.g., if it’s not the same class object). Prefer importing TypeAliasType from typing when available with a fallback to typing_extensions, or use a small compatibility check that handles both stdlib and typing_extensions implementations.

Copilot uses AI. Check for mistakes.
Copy link
Contributor Author

@StefanBRas StefanBRas Feb 17, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wrong, type_extensions just imports from typing if it's available.


if get_origin(annotation) is Annotated:
args = get_args(annotation)
Expand Down
2 changes: 1 addition & 1 deletion sdks/python/hatchet_sdk/runnables/types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ class StepType(str, Enum):
def is_async_fn(
fn: TaskFunc[TWorkflowInput, R],
) -> TypeGuard[AsyncFunc[TWorkflowInput, R]]:
return asyncio.iscoroutinefunction(fn)
return inspect.iscoroutinefunction(fn)
Copy link

Copilot AI Feb 11, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

asyncio.iscoroutinefunction() and inspect.iscoroutinefunction() are not equivalent: asyncio.iscoroutinefunction() also recognizes some asyncio-specific coroutine markers (e.g., generator-based coroutines / functions marked via _is_coroutine). Replacing it with inspect.iscoroutinefunction() can regress detection for those callables. Consider reverting to asyncio.iscoroutinefunction(fn) (or otherwise preserving asyncio’s extended detection) unless you explicitly want to drop support for those cases.

Copilot uses AI. Check for mistakes.


def is_sync_fn(
Expand Down
Loading