Native schema stitching

[arindam04]

Hi
Can anyone provide an example code how to do schema stitching in python using graphene?
Thanks

[Nabellaleen]

Each "node" of a Graphene is an ObjectType, so I think there isn't specifics things to achieve to "assemble" multiple schema togethers.

class Greetings(graphene.ObjectType):
    hello = graphene.String(argument=graphene.String(default_value="stranger"))

    def resolve_hello(self, info, argument):
        return 'Hello ' + argument

class Query(graphene.ObjectType):
    greetings = Greetings
    
    def resolve_greetings(self, info):
        return self

schema = graphene.Schema(query=Query)

So add something else is not more than create a new ObjectType and add it as a "field" in Query

Is it what you are looking for ?

[ekampf]

Schema stitching is the process of creating a single GraphQL schema from multiple underlying remote GraphQL APIs.

We don't have that capability in graphene

[arindam04]

@ekampf Thanks for the clarification. Is there any implementation of this problem in python? Even at graphql-core level. I am looking for a naive approach to stich the the types. Something along the lines of ( This is just a prototype, not complete). I basically want a way to combine objects coming from different schemas to one master schema.
Can you suggest a naive approach to solve this till graphehe comes up with a robust implementation?

for schema_type_name, schema_type in child_schema.get_type_map().items():
if not schema_type_name.startswith('__') and isinstance(schema_type, GraphQLObjectType):
if schema_type_name in parent_schema.get_type_map():
for field_name, field_type in schema_type.fields.items():
if field_name not in schema.get_type(schema_type_name).fields:
parent_schema.get_type(schema_type_name).fields[field_name] = field_type
else:
parent_schema.get_type_map()[schema_type_name] = schema_type

[ekampf]

@arindam04 we don't have such an implementation in Python.
We can look into implementing it after Graphene v3.0

An alternative solution would be to put an Apollo Server as a proxy in front of your python server just for sticking purposes

[jkimbo]

