Использование асинхронных функций для значений по умолчанию в 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 шт):
Думаю, сама идея рандомить значение из бд для установки его здесь как дефолтное не совсем верная.
default=CarService.get_random_location
Вы могли бы просто вычислять current_location_id
в момент создания записи в вашей таблице. default
правильнее было бы использовать, если ваше значение константное и предсказуемое.