_sync.py — anthropic-sdk-python Source File

Architecture documentation for _sync.py, a python file in the anthropic-sdk-python codebase. 7 imports, 1 dependents.

File python AnthropicClient SyncAPI 7 imports 1 dependents 2 functions

Entity Profile

AnthropicClient→ SyncAPI→ _sync.py — anthropic-sdk-python Source File

Dependency Diagram

graph LR
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83["_sync.py"]
  22e54b11_c61c_96dc_bf33_dfc7151750e9["asyncio"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> 22e54b11_c61c_96dc_bf33_dfc7151750e9
  4c73056b_526d_2199_4a80_7fcd0cd9bcbe["functools"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> 4c73056b_526d_2199_4a80_7fcd0cd9bcbe
  89ddcdd7_3ae1_4c7b_41bb_9f1e25f16875["typing"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> 89ddcdd7_3ae1_4c7b_41bb_9f1e25f16875
  37c05070_ca59_d596_7250_de9d1939227f["typing_extensions"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> 37c05070_ca59_d596_7250_de9d1939227f
  40396b02_9c68_4765_9e00_db68fce91470["anyio"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> 40396b02_9c68_4765_9e00_db68fce91470
  2f182a0f_b6d1_3b84_5c3a_c9cbb0568e75["sniffio"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> 2f182a0f_b6d1_3b84_5c3a_c9cbb0568e75
  da11bb5f_632d_41ba_dd79_3e11cea7cd3d["anyio.to_thread"]
  3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 --> da11bb5f_632d_41ba_dd79_3e11cea7cd3d
  6dadb144_3070_6111_357f_214554108905["__init__.py"]
  6dadb144_3070_6111_357f_214554108905 --> 3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83
  style 3d1d47e4_c6e3_27f4_3ae5_5b770d5eec83 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

from __future__ import annotations

import asyncio
import functools
from typing import TypeVar, Callable, Awaitable
from typing_extensions import ParamSpec

import anyio
import sniffio
import anyio.to_thread

T_Retval = TypeVar("T_Retval")
T_ParamSpec = ParamSpec("T_ParamSpec")


async def to_thread(
    func: Callable[T_ParamSpec, T_Retval], /, *args: T_ParamSpec.args, **kwargs: T_ParamSpec.kwargs
) -> T_Retval:
    if sniffio.current_async_library() == "asyncio":
        return await asyncio.to_thread(func, *args, **kwargs)

    return await anyio.to_thread.run_sync(
        functools.partial(func, *args, **kwargs),
    )


# inspired by `asyncer`, https://github.com/tiangolo/asyncer
def asyncify(function: Callable[T_ParamSpec, T_Retval]) -> Callable[T_ParamSpec, Awaitable[T_Retval]]:
    """
    Take a blocking function and create an async one that receives the same
    positional and keyword arguments.

    Usage:

    ```python
    def blocking_func(arg1, arg2, kwarg1=None):
        # blocking code
        return result


    result = asyncify(blocking_function)(arg1, arg2, kwarg1=value1)
    ```

    ## Arguments

    `function`: a blocking regular callable (e.g. a function)

    ## Return

    An async function that takes the same positional and keyword arguments as the
    original one, that when called runs the same original function in a thread worker
    and returns the result.
    """

    async def wrapper(*args: T_ParamSpec.args, **kwargs: T_ParamSpec.kwargs) -> T_Retval:
        return await to_thread(function, *args, **kwargs)

    return wrapper