How to pass database session to Ariadne in Fastapi

Question

I'm new to python, trying to learn and use Fastapi, Ariadne and SQLAlchemy. I'm following the docs and I'm stuck.

I have my dependency injection get_db() for normal REST requests, which provides a Session object, which I pass through few different modules from the request start until I actually do the db work, and honestly I don't get this design. But I left it there.

Then, another problem come up. How can I pass db to Ariadne GraphQL? Should I use context_value or are there any other options?

Answer 1

You need to create a session of SQLAlchemy from ariadne resolvers. And don't forget to close the connection after resolvers finished.

Let's say your database file like following,

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = os.getenv("DB_CONN_STRING")

engine = create_engine(SQLALCHEMY_DATABASE_URL)
LocalSession = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = None
    try:
        db = LocalSession()
        yield db
    finally:
        if db:
            db.close()

• You can do like following on the resolvers

# Import the local session from local database file
from app.database import LocalSession

db = LocalSession()
# Do whatever you need to do
db.close()

Answer 2

The only way I found to open and close SQLA sessions was by using the Ariadne Extensions feature. Overloading methods in the base extension class allow you to open a session when the request starts and close it when the request finishes.

# extension.py
from typing import Any

from ariadne.types import ContextValue, Extension, Resolver
from graphql import GraphQLResolveInfo

from myapp import db


class SQLASessionExtension(Extension):
    def request_started(self, context: ContextValue) -> None:
        self._session = db.Session()
        """Extension hook executed at request's start."""

    def request_finished(self, context: ContextValue) -> None:
        """Extension hook executed at request's end."""
        self._session.close()

    def resolve(
        self, next_: Resolver, obj: Any, info: GraphQLResolveInfo, **kwargs: Any
    ) -> Any:
        info.context["session"] = self._session
        return super().resolve(next_, obj, info, **kwargs)

Then you plug your extension as indicated in the docs

from ariadne.asgi import GraphQL
from ariadne.asgi.handlers import GraphQLHTTPHandler
from myapp.extensions import SQLASessionExtension

app = GraphQL(
    schema,
    http_handler=GraphQLHTTPHandler(
        extensions=[
            SQLASessionExtension
        ],
    ),
)

In your resolver, you get the session from the context

from typing import Any
from sqlalchemy.orm import Session
from graphql import GraphQLResolveInfo

from myapp.models import Flower

def resolve_flowers(_: Any, info: GraphQLResolveInfo) -> list[Flower]:
    session: Session = info.context["session"]
    return session.query(Flower).all()

The docs about Extensions https://ariadnegraphql.org/docs/extensions