Skip to main content

Toolkit for fast and easy strawberry+sqlalchemy apis

Project description

strawberry-alchemy

Batteries-included toolkit for building Strawberry GraphQL APIs backed by SQLAlchemy

CI PyPI version Python versions License: MIT


Source Code: https://github.com/Alteian/strawberry-alchemy


Features

Module What it does
QueryOptimizer Analyzes Strawberry selection sets and builds a single optimized SQLAlchemy query — automatic joinedload / selectinload, column deferral, annotation injection
FilterBuilder Translates Strawberry input types into SQLAlchemy WHERE clauses using a declarative operator system
Repository Generic async CRUD with hard-delete, dependent-map cascading, and lifecycle hooks
Types Relay Connection / Edge / PageInfo pagination, ListResult, BaseNodeType
Mapping Async helpers to convert SQLAlchemy instances to Strawberry types respecting the selected field tree
Permissions Protocol-based permission primitives: IsAuthenticated, RolePermission, OwnerPermission, ObjectAccessPermission, plus resolver pattern and resource-bag
Models Tiny SQLAlchemy DeclarativeBase with UUID primary key, timestamps, and automatic table naming
Utilities camel_to_snake, Ordering enum, common exceptions

Installation

pip install strawberry-alchemy
# or with uv
uv add strawberry-alchemy

Quick Start

1. Define your SQLAlchemy model

# models.py
import uuid
from sqlalchemy import ForeignKey, String
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from strawberry_alchemy.models import Base

class Post(Base):
    user_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("user.id"))
    title: Mapped[str] = mapped_column(String(255))
    body: Mapped[str]

    comments: Mapped[list["Comment"]] = relationship(back_populates="post", cascade="all, delete-orphan")

class Comment(Base):
    post_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("post.id"))
    body: Mapped[str]

    post: Mapped["Post"] = relationship(back_populates="comments")

2. Define an access filter (per-row security)

# access_filters.py
from strawberry_alchemy.filtering import AccessControlFilter

class PostAccessFilter(AccessControlFilter):
    model_class = Post
    # Default: all users can see all posts. Override to scope by user_id.

3. Define your GraphQL types

# types.py
import uuid
from typing import Annotated, ClassVar

import strawberry
from strawberry.types import Info

from strawberry_alchemy import BaseNodeType
from strawberry_alchemy.optimizer import AnnotateExists, optimize_field

@strawberry.type
class PostType(BaseNodeType):
    access_filter: ClassVar = PostAccessFilter()

    user_id: uuid.UUID | None = strawberry.UNSET
    title: str | None = strawberry.UNSET
    body: str | None = strawberry.UNSET
    comments: list[Annotated["CommentType", strawberry.lazy(".types")]] | None = strawberry.UNSET

    @strawberry.field
    @optimize_field(AnnotateExists("comments"))
    async def has_comments(self, info: Info) -> bool:
        return getattr(self, "_comments_exists", False)


@strawberry.type
class CommentType(BaseNodeType):
    access_filter: ClassVar = PostAccessFilter()

    post_id: uuid.UUID | None = strawberry.UNSET
    body: str | None = strawberry.UNSET
    post: Annotated["PostType", strawberry.lazy(".types")] | None = strawberry.UNSET

4. Define filter inputs

# filters.py
import strawberry
from strawberry_alchemy.filtering import IDFilter, StringFilter, DateTimeFilter

@strawberry.input
class PostFilter:
    AND: list["PostFilter"] | None = strawberry.UNSET
    OR: list["PostFilter"] | None = strawberry.UNSET
    id: IDFilter | None = strawberry.UNSET
    title: StringFilter | None = strawberry.UNSET
    body: StringFilter | None = strawberry.UNSET
    created_at: DateTimeFilter | None = strawberry.UNSET

5. Define a deletion handler (cascade deletes)

# deletion.py
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from strawberry_alchemy.repository import BaseDeletionHandler, DependentMap

class PostDeletionHandler(BaseDeletionHandler[Post]):
    async def collect_dependents(
        self, session: AsyncSession, entity_id: UUID, instance: Post
    ) -> DependentMap:
        result = await session.execute(select(Comment.id).where(Comment.post_id == entity_id))
        return {"comments": [row[0] for row in result.fetchall()]}

6. Define your repository and schema

# repositories.py
from pydantic import BaseModel
from strawberry_alchemy import BaseRepository

class PostSchema(BaseModel):
    id: uuid.UUID | None = None
    title: str
    body: str
    user_id: uuid.UUID
    model_config = {"from_attributes": True}

class PostRepository(BaseRepository[Post, PostSchema]):
    relation_models = {"comments": Comment}

    def __init__(self, session: AsyncSession, **kwargs):
        super().__init__(session, model_cls=Post, schema_cls=PostSchema, **kwargs)

7. Write your queries

