rob-blackbourn

"""
Example Client

The client connects without TLS, but using the fully qualified domain name. To
authenticate the server, the FQDN is required. This can be specified either at
connection time, or with the start_tls call.

First the client sends a "PING" over the unencrypted stream  to the server. The
server should respond with "PONG".

rob-blackbourn

import asyncio
import signal
from typing import Any, Mapping

from bareasgi import Application
import pkg_resources
import hypercorn.asyncio
import hypercorn.config

from demo.app import make_application

rob-blackbourn

import graphene

from .queries import Query
from .mutations import Mutations
from .subscriptions import Subscription

SCHEMA = graphene.Schema(query=Query, mutation=Mutations, subscription=Subscription)

rob-blackbourn

import os
from typing import Any, Mapping

from bareasgi import Application
from bareasgi_cors import CORSMiddleware
from bareasgi_graphql_next.graphene import add_graphene
from baretypes import (
    Scope,
    Info,
    Message

rob-blackbourn

import typing

import graphene
from jetblack_tweeter import Tweeter

from .types import TweetType

class Subscription(graphene.ObjectType):
    filter = graphene.Field(
        TweetType,

rob-blackbourn

import typing

import graphene
from jetblack_tweeter import Tweeter

from ..types import TweetType

class UpdateStatus(graphene.Mutation):
    class Arguments:
        status = graphene.String()

rob-blackbourn

import typing

import graphene
from jetblack_tweeter import Tweeter

from .types import SearchResultType

class Query(graphene.ObjectType):
    search_tweets = graphene.Field(
        SearchResultType,

rob-blackbourn

import graphene

class UserType(graphene.ObjectType):
    id = graphene.types.scalars.BigInt(required=True)
    id_str = graphene.String(required=True)
    name = graphene.String(required=True)
    screen_name = graphene.String(required=True)
    location = graphene.String(required=False)
    url = graphene.String(required=False)
    description = graphene.String(required=False)

rob-blackbourn

import asyncio
import os

from jetblack_tweeter import Tweeter
from jetblack_tweeter.clients.bareclient import BareTweeterSession

async def main():
    # Create the twitter client.
    tweeter = Tweeter(
        BareTweeterSession(),

rob-blackbourn

"""GraphQL JSON type"""

import json
from typing import List, Mapping, Union, Any

from graphql.type import GraphQLScalarType
from graphql.language.ast import StringValueNode
from graphql.error import INVALID

JSONType = Union[List[Mapping[str, Any]], Mapping[str, Any]]