BTW @arindam04 schema stitching in Apollo has been deprecated (and replaced by a new architecture: https://blog.apollographql.com/apollo-federation-f260cf525d21).

[japrogramer]

Looks like there needs to be a way to declare the directives
@key (fields: ...)
@external
@requires(fields: ...)

also something equivalent to
extend type ... {}
and
const { ApolloGateway } = require("@apollo/gateway");

while we are at it, why not implement mocks also?

[anlauren]

This is actually a something fairly needed in order to integrate Graphene with other Graphql services.
There's a way to implement the directives #1021, but it doesn't seem like we can express these at a type level to be exported in the schema

[jkimbo]

@anlauren as explained here it is not currently possible to add directives at the type level and won't be until it is supported in graphql-js. Until then I think your only option would be use something like https://github.com/mirumee/ariadne which lets you build your GraphQL server using SDL.

[japrogramer]

@jkimbo
i looked thru all ariadne's docs but couldn't find anything about implementing directives.
Also couldn't find anything about relay,edges or PageInfo or connections in the docs.

Are those features not included with ariadne ? is their a package that adds them ?
Edit:
oh, ok: this explains why mirumee/ariadne#188

[jkimbo]

@japrogramer I’m not familiar with Ariadne so you’d have to raise an issue the repo to get those answers.

[anlauren]

As this seems to suggest`` custom directives are not yet implemented in Ariadne.
i already saw that it was not possible, i am just trying to find a way to make it work with apollo server, as i have a production backend, it is sadly not an option to rewrite the whole backend.
Even if using Ariadne i'll be facing a big and costly rewrite.
It might take some time to get from Graphql js to graphql-core (soon to be the actual version of graphql-next?) and then working with graphene 😢

[japrogramer]

@anlauren like this.

import { ApolloServer } from 'apollo-server';
import { ApolloGateway, RemoteGraphQLDataSource } from '@apollo/gateway';

const remoteGraphs = [
      { name: 'django-product', url: 'https://products-service.dev/graphql' },
      { name: 'reviews', url: 'https://reviews-service.dev/graphql' }
  ]

const gateway = new ApolloGateway({
  serviceList: remoteGraphs,
  buildService({ name, url }) {
    if name.includes('django') {
        return new RemoteGraphQLDataSource({
            url,
            willSendRequest({ request, context }) {
                request.http.headers.set('CSRF-TOKEN', get_token());
            },
        });
    } else {
        return new RemoteGraphQLDataSource({
            url,
            willSendRequest({ request, context }) {
                request.http.headers.set('x-user-id', context.userId);
            },
        });
    }
  },
});

(async () => {
  const { schema, executor } = await gateway.load();

  const server = new ApolloServer({ schema, executor });

  server.listen().then(({ url }) => {
    console.log(`🚀 Server ready at ${url}`);
  });
})();

[anlauren]

@japrogramer yes but then my graphene server needs to implement the "federation" protocol in order to respond to that set up that Apollo gateway.
I am going to add the _service field to see if it at least connects, but i'll never be able to implement the directives in graphene :/

[anlauren]

So i have given it a go anyways, it seems that the above works combined to this code added to your schema in graphene:

schema = ""
class ServiceField(graphene.ObjectType):
    sdl = String()

    def resolve_sdl(parent, _):
        string_schema = str(schema)
        string_schema = string_schema.replace("\n", " ")
        string_schema = string_schema.replace("type Query", "extend type Query")
        string_schema = string_schema.replace("schema {   query: Query   mutation: MutationQuery }", "")
        return string_schema


class Service:
    _service = graphene.Field(ServiceField, name="_service", resolver=lambda x, _: {})

class Query(
    # ...
    Service,
    graphene.ObjectType,
):
    pass

schema = graphene.Schema(query=Query, types=CUSTOM_ATTRIBUTES_TYPES)

So that can work with the old way of doing schema stiching, but for federation we'd need to get creative 🤔

[anlauren]

So, after a bit of workaround I came accross a way to hack Graphene around and actually make the federation protocol work!

First, implement my objects. in that example, Book is defined from another service, DomesticRegion is defined in the graphene service.

We define an entity as:

class EntityType:
    __typename=String()

    @staticmethod
    def get_sdl():
        raise Exception("Any entity needs to implement a get_sdl method")

class DomesticRegion(ObjectType):
    id = String()
    name = String()

    @staticmethod
    def get_sdl():
        return 'type DomesticRegion @key(fields: "id"){ id: String name: String }'

class Book(EntityType, graphene.ObjectType):
    id = String()
    year = String()

    @staticmethod
    def get_sdl():
        return 'extend type Book @key(fields: "id"){ id: String @external year: String }'

    def resolve_year(book, self):
        return "Graphene extended Book and added the year it got from the id {}".format(book.id)

As a side note, in my other service, i define Book that way:

  type Book @key(fields: "id") {
    id: String
    title: String
    author: String
  }
  extend type Query {
    books: [Book]
  }  

Now you'll need to implement the "federation protocol".
First, you'll need to define a list of the Entities you are using and sharing accross services, i do it that way:

custom_entities = {
    "Book": Book,
    "DomesticRegionField": DomesticRegionField
}

and then add these in a union that is defined in the federationrequirement:

class _Entity(graphene.Union):
    class Meta:
        types = (Book, DomesticRegionField,)

These entities need to be queried, so needs an _entity entry in the graph. This entry is resolving at runtime the type of the entity that we are looking for:

class EntityQuery:
    entities = graphene.List(_Entity, name="_entities", representations=List(_Any))

    def resolve_entities(parent, _, representations):
        casted_represenations = []
        for representation in representations:
            model = custom_entities[representation["__typename"]]
            model_aguments = representation.copy()
            model_aguments.pop("__typename")
            casted_represenations.append(
                model(**model_aguments)
            )
        return casted_represenations

This needs to accept an "any" type that we define like this:

class _Any(Scalar):
    '''Anything'''

    __typename=String(required=True)

    @staticmethod
    def serialize(dt):
        return dt

    @staticmethod
    def parse_literal(node):
       return node

    @staticmethod
    def parse_value(value):
        return value

The graph as well needs to be served so the federation service to be aware of it. Following the federation guide, you need to implement an _service entry.
To serve that graph, i am going to use the str(schema) provided by graphene to print the schema, and srtip out what needs to be stripped out to be a readable graph by the federation endpoint.
I am as well going to replace every Entity by its sdl schema using the implemented get_sdl function.

class ServiceField(graphene.ObjectType):
    sdl = String()

    def resolve_sdl(parent, _):
        string_schema = str(schema)
        string_schema = string_schema.replace("\n", " ")

        string_schema = string_schema.replace("type Query", "extend type Query")
        regex = r"schema \{(\w|\!|\s|\:)*\}"
        pattern = re.compile(regex)
        string_schema = pattern.sub( " ", string_schema)

        for entity_name, entity in custom_entities.items():
            regex = r"type " + entity_name + " \{(\w|\!|\s|\:)*\}"
            pattern = re.compile(regex)
            string_schema = pattern.sub(entity.get_sdl() , string_schema)

        return string_schema

finally we can define our query and our schema:

class Query(ObjectType):
   Book = graphene.Field(Book, name="Book")
   _service = graphene.Field(ServiceField, name="_service", resolver=lambda x, _: {})
  domestic_region = Field(DomesticRegion)


schema = graphene.Schema(query=Query)

And that should work! let me know if you have a more glamorous solution, if i forgot anything in this example

[japrogramer]

@anlauren is this a typo

class Query(
   Book = graphene.Field(Book, name="Book")
   _service = graphene.Field(ServiceField, name="_service", resolver=lambda x, _: {})
  domestic_region = Field(DomesticRegion)
)
# should it be?
class Query(object):
   Book = graphene.Field(Book, name="Book")
   _service = graphene.Field(ServiceField, name="_service", resolver=lambda x, _: {})
  domestic_region = Field(DomesticRegion)

[anlauren]

@japrogramer oops yes it is, when I cut off my code to make it clearer.
I usally do

class StuffQuery:
   my_stuff = String()

Query(
   StuffQuery,
   ObjectType
):
   pass

hence the confusion. I'll edit it :)

[cramshaw]

Hey,

I am also trying to do this, it would be great to get it built it. I've done my own hacks, just adding the _service. I ended up with:

from graphene import ObjectType, Field, String
from graphql import build_client_schema, print_schema


class _Service(ObjectType):
    sdl = String()

    def resolve_sdl(self, info):
        #  Import here to avoid circular imports
        from gql.schema import base_schema
        return print_schema(build_client_schema(base_schema.introspect()))


class Query:

    _service = Field(_Service, name="_service")

    def resolve__service(parent, info):
        return _Service()

It doesn't handle the entity stuff. I also create two schemas in gql.schema:

import graphene

import gql.query_schema
import gql.federated_schema

class BaseQuery(gql.query_schema.Query, graphene.ObjectType):
    pass


class FederatedQuery(
    gql.query_schema.Query,
    gql.federated_schema.Query,
    graphene.ObjectType
):
    pass

base_schema = graphene.Schema(query=BaseQuery)

schema = graphene.Schema(query=FederatedQuery)

because according to the federation spec _service, or any of these 'special' fields shouldn't actually appear in the SDL. Or that was my understanding at least.
I think this is a simpler way of building the SDL if nothing else, but really feeling around in the dark here!

[erebus1]

Here is partial and draft implementation of federation specs, based on comments above, with nice api.
Hope, it'll help someone https://github.com/erebus1/graphene-federation

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.