Fix async await syntax for dython

Author: RainingComputers

README.md

@@ -73,7 +73,7 @@ This project is still being worked on and is not ready for public release.
       count: int = 0
 
       while True:
-          sleep(1.0)
+          await sleep(1.0)
           count = count + 1
 
 
@@ -160,7 +160,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
 
 - **if / elif / else**: conditional branching, e.g. `if x > 0: ... elif x == 0: ... else: ...`
 
-- **while**: loop while a condition holds, including infinite loops, e.g. `while True: sleep(1.0)`
+- **while**: loop while a condition holds, including infinite loops, e.g. `while True: await sleep(1.0)`
 
 - **for**: iterate over a list, or use `enumerate` for index and value, e.g. `for i, num in enumerate(numbers): ...`
 
@@ -176,7 +176,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
   @flow
   async def handle():
       data: bytes | DSError
-      data = http_listen("POST", "/numbers")
+      data = await http_listen("POST", "/numbers")
       if not isinstance(data, bytes):
           return data
   ```
@@ -188,7 +188,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
 
   @flow
   async def delay():
-      sleep(1.0)
+      await sleep(1.0)
   ```
 
 - **http_listen**: suspend until an HTTP request hits the given route, returning its body
@@ -199,7 +199,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
   @flow
   async def listener():
       data: bytes | DSError
-      data = http_listen("POST", "/numbers")
+      data = await http_listen("POST", "/numbers")
   ```
 
 - **http_request**: make a durable outbound HTTP call
@@ -210,7 +210,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
   @flow
   async def fetch():
       resp: bytes | DSError
-      resp = http_request("GET", "http://localhost:8000/health", {}, b"")
+      resp = await http_request("GET", "http://localhost:8000/health", {}, b"")
   ```
 
 - **json_marshal and json_unmarshal**: encode to and decode from JSON bytes, decode takes the target type as a generic param
@@ -221,9 +221,9 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
   @flow
   async def codec(data: bytes):
       nums: list[int] | DSError
-      nums = json_unmarshal[list[int]](data)
+      nums = await json_unmarshal[list[int]](data)
       raw: bytes | DSError
-      raw = json_marshal([1, 2, 3])
+      raw = await json_marshal([1, 2, 3])
   ```
 
 - **pure function**: `@pure` marked deterministic functions that don't have any side effects, run inline without a checkpoint, e.g. `status = get_status(resp)` where:
@@ -236,7 +236,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
       return resp.Status
   ```
 
-- **impure function**: `@impure` marked side-effect functions whose result is run once and checkpointed, e.g. `insert_event(name)` where:
+- **impure function**: `@impure` marked side-effect functions whose result is run once and checkpointed, e.g. `await insert_event(name)` where:
 
   ```python
   import psycopg2
@@ -254,7 +254,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
 
   @flow
   async def record(name: str):
-      insert_event(name)
+      await insert_event(name)
   ```
 
 - **Flow call**: call another `@flow` function directly, it runs durably and returns its value
@@ -269,7 +269,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
 
   @flow
   async def caller():
