Question: Support sqlalchemy AsyncSession ?

[Fred7b]

Sqlalchemy in 1.4 support asynchronous I/O. I would like to know about plans to support async 1.4+ sqlalchemy. Will this be implemented in the graphene-sqlalchemy?

[jendrikjoe]

I would be willing to help on this with a PR, if the maintainers are interested 😉
I experimented a bit with it and what worked for me is to just overwrite the get_node method. An MVP is attached (sorry for it being a bit messy, but I copied it from my test project and therefore it is in multiple files 😉)

The question I would have about a possible PR is, if the asyncio part should be its own subpackage, similar to how sqlalchemy handles it (https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html), or if another integration is preferable 😉 If the general structure is agreed, I would start with a draft PR 😉

class UserSql(Base):
    __tablename__ = "users"

    id = Column(String, primary_key=True)
    username = Column(String, nullable=False)


class User(SQLAlchemyObjectType):
    class Meta:
        model = UserSql
        interfaces = (relay.Node,)

    @classmethod
    async def get_node(cls, info, id):
        session = get_session(info.context)
        return await session.get(cls._meta.model, id)


class Query(graphene.ObjectType):
    node = relay.Node.Field()

example_schema = graphene.Schema(query=Query)

from sqlalchemy.ext.asyncio import AsyncSession, async_scoped_session, create_async_engine
from sqlalchemy.orm import sessionmaker


_engine = create_async_engine(settings.database_url)

class SessionProvider:
    def __init__(self, engine=None):
        if engine is None:
            engine = _engine
        self.engine = engine
        self._session_class = sessionmaker(bind=engine, class_=AsyncSession)

    def create_session(self, **kwargs) -> AsyncSession:
        return self._session_class(**kwargs)

    def create_scoped_session(self) -> AsyncSession:
        return async_scoped_session(self._session_class, scopefunc=get_request_id)


session_provider = SessionProvider(engine=_engine)

import pytest

from example_schema import UserSql, example_schema
from base import Base
from session import _engine, session_provider

@pytest.mark.asyncio
async def test_get_node(setup_example_db, session):
    global_id = to_global_id("User", 0)
    executed = await example_schema.execute_async(
        """
        query Node($id: ID!) {
            node(id: $id) {
                ... on User {
                    id
                }
            }
        }
        """,
        variable_values={"id": global_id},
        context_value={"session": session},
    )
    assert executed.errors is None
    user = executed.data["node"]
    invoice_id = from_global_id(user["id"])[1]

    assert invoice_id == str(0)

@pytest.fixture
def example_users():
    return [UserSql(id=0, username="karen"), UserSql(id=1, username="peter")]


@pytest.fixture
async def session():
    session = session_provider.create_session()
    yield session
    await session.close()


@pytest.fixture
async def setup_example_db(example_users, session):
    async with _engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    async with session:
        session.add_all(example_users)
        await session.commit()

[erikwrede]

The idea of implementing async sessions is great.
@jendrikjoe How would your MVP work with batch loading or relationships in general?
I'll look into this next week and follow up with a more detailed reply!

[jendrikjoe]

From what I found relationships should be fine, if not loaded lazily, which would conflict with this section of the sqlalchemy docs: https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession. So setting lazy in the relationships to "selectin" should keep everything working as is 😉
I can as well expand the tests to contain relationships and connections, but I would do that in a PR, so I am not cluttering this thrad 😆

[erikwrede]

We should clearly communicate the lack of support for certain lazy loading techniques then.
Additionally, I think a PR would require a more elegant solution than overriding the get_node classmethod for every type.
Maybe an option in the meta class (e.g. async= True/False) with a global default option would work?
This way we could implement the async functionality without using an extra subpackage - something I prefer since async mutations also implemented that way.

It seems like batching isn't working with sqa-1.4, so maybe that should be fixed first, since an async query should definitely be compatible with all the work done on the N+1 problem.
(#35 (comment))

Check batching.py, it seems to use a single session derived from query context.

[jendrikjoe]

I very much agree on all the point 👍 I will check if I can come up with something that moves this to the meta class 👍 I will check how the async mutations are done to figure, if I can draw inspiration from that. Same goes for the batching part 🙂

[erikwrede]

Hey, I've looked into it again.
If you want you can open a merge request with a proof of concept and we can start with that. This should change as little about things like batching as possible, while ensuring they still work. After that, things like adding the session maker (at the moment, graphene-SQA just uses a single session defined in the graphene-schema) should be taken care of. Looking forward to see what you come up with!

[Fred7b]

hey all. Thank You for response.
A few facts about what I did and what I use to solve problems in my project/

1. I use fastapi + starlette-graphene3
2. I fork into project dir. graphene-sqlalchemy, graphene_sqlalchemy_filter libs. And update logic in method
   get_query() for async
   form

@classmethod
    def get_query(cls, model, info, sort=None, **args):
        query = get_query(model, info.context)
        if sort is not None:
            if not isinstance(sort, list):
                sort = [sort]
            sort_args = []
            # ensure consistent handling of graphene Enums, enum values and
            # plain strings
            for item in sort:
                if isinstance(item, enum.Enum):
                    sort_args.append(item.value.value)
                elif isinstance(item, EnumValue):
                    sort_args.append(item.value)
                else:
                    sort_args.append(item)
            query = query.order_by(*sort_args)
        return query

to

@classmethod
    async def get_query(cls, model, info, sort=None, **args):
        query = None
        if sort is not None:
            if not isinstance(sort, list):
                sort = [sort]
            sort_args = []
            # ensure consistent handling of graphene Enums, enum values and
            # plain strings
            for item in sort:
                if isinstance(item, enum.Enum):
                    sort_args.append(item.value.value)
                elif isinstance(item, EnumValue):
                    sort_args.append(item.value)
                else:
                    sort_args.append(item)
            query = select(model).order_by(*sort_args)
        return query

and any updates in code lib.
I have plans to do mr. But I don't have enough time.

[erikwrede]

Implemented in #350, thanks to @jendrikjoe ☺️

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related topics referencing this issue.