Schema and Execution

class Schema:
    """
    Main GraphQL schema definition and execution engine.
    
    Parameters:
        query: Root query ObjectType (required)
        mutation: Root mutation ObjectType (optional)
        subscription: Root subscription ObjectType (optional)
        types: Additional types to include in schema
        directives: Custom GraphQL directives
        auto_camelcase: Automatically convert snake_case to camelCase
    
    Methods:
        execute(query, *args, **kwargs): Synchronous GraphQL execution
        execute_async(query, *args, **kwargs): Asynchronous GraphQL execution
        subscribe(query, *args, **kwargs): GraphQL subscription execution
        introspect(): Schema introspection query execution
        lazy(type): Lazy type resolution for circular references
    
    Features:
    - Type mapping and validation
    - Query parsing and execution
    - Error handling and formatting
    - Subscription support
    - Schema introspection
    - Integration with graphql-core execution engine
    
    Usage:
        schema = graphene.Schema(
            query=Query,
            mutation=Mutation,
            subscription=Subscription,
            types=[CustomType1, CustomType2],
            auto_camelcase=True
        )
        
        # Execute query
        result = schema.execute('{ users { name email } }')
        print(result.data)
    """

def Schema(query=None, mutation=None, subscription=None, types=None, directives=None, auto_camelcase=True):
    """
    Create a GraphQL schema with the specified root types.
    
    Args:
        query: Root query type defining available queries
        mutation: Root mutation type defining available mutations
        subscription: Root subscription type defining available subscriptions
        types: List of additional types to include in schema
        directives: Custom GraphQL directives
        auto_camelcase: Convert Python snake_case to GraphQL camelCase
        
    Returns:
        Schema: Configured GraphQL schema ready for execution
    """

def execute(schema, query, *args, **kwargs):
    """
    Execute GraphQL query synchronously.
    
    Args:
        schema: GraphQL schema
        query: GraphQL query string or DocumentNode
        variable_values: Query variables (dict)
        context_value: Execution context object
        root_value: Root value for query execution
        operation_name: Specific operation to execute (for multi-operation documents)
        validate: Whether to validate query (default: True)
        middleware: List of middleware functions
        
    Returns:
        ExecutionResult: Result object with data, errors, and extensions
        
    Usage:
        result = schema.execute('''
            query GetUser($id: ID!) {
                user(id: $id) {
                    name
                    email
                }
            }
        ''', variable_values={'id': '1'})
        
        if result.errors:
            print("Errors:", result.errors)
        else:
            print("Data:", result.data)
    """

def execute_async(schema, query, *args, **kwargs):
    """
    Execute GraphQL query asynchronously.
    
    Args:
        Same as execute() but returns awaitable
        
    Returns:
        Awaitable[ExecutionResult]: Async result object
        
    Usage:
        import asyncio
        
        async def run_query():
            result = await schema.execute_async('''
                query {
                    users {
                        name
                        email
                    }
                }
            ''')
            return result.data
            
        data = asyncio.run(run_query())
    """

def subscribe(schema, query, *args, **kwargs):
    """
    Execute GraphQL subscription.
    
    Args:
        Same as execute() but for subscription operations
        
    Returns:
        AsyncIterator[ExecutionResult]: Stream of subscription results
        
    Usage:
        async def handle_subscription():
            subscription = schema.subscribe('''
                subscription {
                    messageAdded {
                        id
                        content
                        user {
                            name
                        }
                    }
                }
            ''')
            
            async for result in subscription:
                if result.data:
                    print("New message:", result.data)
    """

