syncflex 0.1.0

Dead-simple interface for sync/async unification

syncflex

A tiny Python library that lets you write a single generator function and run it seamlessly in sync or async contexts.

It provides two decorators:

• syncify → turns a generator into a blocking synchronous function.

• asyncify → turns a generator into an awaitable coroutine function.

This allows you to share the same core generator logic in both sync and async code paths without duplication.

---

🚀 Installation

pip install syncflex

(or copy the ~30 lines of code straight into your project.)

---

✨ Motivation

Library authors often need to support both synchronous and asynchronous API.

Traditionally, that means a lot of duplicated code:

from typing import Any

import httpx


def pre_processing_code() -> Any: ...


def post_processing_code() -> Any: ...


def sync_api_call(http_client: httpx.Client) -> str:
    pre_processing_code()
    pre_processing_code()
    pre_processing_code()

    response = http_client.request(...)

    post_processing_code()
    post_processing_code()
    post_processing_code()

    return "hello"


async def async_api_call(http_client: httpx.AsyncClientClient) -> str:
    pre_processing_code()
    pre_processing_code()
    pre_processing_code()

    response = await http_client.request(...)

    post_processing_code()
    post_processing_code()
    post_processing_code()

    return "hello"

With syncflex, you can unify the implementation code and expose both sync and async API like so:

from collections.abc import Generator
from typing import Any

import httpx

from syncflex import asyncify, syncify


def pre_processing_code() -> Any: ...


def post_processing_code() -> Any: ...


def _base_api_call(http_client: httpx.Client | httpx.AsyncClient) -> Generator[Any, Any, str]:
    pre_processing_code()
    pre_processing_code()
    pre_processing_code()

    response: httpx.Response = yield http_client.request(...)

    post_processing_code()
    post_processing_code()
    post_processing_code()

    return "hello"


sync_api_call = syncify(_base_api_call)
async_api_call = asyncify(_base_api_call)

📖 Usage

Simple SDK

Below is an quick SDK implementation of JSONPlaceholder.

Without syncflex:

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Self

import httpx


@dataclass
class Post:
    id: int
    user_id: int
    title: str
    body: str

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> Self:
        return cls(id=data["id"], user_id=data["userId"], title=data["title"], body=data["body"])


class SyncSDK:
    def __init__(self, http_client: httpx.Client) -> None:
        self._http_client = http_client

    def get_post(self, id: int) -> Post:
        url = f"/posts/{id}"
        resp: httpx.Response = self._http_client.get(url)
        return Post.from_dict(resp.json())

    def list_posts(self) -> list[Post]:
        url = "/posts"
        resp: httpx.Response = self._http_client.get(url)
        return [Post.from_dict(data) for data in resp.json()]


class AsyncSDK:
    def __init__(self, http_client: httpx.AsyncClient) -> None:
        self._http_client = http_client

    async def get_post(self, id: int) -> Post:
        url = f"/posts/{id}"
        resp: httpx.Response = await self._http_client.get(url)
        return Post.from_dict(resp.json())

    async def list_posts(self) -> list[Post]:
        url = "/posts"
        resp: httpx.Response = await self._http_client.get(url)
        return [Post.from_dict(data) for data in resp.json()]

With syncflex:

from collections.abc import Generator, Mapping
from dataclasses import dataclass
from typing import Any, Self

import httpx

from syncflex import asyncify, syncify


@dataclass
class Post:
    id: int
    user_id: int
    title: str
    body: str

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> Self:
        return cls(id=data["id"], user_id=data["userId"], title=data["title"], body=data["body"])


class BaseSDK:
    def __init__(self, http_client: httpx.Client | httpx.AsyncClient) -> None:
        self._http_client = http_client

    def _get_post(self, id: int) -> Generator[Any, Any, Post]:
        url = f"/posts/{id}"
        # Yield on the part where it could be sync/async
        resp: httpx.Response = yield self._http_client.get(url)
        return Post.from_dict(resp.json())

    def _list_posts(self) -> Generator[Any, Any, list[Post]]:
        url = "/posts"
        # Yield on the part where it could be sync/async
        resp: httpx.Response = yield self._http_client.get(url)
        return [Post.from_dict(data) for data in resp.json()]


class SyncSDK(BaseSDK):
    get_post = syncify(BaseSDK._get_post)
    list_posts = syncify(BaseSDK._list_posts)


class AsyncSDK(BaseSDK):
    get_post = asyncify(BaseSDK._get_post)
    list_posts = asyncify(BaseSDK._list_posts)

See more examples in examples/

⚖️ License

MIT