<reponame>raynardj/forgebox<gh_stars>1-10
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/03_async.ipynb (unless otherwise specified).

__all__ = ['Async']

# Cell
import asyncio
from typing import Callable
def Async(func: Callable):
    """
    JS style async tool
    def download_file(url):
        ...
        return path

    download_async = Async(download_file)

    download_async(url)\
    .then(lambda x: os.system(f"tar -xvf {path}"))
    .catch(logging.error)

    """
    class AsyncFunc:
        f"""{func.__doc__}"""
        def __init__(self):
            self.success=[]

        async def afunc(self, *args, **kwargs):
            try:
                result = func(*args, **kwargs)
                for success_func in self.success:
                    result = success_func(result)
            except Exception as e:
                self.catch_func(e)

        def __repr__(self):
            return f"async function [{func}]"

        def __call__(self, *args, **kwargs):
            f"{func.__doc__}"
            self.args = args
            self.kwargs = kwargs
            self.success = []
            return self

        def then(self, success):
            self.success.append(success)
            return self

        def catch(self, catch):
            assert len(self.success)>0,\
                "You have to define self.then first"
            self.catch_func = catch
            asyncio.gather(self.afunc(*self.args, **self.kwargs))

    return AsyncFunc()
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/03_async.ipynb (unless otherwise specified).

__all__ = ['Async']

# Cell
import asyncio
from typing import Callable
def Async(func: Callable):
    """
    JS style async tool
    def download_file(url):
        ...
        return path

    download_async = Async(download_file)

    download_async(url)\
    .then(lambda x: os.system(f"tar -xvf {path}"))
    .catch(logging.error)

    """
    class AsyncFunc:
        f"""{func.__doc__}"""
        def __init__(self):
            self.success=[]

        async def afunc(self, *args, **kwargs):
            try:
                result = func(*args, **kwargs)
                for success_func in self.success:
                    result = success_func(result)
            except Exception as e:
                self.catch_func(e)

        def __repr__(self):
            return f"async function [{func}]"

        def __call__(self, *args, **kwargs):
            f"{func.__doc__}"
            self.args = args
            self.kwargs = kwargs
            self.success = []
            return self

        def then(self, success):
            self.success.append(success)
            return self

        def catch(self, catch):
            assert len(self.success)>0,\
                "You have to define self.then first"
            self.catch_func = catch
            asyncio.gather(self.afunc(*self.args, **self.kwargs))

    return AsyncFunc()
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/03_async.ipynb (unless otherwise specified). # Cell JS style async tool     def download_file(url):         ...         return path      download_async = Async(download_file)      download_async(url)\     .then(lambda x: os.system(f"tar -xvf {path}"))     .catch(logging.error) {func.__doc__}