semlib.map

map async

map[T, U: BaseModel](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U],
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[U]

map[T, U](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: Bare[U],
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[U]

map[T](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: None = None,
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[str]

Standalone version of map.

Source code in src/semlib/map.py

async def map[T, U: BaseModel, V](  # noqa: A001
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U] | Bare[V] | None = None,
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[U] | list[V] | list[str]:
    """Standalone version of [map][semlib.map.Map.map]."""
    mapper = Map(model=model, max_concurrency=max_concurrency)
    return await mapper.map(iterable, template, return_type=return_type)

map_sync

map_sync[T, U: BaseModel](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U],
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[U]

map_sync[T, U](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: Bare[U],
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[U]

map_sync[T](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: None = None,
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[str]

Standalone synchronous version of map.

Source code in src/semlib/map.py

def map_sync[T, U: BaseModel, V](
    iterable: Iterable[T],
    /,
    template: str | Callable[[T], str],
    *,
    return_type: type[U] | Bare[V] | None = None,
    model: str | None = None,
    max_concurrency: int | None = None,
) -> list[U] | list[V] | list[str]:
    """Standalone synchronous version of [map][semlib.map.Map.map]."""
    mapper = Map(model=model, max_concurrency=max_concurrency)
    return asyncio.run(mapper.map(iterable, template, return_type=return_type))  # type: ignore[return-value]

:::