Dealing with async methods in nacsos_data

The nacsos_data library is at the heart of the NACSOS backend server. In that scenario it makes sense to have all data-intensive functions asynchronous, as it allows the server to deal with other things (e.g. other requests) while it waits for the database. This however means, that most other utility functions are also asynchronous. If you are new to that concept, it can be confusing and annoying.

There is a simple rule of thumb to using it though. Just remember to put an await in front of every function marked with async.

In a jupyter notebook, this works as-is. In a script, you have to run all code in an asyncio context as in the example shown below.

import asyncio
import logging
from sqlalchemy.ext.asyncio import AsyncSession
from nacsos_data.db.crud.annotations import read_assignments_for_scope_for_user
from nacsos_data.db import get_engine_async

logging.basicConfig(format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', level=logging.INFO)
logger = logging.getLogger('my-script')
logger.setLevel(logging.DEBUG)


async def main():
    db_engine = get_engine_async(conf_file='/path/to/config/server.env')
    async with db_engine.session() as session:  # type: AsyncSession

        # do something ...

        # get assignments
        assignments = await read_assignments_for_scope_for_user(
            session,
            assignment_scope_id='some id',
            user_id='some user id',
            limit=20
        )

        # do more ...

if __name__ == '__main__':
    asyncio.run(main())

Using together with typer

Many scripts might use typer to handle command line arguments. Here, you just wrap everything in a separate main function like so:

import asyncio
import logging
from typing import Literal
import typer
from sqlalchemy.ext.asyncio import AsyncSession
from nacsos_data.db.crud.annotations import read_assignments_for_scope_for_user
from nacsos_data.db import get_engine_async

def main(scope_id: str, log_level: Literal['WARN', 'INFO', 'DEBUG'] = 'INFO'):
    logging.basicConfig(format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', level=log_level)
    logger = logging.getLogger('my-script')
    logger.setLevel(log_level)

    async def _main():
        db_engine = get_engine_async(conf_file='/path/to/config/server.env')
        async with db_engine.session() as session:  # type: AsyncSession

            # do something ...

            # get assignments
            assignments = await read_assignments_for_scope_for_user(
                session,
                assignment_scope_id=scope_id,
                user_id='some user id',
                limit=20
            )

            # do more ...

    asyncio.run(_main())

if __name__ == '__main__':
    typer.run(main)