Skip to content

prefect_github.graphql

This is a module containing generic GraphQL tasks

execute_graphql(op, github_credentials, error_key='errors', **vars) 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 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@task
@sync_compatible
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