aiosql

edgeql-queries provides an adapter for aiosql. Using it is quite simple and with its help, you can use aiosql and edgeql-queries together, while still being able to replace it with another adapter. You can use, for example, PostgreSQL with EdgeDB, since EdgeDB is based on PostgreSQL and it is possible that in the future, access to the PostgreSQL server that is used by EdgeDB will be open.

queries.edgeql:

# name: select-movie-by-id!
# Get movie by id.
SELECT Movie {
    title,
    year,
    director: {
        user_id,
    },
}
FILTER .id = <uuid>$0
LIMIT 1;

queries.sql:

-- name: get-username-by-id^
-- Get username by user's ID.
SELECT username
FROM users
WHERE id = :id;

And python code:

import asyncio
from uuid import UUID

import aiosql
import asyncpg
import edgedb
from edgeql_queries.contrib import aiosql as eq_aiosql

MOVIE_ID = UUID("6000323c-cc31-474d-828b-dadaa6404674")

pg_queries = aiosql.from_path("./aiosql_contrib.sql", "asyncpg")
edb_queries = aiosql.from_path(
    "./aiosql_contrib.edgeql",
    driver_adapter=eq_aiosql.EdgeQLAsyncAdapter,
    loader_cls=eq_aiosql.EdgeQLQueryLoader,
    queries_cls=eq_aiosql.EdgeQLQueries,
)


async def main():
    edb_conn = await edgedb.async_connect()
    pg_conn = await asyncpg.connect("postgres://postgres@localhost/postgres")

    movie = await edb_queries.select_movie_by_id(edb_conn, MOVIE_ID)
    user_record = await pg_queries.get_username_by_id(
        pg_conn,
        id=movie.director.user_id,
    )
    print(f"Director's name: {user_record[0]}")


asyncio.run(main())