ASGI Integration

class GraphQL:
    """ASGI application for GraphQL APIs."""
    
    def __init__(
        self,
        schema: GraphQLSchema,
        *,
        context_value: Optional[Any] = None,
        root_value: Optional[Any] = None,
        debug: bool = False,
        explorer: Optional[Explorer] = None,
        extensions: Optional[list[Extension]] = None,
        middleware: Optional[Middlewares] = None,
        error_formatter: Optional[Callable] = None,
        validation_rules: Optional[list] = None,
        require_query: bool = False,
        websocket_handler: Optional[WebSocketHandler] = None,
        on_connect: Optional[OnConnect] = None,
        on_disconnect: Optional[OnDisconnect] = None,
        on_operation: Optional[OnOperation] = None,
        on_complete: Optional[OnComplete] = None,
        keepalive: Optional[float] = None,
        **kwargs
    ):
        """
        Initialize GraphQL ASGI application.
        
        Parameters:
        - schema: Executable GraphQL schema
        - context_value: Context value or factory function
        - root_value: Root value for GraphQL execution
        - debug: Enable debug mode
        - explorer: GraphQL explorer for development
        - extensions: List of GraphQL extensions
        - middleware: ASGI middleware
        - error_formatter: Custom error formatting function
        - validation_rules: Custom validation rules
        - require_query: Require 'query' in request data
        - websocket_handler: Custom WebSocket handler
        - on_connect: WebSocket connection callback
        - on_disconnect: WebSocket disconnection callback
        - on_operation: GraphQL operation callback
        - on_complete: Operation completion callback
        - keepalive: WebSocket keepalive interval
        """

# Callback types
OnConnect = Callable[[dict], Union[dict, bool, Awaitable[Union[dict, bool]]]]
OnDisconnect = Callable[[dict], Union[None, Awaitable[None]]]  
OnOperation = Callable[[dict, dict], Union[dict, Awaitable[dict]]]
OnComplete = Callable[[dict, dict], Union[None, Awaitable[None]]]

# Data types
Operation = dict[str, Any]
Extensions = dict[str, Any]
MiddlewareList = list[Callable]
Middlewares = Union[MiddlewareList, Callable]

# Exception types
class WebSocketConnectionError(Exception):
    """Exception raised for WebSocket connection errors."""

from ariadne import make_executable_schema, gql, QueryType
from ariadne.asgi import GraphQL

# Create schema
type_defs = gql("""
    type Query {
        hello: String!
    }
""")

query = QueryType()

@query.field("hello")
def resolve_hello(*_):
    return "Hello, World!"

schema = make_executable_schema(type_defs, query)

# Create ASGI app
app = GraphQL(schema, debug=True)

# Run with uvicorn
# uvicorn app:app --reload

from ariadne.asgi import GraphQL
import asyncio

async def get_context_value(request):
    """Context factory function."""
    return {
        "request": request,
        "user": await get_current_user(request),
        "db": database_connection
    }

app = GraphQL(
    schema,
    context_value=get_context_value,
    debug=False
)

from ariadne.asgi import GraphQL

async def on_connect(websocket, connection_params):
    """Called when WebSocket connection is established."""
    print(f"WebSocket connected: {connection_params}")
    # Return False to reject connection
    # Return dict to set connection context
    return {"user_id": connection_params.get("user_id")}

async def on_disconnect(websocket, connection_context):
    """Called when WebSocket connection is closed."""
    print(f"WebSocket disconnected: {connection_context}")

async def on_operation(websocket, connection_context, operation):
    """Called for each GraphQL operation."""
    print(f"GraphQL operation: {operation['type']}")
    return operation

app = GraphQL(
    schema,
    on_connect=on_connect,
    on_disconnect=on_disconnect,
    on_operation=on_operation,
    keepalive=10.0  # Send keepalive every 10 seconds
)

from ariadne.asgi import GraphQL

def custom_error_formatter(error, debug=False):
    """Custom error formatter."""
    formatted = {
        "message": str(error),
        "code": getattr(error, "code", "INTERNAL_ERROR")
    }
    
    if debug:
        formatted["locations"] = getattr(error, "locations", None)
        formatted["path"] = getattr(error, "path", None)
    
    return formatted

app = GraphQL(
    schema,
    error_formatter=custom_error_formatter,
    debug=True
)

from ariadne.asgi import GraphQL
from starlette.middleware.cors import CORSMiddleware
from starlette.applications import Starlette

# Create GraphQL app
graphql_app = GraphQL(schema)

# Wrap with Starlette for middleware
app = Starlette()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

app.mount("/graphql", graphql_app)