Schema and Execution

class Schema:
    """Main GraphQL schema class for executing queries, mutations, and subscriptions."""
    
    def __init__(
        self,
        query: Type = None,
        mutation: Type = None,
        subscription: Type = None,
        *,
        types: List[Type] = None,
        extensions: List[SchemaExtension] = None,
        execution_context_class: Type = None,
        introspection: bool = True,
        auto_camel_case: bool = True
    ):
        """
        Initialize a GraphQL schema.
        
        Args:
            query: Root query type
            mutation: Root mutation type (optional)
            subscription: Root subscription type (optional)
            types: Additional types to include in schema
            extensions: Schema extensions for custom functionality
            execution_context_class: Custom execution context class
            introspection: Whether to enable introspection
            auto_camel_case: Whether to auto-convert field names to camelCase
        """
    
    # Schema properties
    query_type: Type  # Root query type
    mutation_type: Optional[Type]  # Root mutation type
    subscription_type: Optional[Type]  # Root subscription type
    types: List[Type]  # All types in the schema

@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "Hello, World!"

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, name: str) -> User:
        return User(id="1", name=name)

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def user_updates(self) -> AsyncIterator[User]:
        # Subscription logic here
        yield User(id="1", name="Alice")

# Create schema
schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription,
    extensions=[
        QueryDepthLimiter(max_depth=10),
        ValidationCache()
    ]
)

def execute_sync(
    self,
    query: str,
    variable_values: Dict[str, Any] = None,
    context_value: Any = None,
    root_value: Any = None,
    operation_name: str = None
) -> ExecutionResult:
    """
    Execute a GraphQL query synchronously.
    
    Args:
        query: GraphQL query string
        variable_values: Variables for the query
        context_value: Context passed to resolvers
        root_value: Root value for top-level resolvers
        operation_name: Name of operation to execute (for multi-operation queries)
    
    Returns:
        ExecutionResult with data and errors
    """

# Simple query execution
result = schema.execute_sync("""
    query {
        hello
    }
""")

print(result.data)  # {'hello': 'Hello, World!'}

# Query with variables
result = schema.execute_sync(
    """
    query GetUser($id: ID!) {
        user(id: $id) {
            name
            email
        }
    }
    """,
    variable_values={"id": "123"}
)

# Query with context
class Context:
    def __init__(self, user=None):
        self.user = user

result = schema.execute_sync(
    """
    query {
        currentUser {
            name
        }
    }
    """,
    context_value=Context(user=current_user)
)

async def execute(
    self,
    query: str,
    variable_values: Dict[str, Any] = None,
    context_value: Any = None,
    root_value: Any = None,
    operation_name: str = None
) -> ExecutionResult:
    """
    Execute a GraphQL query asynchronously.
    
    Args:
        query: GraphQL query string
        variable_values: Variables for the query
        context_value: Context passed to resolvers
        root_value: Root value for top-level resolvers
        operation_name: Name of operation to execute
    
    Returns:
        ExecutionResult with data and errors
    """

import asyncio

async def main():
    # Async query execution
    result = await schema.execute("""
        query {
            users {
                id
                name
                posts {
                    title
                }
            }
        }
    """)
    
    if result.errors:
        print("Errors:", result.errors)
    else:
        print("Data:", result.data)

# Run async execution
asyncio.run(main())

async def subscribe(
    self,
    query: str,
    variable_values: Dict[str, Any] = None,
    context_value: Any = None,
    root_value: Any = None,
    operation_name: str = None
) -> AsyncIterator[ExecutionResult]:
    """
    Execute a GraphQL subscription.
    
    Args:
        query: GraphQL subscription string
        variable_values: Variables for the subscription
        context_value: Context passed to resolvers
        root_value: Root value for top-level resolvers
        operation_name: Name of operation to execute
    
    Yields:
        ExecutionResult for each subscription event
    """

