Использование асинхронных функций для значений по умолчанию в SQLAlchemy

Я использую FastAPI, SQLAlchemy и asyncpg (асинхронное подключение к бд)

В моей модели есть колонка:

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

Функция get_random_location:

@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]

При попытке такого использования я получаю ошибку:

RuntimeWarning: coroutine 'CarService.get_random_location' was never awaited

Как я могу использовать асинхронную функцию для создания значения по умолчанию?

Я не могу использовать синхронную функцию и синхронный менеджер контекста, т.к. соединение с бд асинхронное.

Так же я пробовал использовать asyncio.run_coroutine_threadsafe, но это не сработало:

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

Ответы (1 шт):

Автор решения: Victor Egiazarian

Думаю, сама идея рандомить значение из бд для установки его здесь как дефолтное не совсем верная.

default=CarService.get_random_location

Вы могли бы просто вычислять current_location_id в момент создания записи в вашей таблице. default правильнее было бы использовать, если ваше значение константное и предсказуемое.

→ Ссылка