class Mutation(ObjectType):
    """
    Convenience class for mutation field definitions.
    
    Features:
    - Inherits from ObjectType with mutation-specific helpers
    - Automatic argument extraction from Arguments class or Input class
    - Integration with Relay mutation patterns
    - Custom resolver support
    
    Meta Options:
        output: Custom output type for mutation
        resolver: Custom resolver function
        arguments: Manual argument definitions
        interfaces: Interfaces this mutation implements
    
    Required Method:
        mutate(): Core mutation logic implementation
    
    Class Methods:
        Field(): Create mutation field for schema
    
    Usage:
        class CreateUser(graphene.Mutation):
            class Arguments:
                name = graphene.String(required=True)
                email = graphene.String()
                age = graphene.Int()
            
            # Output fields
            user = graphene.Field(User)
            success = graphene.Boolean()
            
            def mutate(self, info, name, email=None, age=None):
                # Mutation logic
                user = create_user(name=name, email=email, age=age)
                return CreateUser(user=user, success=True)
        
        class RootMutation(graphene.ObjectType):
            create_user = CreateUser.Field()
    """

def Field(cls):
    """
    Class method to create mutation field.
    
    Returns:
        Field: Configured field for use in schema
        
    Usage:
        class Mutation(graphene.ObjectType):
            create_user = CreateUser.Field()
            update_user = UpdateUser.Field()
    """

class Context:
    """
    Container for execution context data.
    
    Features:
    - Holds request-specific information
    - Available to all resolvers
    - Dynamic attribute assignment
    - Integration with web frameworks
    
    Common Usage:
    - User authentication information
    - Database connections and sessions
    - Request objects (HTTP, WebSocket)
    - Data loaders for batching
    - Cache instances
    - Configuration settings
    
    Usage:
        # Create context
        context = graphene.Context(
            user=current_user,
            request=request,
            db=database_session
        )
        
        # Use in execution
        result = schema.execute(
            query,
            context_value=context
        )
        
        # Access in resolvers
        def resolve_posts(self, info):
            user = info.context.user
            db = info.context.db
            return db.query(Post).filter_by(author_id=user.id).all()
    """

class ResolveInfo:
    """
    GraphQL execution info object (from graphql-core).
    
    Attributes:
        field_name: Name of the current field being resolved
        return_type: GraphQL return type expected
        parent_type: Parent GraphQL type
        schema: GraphQL schema instance
        fragments: Query fragment definitions
        operation: GraphQL operation (query/mutation/subscription)
        variable_values: Query variables dict
        context: Execution context object
        root_value: Root value for execution
        path: Field path in query
        
    Usage:
        def resolve_user_posts(self, info):
            # Access field information
            field_name = info.field_name  # "userPosts"
            user = info.context.user
            
            # Access query variables
            limit = info.variable_values.get('limit', 10)
            
            # Access parent object
            user_id = self.id
            
            return get_posts(user_id, limit)
    """

class Dynamic(MountedType):
    """
    Runtime type resolution for lazy fields and circular references.
    
    Parameters:
        type_: Function that returns the actual GraphQL type
        with_schema: Whether to pass schema to type function (default: False)
    
    Methods:
        get_type(schema=None): Resolves type at schema build time
    
    Features:
    - Resolves circular import issues
    - Lazy type evaluation
    - Schema-aware type resolution
    - Integration with complex type hierarchies
    
    Usage:
        # Basic lazy type
        def get_user_type():
            return User
            
        friends = graphene.Dynamic(get_user_type)
        
        # Schema-aware resolution
        def get_type_from_schema(schema):
            return schema.get_type('CustomType')
            
        custom_field = graphene.Dynamic(
            get_type_from_schema, 
            with_schema=True
        )
        
        # In class definition
        class User(graphene.ObjectType):
            name = graphene.String()
            # Self-reference resolved at schema build time
            manager = graphene.Dynamic(lambda: User)
    """

import graphene
from graphene import relay

# Define types
class User(graphene.ObjectType):
    class Meta:
        interfaces = (relay.Node,)
    
    name = graphene.String()
    email = graphene.String()
    posts = relay.ConnectionField('PostConnection')
    created_at = graphene.DateTime()
    
    @classmethod
    def get_node(cls, info, id):
        return get_user_by_id(id)
    
    def resolve_posts(self, info, **args):
        return get_posts_for_user(self.id)

