Use async db calls in WorkflowTrigger

[stevenschaerer]

This closes: #38672

I opted to slightly extend some existing unit tests rather than write mostly identical new ones. If the @sync_to_async decorator and the awaits before self._get_count are removed the tests will fail because the trigger and dummy tasks are executed sequentially.

Should I also remove TaskStateTrigger in this PR? It is not used anymore outside of tests.

[hussein-awala]

Looks good, LGTM

[uranusjr]

I don’t think we can just remove TaskStateTrigger, but let’s deprecate it for future removal.

It would also be a good idea to implement an version of _get_count to use the async SQLAlchemy API if available, and fallback to the sync_to_async version otherwise. Most people likely won’t have the async component (pip install sqlalchemy[asyncio]) but we should start to slowly add support to it in triggerers.

[hussein-awala]

I don’t think we can just remove TaskStateTrigger, but let’s deprecate it for future removal.

It would also be a good idea to implement an version of _get_count to use the async SQLAlchemy API if available, and fallback to the sync_to_async version otherwise. Most people likely won’t have the async component (pip install sqlalchemy[asyncio]) but we should start to slowly add support to it in triggerers.

I'm working on a draft AIP to add this support, I tested it a few months ago and the results were promising: #36504

[hussein-awala]

Also some async features are only available in sqlalchemy 2.0, so maybe we need to focus on this migration first.

[stevenschaerer]

I don’t think we can just remove TaskStateTrigger, but let’s deprecate it for future removal.

It would also be a good idea to implement an version of _get_count to use the async SQLAlchemy API if available, and fallback to the sync_to_async version otherwise. Most people likely won’t have the async component (pip install sqlalchemy[asyncio]) but we should start to slowly add support to it in triggerers.

I have deprecated TaskStateTrigger. Following the other comments I'll leave using async SQLAlchemy for a separate PR.