I'm having a big problem trying to run this the way it's supposed to.
I have a store
function, which is supposed to simulate writing to a database. Currently I'm writing to a text file instead. To do that, I have a boolean which is used to introduce a delay.
import asyncio
import os
import time
async def store(text, delay: bool = False):
mode = "a" if os.path.exists(DB_FILE) else "w"
if delay:
await asyncio.sleep(5)
with open(DB_FILE, mode=mode) as f:
timestamp = str(time.time())
row = f"{text}\t{timestamp}\n"
f.write(row)
f.flush()
print("done saving")
I then have a function that simulates writing several changes to that database, and then gets the latest. If there is a delay in the last store function, then it wont count as "latest".
texts = [("how", False), ("are", False), ("you", True)]
async def run_tests():
initial_text = "hi"
await store(initial_text)
curr_text = initial_text
last_finished = initial_text
for text, delay in texts:
await store(text, delay=delay)
if not delay: # if finished
last_finished = curr_text
assert last_finished == "are"
if __name__ == "__main__":
asyncio.run(run_tests())
I tried running this with both asyncio.run()
and loop.run_until_complete()
, but neither is working. asyncio.sleep
still behaves like time.sleep
.
Ideally I want the line
await store(text, delay=delay)
to execute asynchronously while the rest of the function run_tests
keeps going.
What am i doing wrong?