# queries.py
import strawberry
from strawberry.relay import GlobalID
from strawberry.types import Info

from strawberry_alchemy import OptimizedListConnection, ListResult
from strawberry_alchemy.permissions import IsAuthenticated

@strawberry.type
class PostQueries:
    @strawberry.field(permission_classes=[IsAuthenticated])
    async def node(self, info: Info, id: GlobalID) -> PostType | None:
        return await PostType.resolve_node(node_id=id.node_id, info=info)

    @strawberry.field(permission_classes=[IsAuthenticated])
    async def list(
        self, info: Info,
        limit: int | None = None,
        offset: int | None = None,
        filters: PostFilter | None = strawberry.UNSET,
    ) -> ListResult[PostType]:
        return await PostType.resolve_list(info=info, limit=limit, offset=offset, filters=filters)

    @strawberry.field(permission_classes=[IsAuthenticated])
    async def connection(
        self, info: Info,
        after: str | None = None,
        first: int | None = None,
        filters: PostFilter | None = strawberry.UNSET,
    ) -> OptimizedListConnection[PostType]:
        return await PostType.resolve_connection(info=info, after=after, first=first, filters=filters)


@strawberry.type
class Query:
    @strawberry.field
    def posts(self) -> PostQueries:
        return PostQueries()

8. Write your mutations

# mutations.py
import uuid
import strawberry
from strawberry.types import Info
from strawberry.relay import GlobalID

from strawberry_alchemy.permissions import IsAuthenticated, OwnerPermission

@strawberry.input
class CreatePostInput:
    title: str
    body: str

@strawberry.input
class DeletePostInput:
    id: GlobalID

@strawberry.type
class PostMutations:
    @strawberry.mutation(permission_classes=[IsAuthenticated])
    async def create_post(self, info: Info, input: CreatePostInput) -> PostType:
        session = await info.context.get_session()
        user = await info.context.user
        schema = PostSchema(title=input.title, body=input.body, user_id=user.id)
        result = await PostRepository(session).create(schema=schema)
        return result.to_type(PostType)

    @strawberry.mutation(permission_classes=[IsAuthenticated, OwnerPermission])
    async def delete_post(self, info: Info, input: DeletePostInput) -> bool:
        session = await info.context.get_session()
        await PostRepository(
            session, deletion_handler=PostDeletionHandler()
        ).delete(id=uuid.UUID(input.node_id))
        return True


@strawberry.type
class Mutation:
    @strawberry.field
    def posts(self) -> PostMutations:
        return PostMutations()

9. Assemble the schema

# schema.py
import strawberry
from strawberry.schema.config import StrawberryConfig

schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    config=StrawberryConfig(auto_camel_case=True, relay_max_results=100),
)

The QueryOptimizer runs automatically behind the scenes — no N+1 queries, columns are deferred when not requested, and has_comments is resolved via a SQL EXISTS subquery instead of loading all comments.

Development

git clone https://github.com/Alteian/strawberry-alchemy.git
cd strawberry-alchemy
uv sync

# Lint & test
uv run ruff check .
uv run pytest -v

# Build
uv build

Contributing

See CONTRIBUTING.md for guidelines.

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

strawberry_alchemy-0.1.1.tar.gz (27.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

strawberry_alchemy-0.1.1-py3-none-any.whl (39.7 kB view details)

Uploaded Python 3

File details

Details for the file strawberry_alchemy-0.1.1.tar.gz.

File metadata

  • Download URL: strawberry_alchemy-0.1.1.tar.gz
  • Upload date:
  • Size: 27.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for strawberry_alchemy-0.1.1.tar.gz
Algorithm Hash digest
SHA256 49f92394572dfbc05fedef6bc37e2db2fa3914dfa932ececb3ed9cb2e193a2e4
MD5 f104e9c5d8d110df70e0b55bc6c48de3
BLAKE2b-256 0bec6b0f2aa0f0c0fce3fd1f66cedb49e9cf7cebc8a7ddf9a566788eb4583200

See more details on using hashes here.

Provenance

The following attestation bundles were made for strawberry_alchemy-0.1.1.tar.gz:

Publisher: release.yml on Alteian/strawberry-alchemy

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file strawberry_alchemy-0.1.1-py3-none-any.whl.

File metadata

File hashes

Hashes for strawberry_alchemy-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 3445d590e7c4729e70ad0eefd06ee838318a17502b46fba2126f6bc888969106
MD5 f8342463137f5292b7c544fb49dedc11
BLAKE2b-256 948927a2c7b0559d9c990af13ae14b65f500f90f6a3980c90ff3d2ad7cf5c9ed

See more details on using hashes here.

Provenance

The following attestation bundles were made for strawberry_alchemy-0.1.1-py3-none-any.whl:

Publisher: release.yml on Alteian/strawberry-alchemy

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page