feat: Automated SQLAlchemy integration

Author: KarolJagodzinski

AGENTS.md

@@ -53,6 +53,19 @@ hatch test benchmark --benchmark-storage=file://benchmark/results
 just coverage
 ```
 
+### Running Examples
+
+Use `uv` to run examples from the `examples/` directory. Refer to the docstrings within each example file for specific commands.
+Every example should include an example command of running a particular example with uvicorn.
+```bash
+# Run a basic query example
+uv run examples/basic_query_example.py
+
+# Run an ASGI example with uvicorn
+uv run --with "uvicorn[standard]" --with ariadne \
+    uvicorn examples.basic_query_example:app --reload
+```
+
 ## Code Style Requirements
 
 - **Python 3.10+** with type hints throughout
@@ -96,4 +109,6 @@ Follow [Conventional Commits](https://www.conventionalcommits.org/):
 2. Ensure test coverage meets the 90% minimum requirement
 3. Format code with `just fmt`
 4. Verify type hints with `just types`
-5. Write a clear commit message following the conventional commits format
+5. Ensure the documentation is up-to-date
+6. Write a clear commit message following the conventional commits format
+

README.md

@@ -42,6 +42,7 @@ Documentation is available [here](https://ariadnegraphql.org).
 - Loading schema from `.graphql`, `.gql`, and `.graphqls` files.
 - ASGI and WSGI support, with integrations for Django, FastAPI, Flask, and Starlette.
 - Opt-in automatic resolvers mapping between `camelCase123` and `snake_case_123`.
+- Automated integration with **SQLAlchemy 2.0** for zero-boilerplate resolvers and N+1 prevention.
 - [OpenTelemetry](https://opentelemetry.io/) extension for API monitoring.
 - Built-in [GraphiQL](https://github.com/graphql/graphiql) explorer for development and testing.
 - GraphQL syntax validation via `gql()` helper function.

ariadne/contrib/sqlalchemy/__init__.py

@@ -0,0 +1,18 @@
+try:
+    from .dataloaders import LoaderRegistry, SQLAlchemyRelationLoader
+    from .objects import SQLAlchemyObjectType
+    from .query import SQLAlchemyQueryType
+    from .utils import auto_eager_load
+except ImportError as ex:
+    raise ImportError(
+        "SQLAlchemy integration requires the 'sqlalchemy' and 'aiodataloader' "
+        "packages. Install them using 'pip install \"ariadne[sqlalchemy]\"'."
+    ) from ex
+
+__all__ = [
+    "LoaderRegistry",
+    "SQLAlchemyObjectType",
+    "SQLAlchemyQueryType",
+    "SQLAlchemyRelationLoader",
+    "auto_eager_load",
+]

ariadne/contrib/sqlalchemy/dataloaders.py

@@ -0,0 +1,111 @@
+import logging
+from collections import defaultdict
+from typing import Any
+
+from aiodataloader import DataLoader
+from sqlalchemy import select, tuple_
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import RelationshipProperty
+
+logger = logging.getLogger(__name__)
+
+
+class SQLAlchemyRelationLoader(DataLoader):
+    """
+    DataLoader for SQLAlchemy relationships supporting:
+    - Composite Keys
+    - Many-to-Many (secondary tables)
+    - Result grouping via SQL columns (optimized)
+    """
+
+    def __init__(
+        self,
+        session: AsyncSession,
+        relation_prop: RelationshipProperty,
+        cache: bool = True,
+    ):
+        super().__init__(cache=cache)
+        self.session = session
+        self.relation_prop = relation_prop
+        self.target_model = relation_prop.mapper.class_
+        self.is_list = relation_prop.uselist
+
+        # Identify local and remote columns (handles composite keys)
+        if relation_prop.secondary is not None:
+            self.local_cols = [lp.key for lp, rp in relation_prop.synchronize_pairs]
+            self.remote_cols = [rp.key for lp, rp in relation_prop.synchronize_pairs]
+        else:
+            self.local_cols = [c.key for c in relation_prop.local_columns]
+            self.remote_cols = [c.key for c in relation_prop.remote_side]
+
+        self.secondary = relation_prop.secondary
+
+    def get_query(self, keys: list[Any]):
+        """Builds query. Handles composite IN clause and M2M joins."""
+        target_model = self.target_model
+        stmt = select(target_model)
+
+        if self.secondary is not None:
+            stmt = stmt.join(self.secondary)
+            filter_cols = [self.secondary.c[k] for k in self.remote_cols]  # ty: ignore[invalid-argument-type]
+        else:
+            filter_cols = [getattr(target_model, k) for k in self.remote_cols]  # ty: ignore[invalid-argument-type]
+
+        # Add the filtering columns to the result to allow grouping
+        stmt = stmt.add_columns(*filter_cols)
+
+        if len(filter_cols) > 1:
+            stmt = stmt.where(tuple_(*filter_cols).in_(keys))
+        else:
+            # Flatten keys if they are single-element tuples
+            flat_keys = [k[0] if isinstance(k, (list, tuple)) else k for k in keys]
+            stmt = stmt.where(filter_cols[0].in_(flat_keys))
+
+        return stmt
+
+    async def batch_load_fn(self, keys: list[Any]) -> list[Any]:
+        logger.debug(
+            "SQLAlchemyRelationLoader: Fetching %s for %d parents",
+            self.target_model.__name__,
+            len(keys),
+        )
+        stmt = self.get_query(keys)
+        result = await self.session.execute(stmt)
+        rows = result.all()
+
+        num_filter_cols = len(self.remote_cols)
+        grouped = defaultdict(list)
+
+        for row in rows:
+            item = row[0]
+            # The filter columns are appended after the model instance
+            key_parts = row[1 : 1 + num_filter_cols]
+            key = tuple(key_parts) if num_filter_cols > 1 else key_parts[0]
+            grouped[key].append(item)
+
+        return [
+            grouped[k] if self.is_list else (grouped[k][0] if grouped[k] else None)
+            for k in keys
+        ]
+
+
+class LoaderRegistry:
+    """
+    Keeps one DataLoader instance per relationship per request.
+    """
+
+    def __init__(self, session: AsyncSession):
+        self.session = session
+        self._loaders: dict[
+            tuple[RelationshipProperty, type[DataLoader]], DataLoader
+        ] = {}
+
+    def get_loader(
+        self,
+        relation_prop: RelationshipProperty,
+        loader_class: type[SQLAlchemyRelationLoader] = SQLAlchemyRelationLoader,
+    ) -> DataLoader:
+        key = (relation_prop, loader_class)
+        if key not in self._loaders:
+            self._loaders[key] = loader_class(self.session, relation_prop)
+        return self._loaders[key]

ariadne/contrib/sqlalchemy/objects.py

@@ -0,0 +1,95 @@
+from collections.abc import Callable
+from typing import Any
+
+from graphql import GraphQLSchema
+from sqlalchemy import select
+from sqlalchemy.orm import DeclarativeBase, RelationshipProperty, class_mapper
+
+from ...objects import ObjectType
+from .dataloaders import LoaderRegistry, SQLAlchemyRelationLoader
+
+
+class SQLAlchemyObjectType(ObjectType):
+    """
+    ObjectType specialized for SQLAlchemy models.
+    Automatically binds resolvers for relationships using DataLoaders.
+    """
+
+    model: type[DeclarativeBase]
+    aliases: Any
+    strategies: dict[str, Any]
+    max_depth: int
+    loader_registry_key: str
+
+    def __init__(
+        self,
+        name: str,
+        model: type[DeclarativeBase],
+        *,
+        aliases: dict[str, str] | Callable[[], dict[str, str]] | None = None,
+        strategies: dict[str, Any] | None = None,
+        max_depth: int = 3,
+        loader_registry_key: str = "loader_registry",
+    ):
+        super().__init__(name)
+        self.model = model
+        self.aliases = aliases or {}
+        self.strategies = strategies or {}
+        self.max_depth = max_depth
+        self.loader_registry_key = loader_registry_key
+
+    def bind_to_schema(self, schema: GraphQLSchema) -> None:
+        self._bind_auto_resolvers()
+        super().bind_to_schema(schema)
+
+    def get_base_query(self, info: Any, **kwargs: Any):
+        """
+        Returns the base SQLAlchemy select statement for root queries.
+        Can be overridden to apply default filters.
+        """
+        return select(self.model)
+
+    def _bind_auto_resolvers(self):
+        mapper = class_mapper(self.model)
+
+        if callable(self.aliases):
+            self.aliases = self.aliases()
+
+        # Bind field aliases
+        for gql_field, db_attr in self.aliases.items():
+            if callable(db_attr):
+                self.set_field(gql_field, db_attr)
+            else:
+                self.set_field(
+                    gql_field, lambda obj, *_, _attr=db_attr: getattr(obj, _attr)
+                )
+
+        # Bind relationships
+        for relation in mapper.relationships:
+            if relation.key not in self._resolvers:
+                self.set_field(relation.key, self._create_relation_resolver(relation))
+
+    def _create_relation_resolver(self, relation: RelationshipProperty):
+        async def resolve(obj: Any, info: Any, **kwargs: Any):
+            # If the attribute is already loaded (e.g. via joinedload/selectinload),
+            # return it
+            if relation.key in obj.__dict__:
+                return getattr(obj, relation.key)
+
+            registry: LoaderRegistry = info.context.get(self.loader_registry_key)
+            if registry is None:
+                raise RuntimeError(
+                    "LoaderRegistry not found in context under key "
+                    f"'{self.loader_registry_key}'"
+                )
+
+            # Build the key for the loader
+            local_cols = [c.key for c in relation.local_columns]
+            key = tuple(getattr(obj, k) for k in local_cols)  # ty: ignore[invalid-argument-type]
+            if len(key) == 1:
+                key = key[0]
+
+            loader = registry.get_loader(relation, SQLAlchemyRelationLoader)
+            return await loader.load(key)
+
+        return resolve

ariadne/contrib/sqlalchemy/query.py

@@ -0,0 +1,84 @@
+from collections.abc import Sequence
+from typing import Any
+
+from graphql import GraphQLList, GraphQLNonNull, GraphQLObjectType, GraphQLSchema
+
+from ...objects import ObjectType
+from .objects import SQLAlchemyObjectType
+from .utils import auto_eager_load
+
+
+class SQLAlchemyQueryType(ObjectType):
+    """
+    A custom Query type that automatically binds SQLAlchemy resolvers
+    by inspecting the GraphQLSchema during the make_executable_schema build phase.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        object_types: Sequence[SQLAlchemyObjectType],
+        *,
+        session_key: str = "session",
+    ):
+        super().__init__(name)
+        self.object_types = {ot.name: ot for ot in object_types}
+        self.session_key = session_key
+
+    def bind_to_schema(self, schema: GraphQLSchema) -> None:
+        graphql_type = schema.type_map.get(self.name)
+        if not isinstance(graphql_type, GraphQLObjectType):
+            super().bind_to_schema(schema)
+            return
+
+        for field_name, field_def in graphql_type.fields.items():
+            is_list = False
+            unwrapped_type = field_def.type
+
+            while isinstance(unwrapped_type, (GraphQLList, GraphQLNonNull)):
+                if isinstance(unwrapped_type, GraphQLList):
+                    is_list = True
+                unwrapped_type = unwrapped_type.of_type
+
+            type_name = getattr(unwrapped_type, "name", None)
+
+            if type_name in self.object_types and field_name not in self._resolvers:
+                obj_type = self.object_types[type_name]
+                self.set_field(
+                    field_name, self._create_auto_resolver(obj_type, is_list)
+                )
+
+        super().bind_to_schema(schema)
+
+    def _create_auto_resolver(self, obj_type: SQLAlchemyObjectType, return_list: bool):
+        async def auto_resolve(obj: Any, info: Any, **kwargs: Any):
+            session = info.context.get(self.session_key)
+            if session is None:
+                raise RuntimeError(
+                    f"Session not found in context under key '{self.session_key}'"
+                )
+
+            model = obj_type.model
+            stmt = obj_type.get_base_query(info, **kwargs)
+
+            stmt = auto_eager_load(
+                stmt,
+                info,
+                model,
+                strategies=obj_type.strategies,
+                aliases=obj_type.aliases,
+                max_depth=obj_type.max_depth,
+            )
+
+            for key, value in kwargs.items():
+                db_col_name = obj_type.aliases.get(key, key)
+                if hasattr(model, db_col_name):
+                    stmt = stmt.where(getattr(model, db_col_name) == value)
+
+            result = await session.execute(stmt)
+
+            if return_list:
+                return result.scalars().unique().all()
+            return result.scalars().first()
+
+        return auto_resolve