prefect_github.graphql

This is a module containing generic GraphQL tasks

execute_graphql async

Generic function for executing GraphQL operations.

Parameters:

Name	Type	Description	Default
op	Union[Operation, str]	The operation, either as a valid GraphQL string or sgqlc.Operation.	required
github_credentials	GitHubCredentials	Credentials to use for authentication with GitHub.	required
error_key	str	The key name to look out for in the response that indicates an error has occurred with the request.	'errors'

Returns:

Type	Description
Dict[str, Any]	A dict of the returned fields.

Examples:

Queries the first three issues from the Prefect repository using a string query.

from prefect import flow
from prefect_github import GitHubCredentials
from prefect_github.graphql import execute_graphql

@flow()
def example_execute_graphql_flow():
    op = '''
        query GitHubRepoIssues($owner: String!, $name: String!) {
            repository(owner: $owner, name: $name) {
                issues(last: 3) {
                    nodes {
                        number
                        title
                    }
                }
            }
        }
    '''
    token = "ghp_..."
    github_credentials = GitHubCredentials(token=token)
    params = dict(owner="PrefectHQ", name="Prefect")
    result = execute_graphql(op, github_credentials, **params)
    return result

example_execute_graphql_flow()

Queries the first three issues from Prefect repository using a sgqlc.Operation.

from prefect import flow
from sgqlc.operation import Operation
from prefect_github import GitHubCredentials
from prefect_github.schemas import graphql_schema
from prefect_github.graphql import execute_graphql

@flow()
def example_execute_graphql_flow():
    op = Operation(graphql_schema.Query)
    op_settings = op.repository(
        owner="PrefectHQ", name="Prefect"
    ).issues(
        first=3
    ).nodes()
    op_settings.__fields__("id", "title")
    token = "ghp_..."
    github_credentials = GitHubCredentials(token=token)
    result = execute_graphql(
        op,
        github_credentials,
    )
    return result

example_execute_graphql_flow()

Source code in prefect_github/graphql.py

@task
async def execute_graphql(
    op: Union[Operation, str],
    github_credentials: GitHubCredentials,
    error_key: str = "errors",
    **vars,
) -> Dict[str, Any]:
    # NOTE: Maintainers can update these examples to match their collection!
    """
    Generic function for executing GraphQL operations.

    Args:
        op: The operation, either as a valid GraphQL string or sgqlc.Operation.
        github_credentials: Credentials to use for authentication with GitHub.
        error_key: The key name to look out for in the response
            that indicates an error has occurred with the request.

    Returns:
        A dict of the returned fields.

    Examples:
        Queries the first three issues from the Prefect repository
        using a string query.
        ```python
        from prefect import flow
        from prefect_github import GitHubCredentials
        from prefect_github.graphql import execute_graphql

        @flow()
        def example_execute_graphql_flow():
            op = '''
                query GitHubRepoIssues($owner: String!, $name: String!) {
                    repository(owner: $owner, name: $name) {
                        issues(last: 3) {
                            nodes {
                                number
                                title
                            }
                        }
                    }
                }
            '''
            token = "ghp_..."
            github_credentials = GitHubCredentials(token=token)
            params = dict(owner="PrefectHQ", name="Prefect")
            result = execute_graphql(op, github_credentials, **params)
            return result

        example_execute_graphql_flow()
        ```

        Queries the first three issues from Prefect repository
        using a sgqlc.Operation.
        ```python
        from prefect import flow
        from sgqlc.operation import Operation
        from prefect_github import GitHubCredentials
        from prefect_github.schemas import graphql_schema
        from prefect_github.graphql import execute_graphql

        @flow()
        def example_execute_graphql_flow():
            op = Operation(graphql_schema.Query)
            op_settings = op.repository(
                owner="PrefectHQ", name="Prefect"
            ).issues(
                first=3
            ).nodes()
            op_settings.__fields__("id", "title")
            token = "ghp_..."
            github_credentials = GitHubCredentials(token=token)
            result = execute_graphql(
                op,
                github_credentials,
            )
            return result

        example_execute_graphql_flow()
        ```
    """
    result = await _execute_graphql_op(
        op, github_credentials, error_key=error_key, **vars
    )
    return result