import inspect
import logging
-from collections.abc import Callable, Coroutine
+from collections.abc import AsyncGenerator, Callable, Coroutine
from dataclasses import MISSING, dataclass
from datetime import datetime
from enum import Enum
command: str
signature: inspect.Signature
type_hints: dict[str, Any]
- target: Callable[..., Coroutine[Any, Any, Any]]
+ target: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]]
@classmethod
def parse(
- cls, command: str, func: Callable[..., Coroutine[Any, Any, Any]]
+ cls, command: str, func: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]]
) -> APICommandHandler:
"""Parse APICommandHandler by providing a function."""
type_hints = get_type_hints(func)
import os
import pathlib
import threading
-from collections.abc import Awaitable, Callable, Coroutine
+from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine
from typing import TYPE_CHECKING, Any, Self, TypeGuard, TypeVar, cast
from uuid import uuid4
def register_api_command(
self,
command: str,
- handler: Callable[..., Coroutine[Any, Any, Any]],
+ handler: Callable[..., Coroutine[Any, Any, Any] | AsyncGenerator[Any, Any]],
) -> Callable[[], None]:
"""
Dynamically register a command on the API.