wehub-resource-sync
97e91a83f3
Ruff / Ruff (push) Has been cancelled
Test / Core Tests (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.10) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.11) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.12) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.13) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.9) (push) Has been cancelled
Test / Full Coverage (Python 3.11) (push) Has been cancelled
Test / Core Provider Tests (OpenAI) (push) Has been cancelled
Test / Core Provider Tests (Anthropic) (push) Has been cancelled
Test / Core Provider Tests (Google) (push) Has been cancelled
Test / Core Provider Tests (Other) (push) Has been cancelled
Test / Anthropic Tests (push) Has been cancelled
Test / Gemini Tests (push) Has been cancelled
Test / Google GenAI Tests (push) Has been cancelled
Test / Vertex AI Tests (push) Has been cancelled
Test / OpenAI Tests (push) Has been cancelled
Test / Writer Tests (push) Has been cancelled
Test / Auto Client Tests (push) Has been cancelled
ty / type-check (push) Has been cancelled
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""Small generic helpers owned by the v2 runtime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
from collections.abc import Callable
|
|
from typing import Any, Generic, TypeVar, cast
|
|
|
|
from pydantic import ValidationError
|
|
|
|
R_co = TypeVar("R_co", covariant=True)
|
|
_validation_error_original_str: Callable[[ValidationError], str] | None = None
|
|
|
|
|
|
def is_async(func: Callable[..., Any]) -> bool:
|
|
"""Return whether a callable is async, following wrapped callables."""
|
|
is_coroutine = inspect.iscoroutinefunction(func)
|
|
while callable(wrapped := getattr(func, "__wrapped__", None)):
|
|
func = wrapped
|
|
is_coroutine = is_coroutine or inspect.iscoroutinefunction(func)
|
|
return is_coroutine
|
|
|
|
|
|
class classproperty(Generic[R_co]):
|
|
"""Descriptor for class-level properties."""
|
|
|
|
def __init__(self, method: Callable[[Any], R_co]) -> None:
|
|
self.cproperty = method
|
|
|
|
def __get__(self, _instance: object, cls: type[Any]) -> R_co:
|
|
return self.cproperty(cls)
|
|
|
|
|
|
def disable_pydantic_error_url() -> None:
|
|
"""Disable URLs in Pydantic ValidationError messages."""
|
|
global _validation_error_original_str
|
|
if _validation_error_original_str is None:
|
|
_validation_error_original_str = ValidationError.__str__
|
|
|
|
original_str = _validation_error_original_str
|
|
|
|
def __str__(self: ValidationError) -> str:
|
|
output = original_str(self)
|
|
return "\n".join(
|
|
line
|
|
for line in output.split("\n")
|
|
if "https://errors.pydantic.dev" not in line
|
|
)
|
|
|
|
cast(Any, ValidationError).__str__ = __str__
|