GraphQL API Streams
Relevant source files
The following files were used as context for generating this wiki page:
GraphQL API Streams are a specialized implementation of data extraction within the Meltano Singer SDK, designed specifically to interface with GraphQL-based endpoints. They extend the functionality of REST API Streams by substituting path-based resource targeting with flexible GraphQL query strings.
This module provides an abstract base class, GraphQLStream, which handles the complexities of GraphQL request preparation, payload wrapping, and specialized response validation for handling partial successes and body-level errors common in GraphQL APIs.
Sources: singer_sdk/streams/graphql.py:1-24, CLAUDE.md:130-136
Core Architecture
The GraphQLStream class inherits from RESTStream but modifies the communication pattern. Unlike REST, which often uses multiple HTTP methods (GET, POST, etc.) across various paths, GraphQL streams default to using POST requests directed at a single endpoint, carrying the query logic in the request body.
The diagram shows the inheritance hierarchy where GraphQLStream specializes RESTStream by introducing query-specific logic.
Sources: singer_sdk/streams/graphql.py:27-32, packages/meltano-tap-gitlab/tap_gitlab/graphql_streams.py:24-27
Key Components
| Component | Description |
|---|---|
query | A property or attribute containing the GraphQL query string. If not prefixed with query, the SDK automatically wraps it in a query { ... } block. |
records_jsonpath | A JSONPath expression used to extract records from the response. By default, it looks in $.data.{stream_name}[*]. |
prepare_request_payload | Generates the JSON body containing the query and any variables derived from URL parameters. |
validate_response | A specialized validator that checks for GraphQL-specific errors in the JSON body, even when the HTTP status code is 200. |
Sources: singer_sdk/streams/graphql.py:34-45, singer_sdk/streams/graphql.py:65-79, singer_sdk/streams/graphql.py:84-90
Request Lifecycle and Data Flow
The data flow for a GraphQL stream involves transforming stream configuration and context into a valid GraphQL POST payload.
The flow shows how the query string is prepared and wrapped before being sent as a POST payload alongside variables.
Sources: singer_sdk/streams/graphql.py:53-76, tests/core/test_streams.py:263-276
Payload Preparation
The prepare_request_payload method automatically flattens and cleans the query string. It takes the output of get_url_params and maps it to the variables key in the JSON request body.
Sources: singer_sdk/streams/graphql.py:53-80
# Example implementation in singer_sdk/streams/graphql.py
def prepare_request_payload(self, context, next_page_token):
params = self.get_url_params(context, next_page_token)
query = self.query
if not query.lstrip().startswith("query"):
query = "query { " + query + " }"
return {
"query": (" ".join([line.strip() for line in query.splitlines()])),
"variables": params,
}
Sources: singer_sdk/streams/graphql.py:65-80
Specialized Response Validation
GraphQL APIs often return a 200 OK status even if the query fails, placing error details in an errors array within the response body. GraphQLStream overrides the standard validation to handle these cases.
Error Handling Scenarios
The SDK distinguishes between total failures and partial successes where some data is returned alongside warnings.
| Scenario | Response Content | SDK Action |
|---|---|---|
| Total Failure | errors present, data is None or missing | Raises FatalAPIError |
| Partial Success | errors present, data contains records | Logs a Warning; continues sync |
| Invalid JSON | Body is not valid JSON | Raises FatalAPIError |
| Success | No errors key | Proceeds to record extraction |
Sources: singer_sdk/streams/graphql.py:84-129, tests/core/rest/test_graphql.py:53-112
Implementation Example: GitLab GraphQL
The GitlabGraphQLStream demonstrates a practical implementation using a dynamic query and manual header injection for authentication.
class GraphQLProjectsStream(GitlabGraphQLStream):
"""Gitlab Projects stream."""
name = "projects"
primary_keys = ("id",)
@property
def query(self) -> str:
"""Dynamic GraphQL query."""
return f"project(fullPath: {self.config['project_id']} {{ name }}"
Sources: packages/meltano-tap-gitlab/tap_gitlab/graphql_streams.py:47-56
Conclusion
GraphQL API Streams provide a robust framework for interacting with modern GraphQL endpoints by standardizing the request wrapping and error validation logic. By inheriting from GraphQLStream, developers can focus on defining their schemas and queries while the SDK manages the complexities of payload construction and response integrity checks.