Skip to content

Instantly share code, notes, and snippets.

@mwakaba2

mwakaba2/test_run_sync.py

Last active May 9, 2021 22:49
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save mwakaba2/e76b0bc341bf17c97ff58cd4c672e4d4 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save mwakaba2/e76b0bc341bf17c97ff58cd4c672e4d4 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. mwakaba2 revised this gist May 9, 2021. 1 changed file with 1 addition and 3 deletions.
    4 changes: 1 addition & 3 deletions test_run_sync.py
    View file
    Original file line number Diff line number Diff line change
    @@ -10,17 +10,15 @@

    async def test():
    await run_sync(os.mkdir, 'test1')
    await run_sync(os.mkdir, 'test2')


    async def main():
    await test()


    def clean_up():
    try:
    os.rmdir('test1')
    os.rmdir('test2')
    except FileNotFoundError:
    pass

  2. mwakaba2 revised this gist May 9, 2021. 1 changed file with 1 addition and 0 deletions.
    1 change: 1 addition & 0 deletions test_run_sync.py
    View file
    Original file line number Diff line number Diff line change
    @@ -7,6 +7,7 @@
    from anyio import run_sync_in_worker_thread as run_sync
    import asyncio


    async def test():
    await run_sync(os.mkdir, 'test1')
    await run_sync(os.mkdir, 'test2')
  3. mwakaba2 revised this gist May 9, 2021. 1 changed file with 4 additions and 2 deletions.
    6 changes: 4 additions & 2 deletions test_run_sync.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,11 +1,11 @@
    import os
    import asyncio

    try:
    from anyio.to_thread import run_sync
    except ImportError:
    # fallback on anyio v2 for python version < 3.7
    from anyio import run_sync_in_worker_thread as run_sync

    import asyncio

    async def test():
    await run_sync(os.mkdir, 'test1')
    @@ -15,13 +15,15 @@ async def test():
    async def main():
    await test()


    def clean_up():
    try:
    os.rmdir('test1')
    os.rmdir('test2')
    except FileNotFoundError:
    pass


    if __name__ == '__main__':
    clean_up()
    loop = asyncio.get_event_loop()
  4. mwakaba2 created this gist May 9, 2021.
    29 changes: 29 additions & 0 deletions test_run_sync.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,29 @@
    import os
    import asyncio
    try:
    from anyio.to_thread import run_sync
    except ImportError:
    # fallback on anyio v2 for python version < 3.7
    from anyio import run_sync_in_worker_thread as run_sync


    async def test():
    await run_sync(os.mkdir, 'test1')
    await run_sync(os.mkdir, 'test2')


    async def main():
    await test()

    def clean_up():
    try:
    os.rmdir('test1')
    os.rmdir('test2')
    except FileNotFoundError:
    pass

    if __name__ == '__main__':
    clean_up()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()