Debugging Kubernetes

23 Jul 2024

Helpful tips for debugging applications running in k8s

build/k8s.png

Handling multiple errors in Rust iterator adapters

17 Dec 2023

Approaches for handling multiple errors within iterator adapters

build/rust.png

Better FastAPI Background Jobs

29 Aug 2022

A more featureful background task runner for Async apps like FastAPI or Discord bots

build/fastapi_logo.png

Useful Linux Examples

21 Dec 2021

A plethora of helpful tips for working in Linux

build/bash_logo.png
Continue to all blog posts

Don't async past me... Await!

Problem

The following code will (perhaps unexpectedly) let anyone access the functionality intended only for authenticated users

async def is_authenticated(user):
    if user == 'admin':
        return True
    return False

def do_stuff_that_requires_auth():
    print("I'm authenticated")

def main():
    user = 'not-an-admin'
    if not is_authenticated(user):
        print("You do not have permission for this")
        return
    do_stuff_that_requires_auth()

if __name__ == '__main__':
    main()

If we run it we’ll get:

<input>:11: RuntimeWarning: coroutine 'is_authenticated' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
I'm authenticated

At first glance it might not be obvious why this is, it’s because is_authenticated('anything') returns a coroutine object which is “true” i.e.

bool(is_authenticated('fake_user')) is True
<input>:1: RuntimeWarning: coroutine 'is_authenticated' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
True

Solution

The RuntimeWarning’s give a clue about how to fix it, we need to await the coroutine result, which in tern means that main() needs to become an async function too. Now main being async itself means that we need to call it in a different way:

import asyncio

async def is_authenticated(user):
    if user == 'admin':
        return True
    return False

def do_stuff_that_requires_auth():
    print("I'm authenticated")

async def main():
    user = 'not-an-admin'
    if not await is_authenticated(user):
        print("You do not have permission for this")
        return
    do_stuff_that_requires_auth()

if __name__ == '__main__':
    asyncio.run(main())
You do not have permission for this

See the Python Docs on Coroutines and Tasks for more details