class Post(graphene.ObjectType):
    class Meta:
        interfaces = (relay.Node,)
    
    title = graphene.String()
    content = graphene.String()
    author = graphene.Field(User)
    published_at = graphene.DateTime()
    
    @classmethod
    def get_node(cls, info, id):
        return get_post_by_id(id)

# Connections
class PostConnection(relay.Connection):
    class Meta:
        node = Post

class UserConnection(relay.Connection):
    class Meta:
        node = User

# Query
class Query(graphene.ObjectType):
    # Node interface
    node = relay.Node.Field()
    
    # Collections
    users = relay.ConnectionField(UserConnection)
    posts = relay.ConnectionField(PostConnection)
    
    # Single objects
    user = graphene.Field(User, id=graphene.ID(required=True))
    post = graphene.Field(Post, id=graphene.ID(required=True))
    
    # Search
    search = graphene.List(
        graphene.Union(
            'SearchResult',
            types=(User, Post)
        ),
        query=graphene.String(required=True)
    )
    
    def resolve_users(self, info, **args):
        return User.objects.all()
    
    def resolve_posts(self, info, **args):
        return Post.objects.all()
    
    def resolve_user(self, info, id):
        return get_user_by_id(id)
    
    def resolve_post(self, info, id):
        return get_post_by_id(id)
    
    def resolve_search(self, info, query):
        results = []
        results.extend(search_users(query))
        results.extend(search_posts(query))
        return results

# Mutations
class CreateUser(graphene.Mutation):
    class Arguments:
        name = graphene.String(required=True)
        email = graphene.String(required=True)
    
    user = graphene.Field(User)
    success = graphene.Boolean()
    
    def mutate(self, info, name, email):
        # Validate
        if not email or '@' not in email:
            raise Exception('Invalid email')
        
        # Create user
        user = create_user(name=name, email=email)
        return CreateUser(user=user, success=True)

class CreatePost(graphene.Mutation):
    class Arguments:
        title = graphene.String(required=True)
        content = graphene.String(required=True)
        author_id = graphene.ID(required=True)
    
    post = graphene.Field(Post)
    
    def mutate(self, info, title, content, author_id):
        # Check authentication
        current_user = info.context.user
        if not current_user:
            raise Exception('Authentication required')
        
        # Create post
        post = create_post(
            title=title,
            content=content,
            author_id=author_id
        )
        return CreatePost(post=post)

class Mutation(graphene.ObjectType):
    create_user = CreateUser.Field()
    create_post = CreatePost.Field()

# Subscriptions (optional)
class Subscription(graphene.ObjectType):
    post_added = graphene.Field(Post)
    
    def resolve_post_added(self, info):
        # Return async iterator for real-time updates
        return post_subscription_stream()

# Create schema
schema = graphene.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription,
    auto_camelcase=True
)

import graphene
from datetime import datetime

# Context setup
class RequestContext(graphene.Context):
    def __init__(self, request, user=None, db_session=None):
        super().__init__()
        self.request = request
        self.user = user
        self.db_session = db_session
        self.timestamp = datetime.now()

# Execution function
def execute_graphql_query(query, variables=None, user=None, request=None):
    """Execute GraphQL query with proper context."""
    
    # Create context
    context = RequestContext(
        request=request,
        user=user,
        db_session=get_db_session()
    )
    
    try:
        # Execute query
        result = schema.execute(
            query,
            variable_values=variables,
            context_value=context
        )
        
        # Handle errors
        if result.errors:
            for error in result.errors:
                print(f"GraphQL Error: {error}")
        
        return {
            'data': result.data,
            'errors': [str(e) for e in result.errors] if result.errors else None
        }
    
    except Exception as e:
        return {
            'data': None,
            'errors': [f"Execution error: {str(e)}"]
        }
    
    finally:
        # Cleanup
        if context.db_session:
            context.db_session.close()

