Created
June 13, 2025 00:46
-
-
Save joshuadavidthomas/b55e3245b11338e8f992ee12a4b21573 to your computer and use it in GitHub Desktop.
Async compatible wrapper around typer
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from __future__ import annotations | |
| import asyncio | |
| import inspect | |
| from functools import wraps | |
| from typing import Any | |
| from typing import Protocol | |
| from typing import cast | |
| from typing import override | |
| from typer import Typer | |
| from typer.models import CommandFunctionType | |
| class CommandDecorator(Protocol): | |
| """A function that decorates a Typer command.""" | |
| def __call__(self, __func: CommandFunctionType, /) -> CommandFunctionType: ... | |
| class ATyper(Typer): | |
| """Extended Typer class that supports async functions in commands and callbacks.""" | |
| @override | |
| def callback(self, **kwargs: Any) -> CommandDecorator: | |
| """Override callback to support async functions.""" | |
| decorator = super().callback(**kwargs) | |
| return self._async_wrap_decorator(decorator) | |
| @override | |
| def command(self, name: str | None = None, **kwargs: Any) -> CommandDecorator: | |
| """Override command to support async functions.""" | |
| decorator = super().command(name, **kwargs) | |
| return self._async_wrap_decorator(decorator) | |
| def _async_wrap_decorator(self, decorator: CommandDecorator) -> CommandDecorator: | |
| """Wrap a decorator to make it async-aware.""" | |
| def wrapper(func: CommandFunctionType) -> CommandFunctionType: | |
| return async_me_maybe(decorator, func) | |
| return cast(CommandDecorator, wrapper) | |
| def async_me_maybe( | |
| decorator: CommandDecorator, | |
| func: CommandFunctionType, | |
| ) -> CommandFunctionType: | |
| """Wrap async functions with asyncio.run.""" | |
| if inspect.iscoroutinefunction(func): | |
| @wraps(func) | |
| def runner(*args: object, **kwargs: object) -> object: | |
| result: object = asyncio.run(func(*args, **kwargs)) | |
| return result | |
| return decorator(cast(CommandFunctionType, runner)) | |
| else: | |
| return decorator(func) |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This started out as copying code out of a typer issue about async compat. I spiced it up a bit to make it mostly work with type checkers.