A library for creating GraphQL APIs using dataclasses and type annotations with extensive framework integration support.
Complete Relay specification implementation with Node interface, connections, pagination, and global object identification. The Relay specification provides standards for cursor-based pagination, global object identification, and mutations.
Global object identification following the Relay Node interface pattern.
class Node:
"""Relay Node interface for global object identification."""
id: NodeID
class NodeID:
"""Node ID type with automatic encoding/decoding."""
def __init__(self, node_id: str, type_name: str = None): ...
class GlobalID:
"""Global ID implementation with type and ID encoding."""
def __init__(self, type_name: str, node_id: str): ...
@classmethod
def from_id(cls, global_id: str) -> "GlobalID": ...
def to_id(self) -> str: ...
class GlobalIDValueError(ValueError):
"""Exception raised when GlobalID parsing fails."""Usage Example:
@strawberry.type
class User(strawberry.relay.Node):
id: strawberry.relay.NodeID
name: str
email: str
@classmethod
def resolve_node(cls, node_id: str, info: strawberry.Info):
"""Resolve node from global ID."""
user_data = get_user_by_id(node_id)
return cls(
id=strawberry.relay.NodeID(node_id, "User"),
name=user_data["name"],
email=user_data["email"]
)
@strawberry.type
class Post(strawberry.relay.Node):
id: strawberry.relay.NodeID
title: str
content: str
@classmethod
def resolve_node(cls, node_id: str, info: strawberry.Info):
post_data = get_post_by_id(node_id)
return cls(
id=strawberry.relay.NodeID(node_id, "Post"),
title=post_data["title"],
content=post_data["content"]
)
@strawberry.type
class Query:
# Relay requires a node field on Query
node: strawberry.relay.Node = strawberry.relay.node()Decorator to mark types as Relay nodes with automatic node resolution.
def node(
cls=None,
*,
node_resolver: Callable = None
) -> Any:
"""
Decorator to mark types as Relay nodes.
Args:
cls: The class to mark as a node
node_resolver: Custom node resolver function
Returns:
Node-enabled GraphQL type
"""Usage Example:
@strawberry.relay.node
@strawberry.type
class User:
id: strawberry.relay.NodeID
name: str
email: str
# The node decorator automatically adds node resolution
@strawberry.type
class Query:
node: strawberry.relay.Node = strawberry.relay.node()
@strawberry.field
def user(self, id: strawberry.ID) -> User:
return User.resolve_node(id, info)Cursor-based pagination following the Relay Connection specification.
class Connection:
"""Relay connection interface for paginated results."""
edges: List[Edge]
page_info: PageInfo
class Edge:
"""Connection edge containing a node and cursor."""
node: Node
cursor: str
class PageInfo:
"""Pagination information for connections."""
has_previous_page: bool
has_next_page: bool
start_cursor: Optional[str]
end_cursor: Optional[str]
class ListConnection:
"""List-based connection implementation."""
@classmethod
def resolve_connection(
cls,
nodes: List[Any],
*,
first: int = None,
after: str = None,
last: int = None,
before: str = None
) -> Connection: ...Usage Examples:
@strawberry.type
class UserConnection(strawberry.relay.Connection[User]):
"""Connection for User nodes."""
edges: List[strawberry.relay.Edge[User]]
page_info: strawberry.relay.PageInfo
@strawberry.type
class UserEdge(strawberry.relay.Edge[User]):
"""Edge for User nodes."""
node: User
cursor: str
@strawberry.type
class Query:
@strawberry.field
def users(
self,
first: int = None,
after: str = None,
last: int = None,
before: str = None
) -> UserConnection:
# Get all users (in practice, this would be a database query)
all_users = get_all_users()
# Convert to User objects
user_nodes = [
User(
id=strawberry.relay.NodeID(str(u.id), "User"),
name=u.name,
email=u.email
)
for u in all_users
]
# Use ListConnection for automatic pagination
return strawberry.relay.ListConnection.resolve_connection(
user_nodes,
first=first,
after=after,
last=last,
before=before
)Decorator to create connection fields with automatic pagination.
def connection(
resolver: Callable = None,
*,
name: str = None,
description: str = None,
permission_classes: List[Type[BasePermission]] = None,
extensions: List[FieldExtension] = None
) -> Any:
"""
Decorator to create Relay connection fields.
Args:
resolver: Resolver function that returns nodes
name: Custom field name
description: Field description
permission_classes: Permission classes for authorization
extensions: Field extensions to apply
Returns:
Connection field with automatic pagination
"""Usage Example:
@strawberry.type
class User:
id: strawberry.relay.NodeID
name: str
@strawberry.relay.connection(
description="User's posts with cursor-based pagination"
)
def posts(
self,
first: int = None,
after: str = None,
last: int = None,
before: str = None
) -> List[Post]:
# Return list of Post objects
# Connection decorator handles pagination automatically
return get_posts_by_user_id(self.id)
@strawberry.type
class Query:
@strawberry.relay.connection
def all_users(
self,
first: int = None,
after: str = None,
last: int = None,
before: str = None,
name_filter: str = None
) -> List[User]:
return get_users(name_filter=name_filter)Utilities for encoding and decoding cursors.
def to_base64(value: str) -> str:
"""
Encode string to base64 for cursor encoding.
Args:
value: String value to encode
Returns:
Base64 encoded string
"""
def from_base64(cursor: str) -> str:
"""
Decode base64 cursor back to string.
Args:
cursor: Base64 encoded cursor
Returns:
Decoded string value
"""Usage Example:
from strawberry.relay import to_base64, from_base64
# Custom cursor creation
def create_cursor(post_id: str, created_at: datetime) -> str:
cursor_data = f"{post_id}:{created_at.isoformat()}"
return to_base64(cursor_data)
def parse_cursor(cursor: str) -> tuple[str, datetime]:
cursor_data = from_base64(cursor)
post_id, created_at_str = cursor_data.split(":", 1)
created_at = datetime.fromisoformat(created_at_str)
return post_id, created_at
@strawberry.type
class Post:
id: strawberry.relay.NodeID
title: str
created_at: datetime
def get_cursor(self) -> str:
return create_cursor(str(self.id), self.created_at)@strawberry.type
class PostConnection:
edges: List[strawberry.relay.Edge[Post]]
page_info: strawberry.relay.PageInfo
total_count: int # Additional field not in standard Connection
@classmethod
def from_posts(
cls,
posts: List[Post],
first: int = None,
after: str = None,
last: int = None,
before: str = None,
total_count: int = None
) -> "PostConnection":
# Custom pagination logic
start_index = 0
end_index = len(posts)
if after:
try:
after_id, _ = parse_cursor(after)
start_index = next(
i for i, post in enumerate(posts)
if str(post.id) == after_id
) + 1
except (ValueError, StopIteration):
start_index = 0
if before:
try:
before_id, _ = parse_cursor(before)
end_index = next(
i for i, post in enumerate(posts)
if str(post.id) == before_id
)
except (ValueError, StopIteration):
end_index = len(posts)
if first is not None:
end_index = min(end_index, start_index + first)
if last is not None:
start_index = max(start_index, end_index - last)
selected_posts = posts[start_index:end_index]
edges = [
strawberry.relay.Edge(
node=post,
cursor=post.get_cursor()
)
for post in selected_posts
]
page_info = strawberry.relay.PageInfo(
has_previous_page=start_index > 0,
has_next_page=end_index < len(posts),
start_cursor=edges[0].cursor if edges else None,
end_cursor=edges[-1].cursor if edges else None
)
return cls(
edges=edges,
page_info=page_info,
total_count=total_count or len(posts)
)
@strawberry.type
class Query:
@strawberry.field
def posts(
self,
first: int = None,
after: str = None,
last: int = None,
before: str = None,
category: str = None
) -> PostConnection:
# Get posts with filtering
posts = get_posts(category=category)
total_count = get_posts_count(category=category)
return PostConnection.from_posts(
posts=posts,
first=first,
after=after,
last=last,
before=before,
total_count=total_count
)@strawberry.type
class User(strawberry.relay.Node):
id: strawberry.relay.NodeID
name: str
@strawberry.relay.connection
def posts(
self,
first: int = None,
after: str = None,
status: str = "published"
) -> List[Post]:
return get_user_posts(
user_id=self.id,
status=status
)
@strawberry.relay.connection
def followers(
self,
first: int = None,
after: str = None
) -> List["User"]:
return get_user_followers(self.id)
@strawberry.type
class Post(strawberry.relay.Node):
id: strawberry.relay.NodeID
title: str
author: User
@strawberry.relay.connection
def comments(
self,
first: int = None,
after: str = None
) -> List[Comment]:
return get_post_comments(self.id)@strawberry.enum
class PostSortOrder(Enum):
CREATED_ASC = "created_asc"
CREATED_DESC = "created_desc"
TITLE_ASC = "title_asc"
TITLE_DESC = "title_desc"
@strawberry.input
class PostFilter:
category: Optional[str] = None
published: Optional[bool] = None
author_id: Optional[strawberry.ID] = None
@strawberry.type
class Query:
@strawberry.relay.connection
def posts(
self,
first: int = None,
after: str = None,
last: int = None,
before: str = None,
filter: PostFilter = None,
sort_by: PostSortOrder = PostSortOrder.CREATED_DESC
) -> List[Post]:
return get_posts_with_filter_and_sort(
filter=filter,
sort_by=sort_by
)Extension for automatic node resolution.
class NodeExtension(FieldExtension):
"""Extension for Relay node resolution."""
def apply(self, field: StrawberryField) -> StrawberryField: ...Extension for automatic connection resolution.
class ConnectionExtension(FieldExtension):
"""Extension for Relay connection resolution."""
def apply(self, field: StrawberryField) -> StrawberryField: ...Relay mutation pattern with input and payload types:
@strawberry.input
class CreatePostInput:
title: str
content: str
category_id: strawberry.ID
@strawberry.type
class CreatePostPayload:
post: Optional[Post]
user_error: Optional[str]
client_mutation_id: Optional[str]
@strawberry.type
class Mutation:
@strawberry.mutation
def create_post(
self,
input: CreatePostInput,
client_mutation_id: str = None
) -> CreatePostPayload:
try:
post = create_new_post(
title=input.title,
content=input.content,
category_id=input.category_id
)
return CreatePostPayload(
post=post,
user_error=None,
client_mutation_id=client_mutation_id
)
except ValidationError as e:
return CreatePostPayload(
post=None,
user_error=str(e),
client_mutation_id=client_mutation_id
)import strawberry
from strawberry import relay
from typing import List, Optional
@strawberry.relay.node
@strawberry.type
class User:
id: strawberry.relay.NodeID
name: str
email: str
@strawberry.relay.connection
def posts(self) -> List["Post"]:
return get_user_posts(self.id)
@strawberry.relay.node
@strawberry.type
class Post:
id: strawberry.relay.NodeID
title: str
content: str
author_id: strawberry.ID
@strawberry.field
def author(self) -> User:
return User.resolve_node(self.author_id, info)
@strawberry.type
class Query:
# Required node field for Relay
node: strawberry.relay.Node = strawberry.relay.node()
@strawberry.relay.connection
def users(self) -> List[User]:
return get_all_users()
@strawberry.relay.connection
def posts(self, category: str = None) -> List[Post]:
return get_posts(category=category)
schema = strawberry.Schema(query=Query)Example Query:
query RelayQuery {
# Node query
node(id: "VXNlcjox") {
id
... on User {
name
email
}
}
# Connection query
users(first: 5, after: "cursor") {
edges {
node {
id
name
posts(first: 3) {
edges {
node {
title
}
}
}
}
cursor
}
pageInfo {
hasNextPage
hasPreviousPage
startCursor
endCursor
}
}
}Install with Tessl CLI
npx tessl i tessl/pypi-strawberry-graphql