0_graphql.py

pass

aridine extension

from typing import Set
import time
import random
from ariadne import gql, load_schema_from_path
from ariadne import (
    QueryType,
    ObjectType,
    make_executable_schema
)
from ariadne.asgi import GraphQL
from ariadne.contrib.tracing.apollotracing import ApolloTracingExtension
from ariadne.types import Extension


from app.config import config

"""
a entrypoint use for uvicorn to load asgi app

..code-block:
    $> uvicorn app.graphql:app --reload

"""


# ...or inside Python files
type_defs = """
type Query {
  user: User
}

type User {
  id: ID
  username: String!
  friends: [User!]!
}

"""


query = QueryType()
user = ObjectType("User")

@query.field("user")
async def resolve_user(*_):
    return {
        'id': 0,
        'username': 'testname',
        'friends': [{'id': 1}, {'id': 2}, {'id': 3}]
    }

@user.field("friends")
async def resolve_user_friends(parent, info):
    return [{
        'id': row['id'],
        'username': f"user-{row['id']}",
        'friends': [{'id': random.randint(0, 100)}
                    for i in range(0, random.randint(0, 5))],
    } for row in parent.get('friends', [])]


def custom_middleware(resolver, obj, info, **args):
    # print("m")

    value = resolver(obj, info, **args)
    # if isinstance(value, str):
    #     return value.lower()
    # context={'request': <starlette.requests.Request object at 0x104b83890>}
    return value


from inspect import isawaitable
from typing import Any, AsyncGenerator, Callable, List, Optional, Tuple, Union
from typing_extensions import Protocol
from ariadne.types import Resolver
from graphql import (
    DocumentNode,
    ExecutionResult,
    GraphQLError,
    GraphQLResolveInfo,
    GraphQLSchema,
)

class HelloExt(Extension):
    def __init__(self):
        print("init")
        self.start_timestamp = None
        self.end_timestamp = None

    def request_started(self, context):
        print("AA", context)
        self.start_timestamp = time.perf_counter_ns()

    def request_finished(self, context):
        print("BB", context)


    async def resolve(
        self, next_: Resolver, parent: Any, info: GraphQLResolveInfo, **kwargs
    ):
        print("resolve")
        result = await super().resolve(next_, parent, info, **kwargs)
        self.end_timestamp = time.perf_counter_ns()
        return result

    def format(self, context):
        if self.start_timestamp and self.end_timestamp:
            print("ZZZZ")
            return {
                "hello": self.start_timestamp - self.end_timestamp
            }


# schema = make_executable_schema(type_defs, resolvers)
resolvers = [
    query,
    user
]
schema = make_executable_schema(type_defs, resolvers)

# a starlette asgi app
app = GraphQL(schema,
              debug=config['global']['debug'],
              extensions=[
                HelloExt,
                ApolloTracingExtension
              ],
              middleware=[custom_middleware])

test-fastapi.py

print("B")
from typing import Set

from fastapi import FastAPI
from pydantic import BaseModel, Field


# from starlette.graphql import GraphQLApp
from fastapi import FastAPI
from ariadne import gql, load_schema_from_path
from ariadne import QueryType, make_executable_schema
from ariadne.asgi import GraphQL
from ariadne.contrib.tracing.apollotracing import ApolloTracingExtension


app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


@app.get("/items/")
async def read_items():
    return [{"item_id": "Foo"}]


@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}


class Item(BaseModel):
    name: str
    description: str = Field(None, title="this is title", max_length=50)
    price: float
    tax: float = None
    tags: Set[str] = []


@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(*, item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    \f
    :param item: User input.
    """
    return item

# ...or inside Python files
type_defs = """
type Query {
  user: User
}

type User {
  id: ID
  username: String!
}
"""

query = QueryType()


@query.field("user")
async def resolve_user(*_):
    return {
        'id': 0,
        'username': 'testname'
    }


def custom_middleware(resolver, obj, info, **args):
    # print(resolver)
    print(obj, info, args)
    value = resolver(obj, info, **args)
    # if isinstance(value, str):
    #     return value.lower()
    # context={'request': <starlette.requests.Request object at 0x104b83890>}
    return value

# schema = make_executable_schema(type_defs, resolvers)
schema = make_executable_schema(type_defs, query)
graphql_app = GraphQL(schema,
                      debug=True,
                      extensions=[ApolloTracingExtension],
                      middleware=[custom_middleware])

app.add_route("/gql", graphql_app)