-      total: int = adder(3, 5)
+      total: int = await adder(3, 5)
   ```
 
 - **create_task**: spawn a child flow that runs concurrently
@@ -279,7 +279,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
 
   @flow
   async def child(n: int):
-      sleep(1.0)
+      await sleep(1.0)
 
 
   @flow
@@ -311,7 +311,7 @@ You can think of a `@flow` function as a durable goroutine that can survive appl
 
   @flow
   async def task(wg: dpslio.WaitGroup):
-      sleep(1.0)
+      await sleep(1.0)
       await wg.done()
 
 

pkg/runtimes/dython/deepslate/imports.py

@@ -17,7 +17,7 @@
 class DeepslateAnnotated[**P, R]:
     def __init__(
         self,
-        func: Callable[P, R],
+        func: Callable[P, Any],
         metadata: FuncMetadata,
         impl: Callable[..., Any],
     ) -> None:
@@ -48,14 +48,21 @@ def pure[**P, R](func: Callable[P, R]) -> DeepslateAnnotated[P, R]:
     return _mark_pure(func)
 
 
-def impure[**P, R](func: Callable[P, R]) -> DeepslateAnnotated[P, R]:
+def impure[**P, R](
+    func: Callable[P, R],
+) -> DeepslateAnnotated[P, collections.abc.Coroutine[Any, Any, R]]:
     return _mark_impure(func)
 
 
 def block[**P, R](
     impl_fn: Callable[..., Any],
-) -> Callable[[Callable[P, R]], DeepslateAnnotated[P, R]]:
-    def decorator(stub_func: Callable[P, R]) -> DeepslateAnnotated[P, R]:
+) -> Callable[
+    [Callable[P, R]],
+    DeepslateAnnotated[P, collections.abc.Coroutine[Any, Any, R]],
+]:
+    def decorator(
+        stub_func: Callable[P, R],
+    ) -> DeepslateAnnotated[P, collections.abc.Coroutine[Any, Any, R]]:
         return _mark_block(impl_fn, stub_func)
 
     return decorator
@@ -70,13 +77,15 @@ def _mark_pure[**P, R](func: Callable[P, R]) -> DeepslateAnnotated[P, R]:
     return DeepslateAnnotated(func, _non_block_func_metadata(func, FuncType.PURE), func)
 
 
-def _mark_impure[**P, R](func: Callable[P, R]) -> DeepslateAnnotated[P, R]:
+def _mark_impure[**P, R](
+    func: Callable[P, R],
+) -> DeepslateAnnotated[P, collections.abc.Coroutine[Any, Any, R]]:
     return DeepslateAnnotated(func, _non_block_func_metadata(func, FuncType.IMPURE), func)
 
 
 def _mark_block[**P, R](
     impl_func: Callable[..., Any], stub_func: Callable[P, R]
-) -> DeepslateAnnotated[P, R]:
+) -> DeepslateAnnotated[P, collections.abc.Coroutine[Any, Any, R]]:
     return DeepslateAnnotated(
         stub_func, _block_func_metadata(impl_func, stub_func), impl=impl_func
     )

pkg/runtimes/dython/deepslate/tests/test_worker.py

@@ -32,11 +32,11 @@ async def counter():
     numbers: list[int]
 
     while True:
-        request_data = http_listen("POST", "/numbers")
+        request_data = await http_listen("POST", "/numbers")
         if not isinstance(request_data, bytes):
             return request_data
 
-        numbers = json_unmarshal[list[int]](request_data)
+        numbers = await json_unmarshal[list[int]](request_data)
 
         for idx, num in enumerate(numbers):
             if num % 2 == 0:
@@ -84,9 +84,9 @@ async def counter():
 async def sleep_test():
     counter: int = 0
     counter = 1
-    sleep(1.0)
+    await sleep(1.0)
     counter = 2
-    sleep(1.0)
+    await sleep(1.0)
     counter = 3
 """
             },
@@ -131,11 +131,11 @@ async def http_request_test():
     status: str = ""
     url: str = "http://localhost:18769/health"
 
-    response_data = http_request("GET", url, {}, b"")
+    response_data = await http_request("GET", url, {}, b"")
     if not isinstance(response_data, bytes):
         return response_data
 
-    resp = json_unmarshal[HealthResponse](response_data)
+    resp = await json_unmarshal[HealthResponse](response_data)
     status = get_status(resp)
 """,
             },
@@ -177,7 +177,7 @@ def get_health_status(url: str) -> str:
 @flow
 async def impure_http_request_test():
     status: str = ""
-    status = get_health_status("http://localhost:18769/health")
+    status = await get_health_status("http://localhost:18769/health")
 """,
             },
             steps=[
@@ -210,15 +210,15 @@ async def echo_request_test():
     value: int = 0
     url: str = "http://localhost:18769/echo"
 
-    body = json_marshal(42)
+    body = await json_marshal(42)
     if not isinstance(body, bytes):
         return body
 
-    response_data = http_request("POST", url, {}, body)
+    response_data = await http_request("POST", url, {}, body)
     if not isinstance(response_data, bytes):
         return response_data
 
-    value = json_unmarshal[int](response_data)
+    value = await json_unmarshal[int](response_data)
 """,
             },
             steps=[
@@ -266,7 +266,7 @@ def noop() -> None:
 @flow
 async def impure_void_test():
     x: int = 1
-    noop()
+    await noop()
     x = 2
 """,
             },
@@ -296,7 +296,7 @@ async def impure_void_test():
 @flow
 async def child_flow():
     request_data: bytes | DSError
-    request_data = http_listen("POST", "/child")
+    request_data = await http_listen("POST", "/child")
     if not isinstance(request_data, bytes):
         return request_data
 """,
@@ -337,7 +337,7 @@ async def parent_flow():
 @flow
 async def child_wg_flow_a(wg: dpslio.WaitGroup):
     request_data: bytes | DSError
-    request_data = http_listen("POST", "/child_wg_a")
+    request_data = await http_listen("POST", "/child_wg_a")
     if not isinstance(request_data, bytes):
         return request_data
     await wg.done()
@@ -350,7 +350,7 @@ async def child_wg_flow_a(wg: dpslio.WaitGroup):
 @flow
 async def child_wg_flow_b(wg: dpslio.WaitGroup):
     request_data: bytes | DSError
-    request_data = http_listen("POST", "/child_wg_b")
+    request_data = await http_listen("POST", "/child_wg_b")
     if not isinstance(request_data, bytes):
         return request_data
     await wg.done()
@@ -421,9 +421,9 @@ def format_route(value: int) -> str:
 @flow
 async def child_chan_flow(ch: dpslio.Queue[int], wg: dpslio.WaitGroup):
     value: int = await ch.get()
-    route: str = format_route(value)
+    route: str = await format_route(value)
     request_data: bytes | DSError
-    request_data = http_listen("GET", route)
+    request_data = await http_listen("GET", route)
     if not isinstance(request_data, bytes):
         return request_data
     await wg.done()
@@ -485,7 +485,7 @@ async def chan_parent_flow() -> None:
 
 @flow
 async def sel_child_flow(ch: dpslio.Queue[int], v: int, d: float, wg: dpslio.WaitGroup):
-    sleep(d)
+    await sleep(d)
     await ch.put(v)
     await wg.done()
 """,
@@ -535,7 +535,7 @@ async def select_parent_flow() -> None:
 async def child_mutex_flow(mu: dpslio.Lock, ch: dpslio.Queue[int]):
     await mu.acquire()
     request_data: bytes | DSError
-    request_data = http_listen("POST", "/mutex_child")
+    request_data = await http_listen("POST", "/mutex_child")
     if not isinstance(request_data, bytes):
         return request_data
     await ch.put(1)
@@ -601,7 +601,7 @@ async def adder(a: int, b: int) -> int:
 
 @flow
 async def caller_flow():
-    sum_val: int = adder(3, 5)
+    sum_val: int = await adder(3, 5)
 """,
             },
             steps=[
@@ -623,7 +623,7 @@ async def caller_flow():
 @flow
 async def noop(a: int, b: int) -> None:
     result: int = a + b
-    sleep(1.0)
+    await sleep(1.0)
 """,
                 "caller_flow.py": """\
 from deepslate import flow
@@ -633,7 +633,7 @@ async def noop(a: int, b: int) -> None:
 @flow
 async def caller_flow():
     finished: int = 0
-    noop(3, 5)
+    await noop(3, 5)
     finished = 1
 """,
             },
@@ -684,7 +684,7 @@ async def select_default_flow() -> None:
 
 @flow
 async def sel_child_flow(ch: dpslio.Queue[int], v: int, d: float, wg: dpslio.WaitGroup):
-    sleep(d)
+    await sleep(d)
     await ch.put(v)
     await wg.done()
 """,
@@ -733,10 +733,10 @@ async def select_second_parent_flow() -> None:
 async def suspend_child(url: str) -> bytes:
     resp: bytes | DSError = b""
     while True:
-        resp = http_request("GET", url, {}, b"")
+        resp = await http_request("GET", url, {}, b"")
         if isinstance(resp, bytes):
             return resp
-        sleep(2.0)
+        await sleep(2.0)
 """,
                 "join_caller.py": """\
 from deepslate import flow
@@ -746,7 +746,7 @@ async def suspend_child(url: str) -> bytes:
 @flow
 async def join_caller() -> None:
     url: str = "http://localhost:18769/health"
-    raw: bytes = suspend_child(url)
+    raw: bytes = await suspend_child(url)
 """,
             },
             steps=[

pkg/transpilers/dython/constructors.go

@@ -174,7 +174,7 @@ func funcCallImpure(
 	retVarName string,
 	line int,
 ) string {
-	callStr := funcCallPure(funcName, args, typeParam)
+	callStr := funcCallBlock(funcName, args, typeParam)
 
 	var stmt string
 	if retVarName != "" {

pkg/transpilers/dython/testutil_test.go

@@ -206,12 +206,14 @@ func runTranspilerErrorTest(t *testing.T, tt transpileErrorTestCase) {
 }
 
 func addPreludeImportOutput(code string) string {
-	return "from deepslate.prelude import *\nfrom deepslate import dpslio\n" +
+	return "from deepslate.prelude import *\n" +
+		"from deepslate import dpslio, pure, impure\n" +
 		code + testStubs()
 }
 
 func addPreludeImportInput(code string) string {
-	return "from deepslate.prelude import *\nfrom deepslate import dpslio\n\n\n" +
+	return "from deepslate.prelude import *\n" +
+		"from deepslate import dpslio, pure, impure\n\n\n" +
 		testStubs() +
 		"async def child_a() -> None:\n    pass\n\n\n" +
 		"async def child_b() -> None:\n    pass\n\n\n" +
@@ -220,10 +222,10 @@ func addPreludeImportInput(code string) string {
 
 func testStubs() string {
 	return "" +
-		"def pure_add(a: int, b: int) -> int:\n    return a + b\n\n\n" +
-		"def impure_log(msg: str) -> None:\n    pass\n\n\n" +
-		"def impure_read(path: str) -> str:\n    return path\n\n\n" +
-		"def child_ret() -> int:\n    return 0\n\n\n"
+		"@pure\ndef pure_add(a: int, b: int) -> int:\n    return a + b\n\n\n" +
+		"@impure\ndef impure_log(msg: str) -> None:\n    pass\n\n\n" +
+		"@impure\ndef impure_read(path: str) -> str:\n    return path\n\n\n" +
+		"async def child_ret() -> int:\n    return 0\n\n\n"
 }
 
 func mockImportMetadata(