I’m using FastAPI, SQLAlchemy and asyncpg.

I have a column in my model:

current_location_id = Column(
    ForeignKey("location.id"),
    default=CarService.get_random_location
)

The get_random_location function:

@classmethod
async def get_random_location(cls) -> int:
    async with async_session_maker() as session:
        query = select(Location.id)
        instances = await session.execute(query)
        result = await instances.fetchall()

    return random.choice(result)[0]

When trying to use it this way, I get the error:

RuntimeWarning: coroutine 'CarService.get_random_location' was never awaited

How can I use an asynchronous function to generate a default value?

I cannot use a synchronous function and a synchronous context manager because the database connection is asynchronous.

I tried using asyncio.run_coroutine_threadsafe to run the asynchronous function get_random_location, but it didn’t work:

current_location_id = Column(
    ForeignKey("location.id"),
    default=asyncio.run_coroutine_threadsafe(
        CarService.get_random_location(),
        loop=asyncio.get_event_loop()
    )
)