# Usage
query = '''
    query GetUserPosts($userId: ID!, $first: Int) {
        user(id: $userId) {
            name
            email
            posts(first: $first) {
                edges {
                    node {
                        title
                        publishedAt
                    }
                }
                pageInfo {
                    hasNextPage
                }
            }
        }
    }
'''

result = execute_graphql_query(
    query=query,
    variables={'userId': '1', 'first': 10},
    user=current_user,
    request=request
)

import graphene
from dataloader import DataLoader

class User(graphene.ObjectType):
    name = graphene.String()
    email = graphene.String()
    posts = graphene.List('Post')
    post_count = graphene.Int()
    
    # Simple resolver
    def resolve_name(self, info):
        return self.full_name or f"{self.first_name} {self.last_name}"
    
    # Resolver with arguments
    def resolve_posts(self, info, published_only=False, limit=None):
        posts = get_posts_for_user(self.id)
        
        if published_only:
            posts = [p for p in posts if p.is_published]
        
        if limit:
            posts = posts[:limit]
            
        return posts
    
    # Computed field with database access
    def resolve_post_count(self, info):
        db = info.context.db_session
        return db.query(Post).filter_by(author_id=self.id).count()
    
    # Async resolver
    async def resolve_async_data(self, info):
        async_service = info.context.async_service
        return await async_service.get_user_data(self.id)

# DataLoader integration for N+1 problem
def create_user_loader():
    async def load_users(user_ids):
        users = await get_users_by_ids(user_ids)
        return [users.get(id) for id in user_ids]
    
    return DataLoader(load_users)

class Post(graphene.ObjectType):
    title = graphene.String()
    author = graphene.Field(User)
    
    def resolve_author(self, info):
        # Use DataLoader to batch user requests
        user_loader = info.context.user_loader
        return user_loader.load(self.author_id)

# Context with DataLoader
class Context(graphene.Context):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.user_loader = create_user_loader()

import graphene
from graphql import GraphQLError

class CreateUser(graphene.Mutation):
    class Arguments:
        name = graphene.String(required=True)
        email = graphene.String(required=True)
        age = graphene.Int()
    
    user = graphene.Field(User)
    success = graphene.Boolean()
    errors = graphene.List(graphene.String)
    
    def mutate(self, info, name, email, age=None):
        errors = []
        
        # Input validation
        if len(name.strip()) < 2:
            errors.append("Name must be at least 2 characters")
        
        if not self.is_valid_email(email):
            errors.append("Invalid email format")
        
        if age is not None and (age < 0 or age > 150):
            errors.append("Age must be between 0 and 150")
        
        # Business logic validation
        if email_exists(email):
            errors.append("Email already exists")
        
        if errors:
            return CreateUser(success=False, errors=errors)
        
        try:
            # Create user
            user = create_user(name=name, email=email, age=age)
            return CreateUser(user=user, success=True, errors=[])
            
        except Exception as e:
            # Log error
            logger.error(f"User creation failed: {e}")
            
            # Return user-friendly error
            return CreateUser(
                success=False, 
                errors=["Failed to create user. Please try again."]
            )
    
    @staticmethod
    def is_valid_email(email):
        import re
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return re.match(pattern, email) is not None

# Custom exception handling
def format_error(error):
    """Custom error formatting for client consumption."""
    if isinstance(error, ValidationError):
        return {
            'message': error.message,
            'code': 'VALIDATION_ERROR',
            'field': error.field
        }
    elif isinstance(error, AuthenticationError):
        return {
            'message': 'Authentication required',
            'code': 'AUTH_ERROR'
        }
    else:
        # Log unexpected errors
        logger.error(f"Unexpected GraphQL error: {error}")
        return {
            'message': 'An unexpected error occurred',
            'code': 'INTERNAL_ERROR'
        }

# Use custom error formatting
schema = graphene.Schema(
    query=Query,
    mutation=Mutation,
    format_error=format_error
)