Add basic asyncio support #43944
Add basic asyncio support #43944
Conversation
dstandish
requested review from
potiuk,
ashb and
jedcunningham
as code owners
|
Is this related to AIP-70? |
| @@ -221,6 +221,7 @@ def test_get_documentation_package_path(): | |||
| """ | |||
| "apache-airflow-providers-common-sql>=1.17.0", | |||
| "apache-airflow>=2.8.0", | |||
| "asyncpg>=0.30.0", | |||
There was a problem hiding this comment.
psycopg 3 also has async support but might conflict with psycopg 2 already installed.
There was a problem hiding this comment.
psycopg 3 also has async support but might conflict with psycopg 2 already installed.
True, yeah. We could in future look at allowing diff libs.
There was a problem hiding this comment.
Benchmark, so take with a pinch of salt, but https://pypi.org/project/asyncpg/
In our testing asyncpg is, on average, 5x faster than psycopg3.
That's def a big enough of a change that we should see if we notice any difference
I think you prob know the answer. The AIP deals with asyncio, so yes strictly speaking related in that sense. But not "part of". That AIP has been in draft a long time not sure actively worked on. Anyway, I don't think what I'm doing here really requires an AIP. |
|
There is a similar draft PR open as POC for it though not active. So just wanted to confirm. Thanks for the details. |
|
@pierrejeambrun A FastAPI dep to give an async session would be good, but I think anything in the models etc should be passed an explicit session. |
| @@ -199,13 +209,25 @@ def load_policy_plugins(pm: pluggy.PluginManager): | |||
| pm.load_setuptools_entrypoints("airflow.policy") | |||
|
|
|||
|
|
|||
| def _get_async_conn_uri_from_sync(sync_uri): | |||
| scheme, rest = sync_uri.split(":", maxsplit=1) | |||
There was a problem hiding this comment.
SQLA has stuff built in to parse URLs/schemas. Maybe we should use that instead?
There was a problem hiding this comment.
| @@ -441,6 +463,9 @@ def configure_orm(disable_connection_pool=False, pool_class=None): | |||
|
|
|||
| global Session | |||
| global engine | |||
| global async_engine | |||
| global create_async_session | |||
There was a problem hiding this comment.
| global create_async_session |
There was a problem hiding this comment.
i think it's necessary ... it doesn't get set until configure_orm is called
|
NICE! Yeah. I really like to see it working with mysql compatibility and benchmarks 📦 (because of course MySQL is our beloved database) |
i would say let's wait and see how / where / when we need it |
dstandish
force-pushed
the
add-async-engine-and-session
branch
from
0af9d4d
to
620902e
Compare
|
Lovely! Thank you for adding this, will try to start using this soon :) |
dstandish
force-pushed
the
add-async-engine-and-session
branch
from
5765c6c
to
8b5ab77
Compare
This is meant to be sort of a hello world for asyncio support in airflow. It will be refined and extended in the future. But I think airflow ultimately really needs to go in this direction: in the new REST API, in the new AIP-72 internal API server, in triggers, and ultimately, in the scheduler.
dstandish
force-pushed
the
add-async-engine-and-session
branch
from
8b5ab77
to
6791e41
Compare
This is meant to be sort of a hello world for asyncio support in airflow. It will be refined and extended in the future. E.g. probably we would add more config flexibility re the libraries, connect args etc. But it's good to start simple.
Anyway, ultimately, I think Airflow really needs to go in this direction: in the new REST API, in the new AIP-72 internal API server, in triggers, and ultimately, in the scheduler.