async def handle_subscription():
    subscription_query = """
        subscription {
            messageUpdates(roomId: "123") {
                id
                content
                author {
                    name
                }
                timestamp
            }
        }
    """
    
    async for result in schema.subscribe(subscription_query):
        if result.errors:
            print("Subscription error:", result.errors)
        else:
            message = result.data["messageUpdates"]
            print(f"New message: {message['content']} from {message['author']['name']}")

class ExecutionResult:
    """Result of GraphQL query or mutation execution."""
    
    data: Optional[Dict[str, Any]]  # Query result data
    errors: Optional[List[GraphQLError]]  # Execution errors
    extensions: Optional[Dict[str, Any]]  # Extension data

result = schema.execute_sync("""
    query {
        user(id: "invalid") {
            name
        }
    }
""")

if result.errors:
    for error in result.errors:
        print(f"Error: {error.message}")
        print(f"Path: {error.path}")
        print(f"Locations: {error.locations}")
else:
    print(f"Data: {result.data}")

# Access extensions (from schema extensions)
if result.extensions:
    print(f"Query complexity: {result.extensions.get('complexity')}")
    print(f"Execution time: {result.extensions.get('execution_time')}")

class SubscriptionExecutionResult:
    """Result of GraphQL subscription execution."""
    
    def __aiter__(self) -> AsyncIterator[ExecutionResult]:
        """Async iterator for subscription results."""
    
    async def __anext__(self) -> ExecutionResult:
        """Get next subscription result."""

class ExecutionContext:
    """Context for GraphQL operation execution."""
    
    schema: Schema  # GraphQL schema
    document: DocumentNode  # Parsed GraphQL document
    root_value: Any  # Root value for resolvers
    context_value: Any  # Context passed to resolvers
    variable_values: Dict[str, Any]  # Operation variables
    operation_name: Optional[str]  # Name of operation being executed
    field_resolver: Callable  # Field resolver function
    type_resolver: Callable  # Type resolver function
    subscribe_field_resolver: Callable  # Subscription field resolver

# Register additional types that might not be directly referenced
@strawberry.type
class AdminUser:
    id: strawberry.ID
    name: str
    admin_level: int

schema = strawberry.Schema(
    query=Query,
    types=[AdminUser]  # Ensure AdminUser is included in schema
)

# Disable introspection for production
production_schema = strawberry.Schema(
    query=Query,
    introspection=False  # Disable introspection queries
)

# Or use extension
from strawberry.extensions import DisableIntrospection

schema = strawberry.Schema(
    query=Query,
    extensions=[DisableIntrospection()]
)

# Disable automatic camelCase conversion
schema = strawberry.Schema(
    query=Query,
    auto_camel_case=False  # Keep original Python field names
)

@strawberry.type
class User:
    user_id: strawberry.ID  # Will remain as 'user_id' in GraphQL
    first_name: str  # Will remain as 'first_name' in GraphQL

class CustomError(Exception):
    """Custom application error."""
    def __init__(self, message: str, code: str = None):
        self.message = message
        self.code = code
        super().__init__(message)

@strawberry.type
class Query:
    @strawberry.field
    def risky_operation(self) -> str:
        try:
            return perform_risky_operation()
        except CustomError as e:
            # Errors are automatically added to the result
            raise e
        except Exception as e:
            # Generic error handling
            raise GraphQLError(f"Operation failed: {str(e)}")

# Execute with error handling
result = schema.execute_sync("""
    query {
        riskyOperation
    }
""")

if result.errors:
    for error in result.errors:
        print(f"Error: {error.message}")
        if hasattr(error.original_error, 'code'):
            print(f"Code: {error.original_error.code}")

from strawberry.printer import print_schema

# Print schema SDL (Schema Definition Language)
sdl = print_schema(schema)
print(sdl)

type Query {
  hello: String!
  user(id: ID!): User
}

type User {
  id: ID!
  name: String!
  email: String!
}

type Mutation {
  createUser(name: String!): User!
}

# Execute specific operation from multi-operation query
multi_operation_query = """
    query GetUsers {
        users { id name }
    }
    
    query GetPosts {
        posts { id title }
    }
"""

# Execute specific operation
result = schema.execute_sync(
    multi_operation_query,
    operation_name="GetUsers"
)