GraphQL API Advanced Usage

Advanced GraphQL patterns for complex queries, mutations, and integrations. Optimizing performance and handling large result sets.

You Might Be Asking:

• How do I write complex GraphQL queries?
• Can I batch operations?
• How do I handle pagination?

1. Complex queries with fragments:

fragment DatasetInfo on Dataset {
  urn
  name
  platform { name }
  properties {
    description
  }
  ownership {
    owners {
      owner {
        ... on CorpUser {
          username
        }
      }
    }
  }
}

query comprehensiveQuery($urn: String!) {
  dataset(urn: $urn) {
    ...DatasetInfo

    upstream: relationships(
      input: {
        types: ["DownstreamOf"]
        direction: INCOMING
      }
    ) {
      relationships {
        entity {
          ...DatasetInfo
        }
      }
    }
  }
}

2. Batch operations:

from datahub.ingestion.graph.client import DatahubClientConfig, DataHubGraph

class BatchGraphQLClient:
    """
    Efficient batch operations
    """

    def __init__(self):
        self.graph = DataHubGraph(DatahubClientConfig())

    def batch_get_datasets(self, urns: list):
        """
        Fetch multiple datasets using aliases
        """
        query_parts = ["query {"]

        for i, urn in enumerate(urns):
            alias = f"d{i}"
            query_parts.append(f'''
                {alias}: dataset(urn: "{urn}") {{
                    urn
                    name
                    properties {{ description }}
                }}
            ''')

        query_parts.append("}")
        query = "\n".join(query_parts)

        result = self.graph.execute_graphql(query)

        # Map results back
        datasets = {}
        for i, urn in enumerate(urns):
            alias = f"d{i}"
            if result['data'].get(alias):
                datasets[urn] = result['data'][alias]

        return datasets


# Usage
client = BatchGraphQLClient()
urns = [
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table1,PROD)",
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table2,PROD)"
]
datasets = client.batch_get_datasets(urns)

3. Pagination:

def paginated_search(query: str, page_size: int = 100):
    """
    Generator for paginated results
    """
    graph = DataHubGraph(DatahubClientConfig())
    start = 0

    while True:
        result = graph.execute_graphql("""
            query paginatedSearch($input: SearchInput!) {
              search(input: $input) {
                total
                searchResults {
                  entity {
                    urn
                    ... on Dataset { name }
                  }
                }
              }
            }
        """, variables={
            "input": {
                "type": "DATASET",
                "query": query,
                "start": start,
                "count": page_size
            }
        })

        results = result['data']['search']['searchResults']
        if not results:
            break

        for item in results:
            yield item

        start += page_size


# Usage
for result in paginated_search("analytics"):
    print(result['entity']['name'])

Use fragments to reduce duplication. Batch operations to reduce round trips. Implement caching. Use field selection to minimize payload. Leverage parallel execution. Monitor query performance. Use pagination for large sets. Handle rate limits. Test in GraphQL playground.

• GraphQL API
• GraphQL Queries