GraphQL Fundamentals

# Get user
GET /api/users/123

# Response: Fixed structure with ALL fields
{
    "id": 123,
    "name": "Alice",
    "email": "alice@example.com",
    "bio": "Software engineer",
    "avatar": "https://...",
    "follower_count": 1523,
    "following_count": 342,
    "created_at": "2020-01-15",
    "last_login": "2024-01-15T10:30:00Z"
}

# Mobile app only needs:
{
    "name": "Alice",
    "avatar": "https://..."
}

# But basic REST returns everything:
GET /api/users/123

{
    "id": 123,
    "name": "Alice",
    "email": "alice@example.com",           # ❌ Not needed
    "bio": "Software engineer",             # ❌ Not needed
    "avatar": "https://...",
    "follower_count": 1523,                 # ❌ Not needed
    "following_count": 342,                 # ❌ Not needed
    "created_at": "2020-01-15",             # ❌ Not needed
    "last_login": "2024-01-15T10:30:00Z"    # ❌ Not needed
}

# Wasted: 7 out of 9 fields (78% wasted bandwidth)

# User profile page needs:

# Request 1: User basic info
GET /api/users/123

# Request 2: User's posts
GET /api/users/123/posts
Response: {
    "posts": [
        {"id": 1, "title": "Post 1", "body": "...", "created_at": "..."},
        {"id": 2, "title": "Post 2", "body": "...", "created_at": "..."}
    ]
}

# Request 3: User's followers
GET /api/users/123/followers

# Request 4: User's following
GET /api/users/123/following

# Already 4 requests just for basic profile!
# Need comments too? Add more requests...

# Request 1: Get post comments
GET /api/posts/1/comments
Response: {
    "comments": [
        {"id": 1, "author_id": 456, "text": "Great post!"},
        {"id": 2, "author_id": 789, "text": "Thanks!"},
        {"id": 3, "author_id": 321, "text": "Awesome!"}
    ]
}

# Problem: Comments only have author_id, not author details
# Need to fetch each author separately:

# Request 2: Get author for comment 1
GET /api/users/456

# Request 3: Get author for comment 2
GET /api/users/789

# Request 4: Get author for comment 3
GET /api/users/321

# Total: 1 request for comments + N requests for authors = N+1 problem
# With 10 comments → 11 requests!
# With 100 comments → 101 requests!

# GraphQL: Single endpoint, client specifies exact data needed

# Query everything for user profile in ONE request
POST /graphql
{
  user(id: 123) {
    name
    avatar
    posts(limit: 5) {
      title
      createdAt
      comments {
        text
        author {
          name
          avatar
        }
      }
    }
    followerCount
  }
}

# Response: Exactly what was requested, nothing more
{
  "data": {
    "user": {
      "name": "Alice",
      "avatar": "https://...",
      "posts": [
        {
          "title": "Post 1",
          "createdAt": "2024-01-15",
          "comments": [
            {
              "text": "Great post!",
              "author": {
                "name": "Bob",
                "avatar": "https://..."
              }
            }
          ]
        }
      ],
      "followerCount": 1523
    }
  }
}

# Benefits:
# ✓ 1 request instead of 9
# ✓ No over-fetching (didn't include email, bio, etc.)
# ✓ No under-fetching (got user, posts, comments, authors in one query)
# ✓ Nested relationships resolved automatically


# Mobile app needs less data? Just request less:
{
  user(id: 123) {
    name
    avatar
  }
}

# Response:
{
  "data": {
    "user": {
      "name": "Alice",
      "avatar": "https://..."
    }
  }
}

# Same endpoint, different query, different response
# No need for separate /api/users/simplified endpoint

# GraphQL Schema Definition Language

# Scalar types (built-in primitives)
# - Int: 32-bit integer
# - Float: floating point number
# - String: UTF-8 string
# - Boolean: true or false
# - ID: unique identifier (string, but semantically an ID)


# Object type with fields
type User {
  id: ID!              # ! means required (non-nullable)
  name: String!        # Required string
  email: String        # Optional string (can be null)
  age: Int
  isActive: Boolean!
  posts: [Post!]!      # Required list of required Posts
  followerCount: Int!
}


# Another object type
type Post {
  id: ID!
  title: String!
  body: String!
  author: User!        # Reference to User type
  comments: [Comment!]!
  publishedAt: String  # ISO 8601 timestamp
}


# Nested type
type Comment {
  id: ID!
  text: String!
  author: User!
  post: Post!
  createdAt: String!
}


# Query type (entry point for reads)
type Query {
  # Get single user by ID
  user(id: ID!): User

  # Get all users
  users: [User!]!

  # Get user's posts
  userPosts(userId: ID!, limit: Int = 10): [Post!]!

  # Search posts
  searchPosts(query: String!): [Post!]!
}


# Mutation type (entry point for writes)
type Mutation {
  # Create user
  createUser(name: String!, email: String!): User!

  # Update user
  updateUser(id: ID!, name: String, email: String): User!

  # Delete user
  deleteUser(id: ID!): Boolean!

  # Create post
  createPost(userId: ID!, title: String!, body: String!): Post!
}


# Custom scalar for DateTime
scalar DateTime


# Enum type
enum UserRole {
  ADMIN
  MODERATOR
  USER
  GUEST
}


# Input type (for mutations)
input CreateUserInput {
  name: String!
  email: String!
  age: Int
  role: UserRole = USER  # Default value
}


# Interface (shared fields)
interface Node {
  id: ID!
  createdAt: DateTime!
}


# Union type (can be one of several types)
union SearchResult = User | Post | Comment

# Complete schema for a blog API

type User {
  id: ID!
  username: String!
  email: String!
  posts: [Post!]!
  comments: [Comment!]!
}

type Post {
  id: ID!
  title: String!
  body: String!
  author: User!
  comments: [Comment!]!
  tags: [String!]!
  publishedAt: String
  updatedAt: String!
}

type Comment {
  id: ID!
  text: String!
  author: User!
  post: Post!
  createdAt: String!
}

type Query {
  # Users
  user(id: ID!): User
  users(limit: Int = 10, offset: Int = 0): [User!]!

  # Posts
  post(id: ID!): Post
  posts(limit: Int = 10, offset: Int = 0): [Post!]!
  postsByUser(userId: ID!): [Post!]!

  # Search
  searchPosts(query: String!): [Post!]!
}

type Mutation {
  # User mutations
  createUser(username: String!, email: String!, password: String!): User!
  updateUser(id: ID!, username: String, email: String): User!

  # Post mutations
  createPost(title: String!, body: String!, tags: [String!]): Post!
  updatePost(id: ID!, title: String, body: String): Post!
  deletePost(id: ID!): Boolean!
  publishPost(id: ID!): Post!

  # Comment mutations
  createComment(postId: ID!, text: String!): Comment!
  deleteComment(id: ID!): Boolean!
}

# Schema defines the contract between client and server
# Client knows exactly what's available (introspection)
# Server validates all queries against schema automatically

# Basic query - get single user
query GetUser {
  user(id: "123") {
    name
    email
  }
}

# Response:
{
  "data": {
    "user": {
      "name": "Alice",
      "email": "alice@example.com"
    }
  }
}


# Query with nested fields
query GetUserWithPosts {
  user(id: "123") {
    name
    posts {
      title
      publishedAt
    }
  }
}

# Response:
{
  "data": {
    "user": {
      "name": "Alice",
      "posts": [
        {
          "title": "GraphQL Tutorial",
          "publishedAt": "2024-01-15"
        },
        {
          "title": "Python Best Practices",
          "publishedAt": "2024-01-10"
        }
      ]
    }
  }
}


# Query with arguments
query GetRecentPosts {
  posts(limit: 5, offset: 0) {
    title
    author {
      name
    }
  }
}


# Multiple queries in one request (parallel execution)
query GetDashboardData {
  currentUser: user(id: "123") {
    name
    email
  }

  recentPosts: posts(limit: 5) {
    title
    author {
      name
    }
  }

  stats: userStats(userId: "123") {
    postCount
    commentCount
  }
}


# Query with variables (for dynamic queries)
query GetUser($userId: ID!) {
  user(id: $userId) {
    name
    email
  }
}

# Variables (sent separately):
{
  "userId": "123"
}


# Fragments (reusable field sets)
query GetUsersWithPostInfo {
  user1: user(id: "123") {
    ...UserFields
  }

  user2: user(id: "456") {
    ...UserFields
  }
}

fragment UserFields on User {
  name
  email
  posts {
    title
    publishedAt
  }
}

# Resolver anatomy

def resolve_user(parent, info, id):
    """
    Resolver for user(id: ID!): User

    Args:
        parent: Parent object (None for Query root)
        info: GraphQL execution info (context, etc.)
        id: Query argument

    Returns:
        User object or None
    """
    # Fetch from database
    user = database.get_user(id)
    return user


def resolve_user_posts(parent, info, limit=10):
    """
    Resolver for User.posts field

    Args:
        parent: User object (parent field)
        info: Execution info
        limit: Argument with default

    Returns:
        List of Post objects
    """
    # parent is the User object
    user_id = parent.id

    # Fetch user's posts
    posts = database.get_posts_by_user(user_id, limit=limit)
    return posts


def resolve_post_author(parent, info):
    """
    Resolver for Post.author field

    Args:
        parent: Post object
        info: Execution info

    Returns:
        User object
    """
    # parent is the Post object
    author_id = parent.author_id

    # Fetch author
    author = database.get_user(author_id)
    return author


# Resolver execution flow for this query:
# query {
#   user(id: "123") {        # 1. resolve_user() called
#     name                   # 2. Return user.name (no resolver, direct field access)
#     posts {                # 3. resolve_user_posts(parent=user) called
#       title                # 4. Return post.title for each post
#       author {             # 5. resolve_post_author(parent=post) called for each
#         name               # 6. Return author.name
#       }
#     }
#   }
# }

# GraphQL executes resolvers as needed to build response tree

# Create mutation
mutation CreatePost {
  createPost(
    title: "GraphQL is Awesome"
    body: "GraphQL makes APIs flexible and efficient..."
    tags: ["graphql", "api"]
  ) {
    id
    title
    author {
      name
    }
    publishedAt
  }
}

# Response:
{
  "data": {
    "createPost": {
      "id": "789",
      "title": "GraphQL is Awesome",
      "author": {
        "name": "Alice"
      },
      "publishedAt": "2024-01-15T10:30:00Z"
    }
  }
}


# Update mutation
mutation UpdatePost {
  updatePost(
    id: "789"
    title: "GraphQL is Amazing"
    body: "Updated content..."
  ) {
    id
    title
    body
    updatedAt
  }
}


# Delete mutation
mutation DeletePost {
  deletePost(id: "789")  # Returns Boolean
}

# Response:
{
  "data": {
    "deletePost": true
  }
}


# Mutation with variables (best practice)
mutation CreatePost($input: CreatePostInput!) {
  createPost(input: $input) {
    id
    title
    author {
      name
    }
  }
}

# Variables:
{
  "input": {
    "title": "New Post",
    "body": "Content...",
    "tags": ["tutorial"]
  }
}


# Multiple mutations (executed sequentially, not parallel!)
mutation CreatePostAndComment {
  post: createPost(title: "Title", body: "Body") {
    id
  }

  comment: createComment(postId: "789", text: "Great!") {
    id
    text
  }
}

# Mutations run in order (comment after post)
# Queries run in parallel

def resolve_create_post(parent, info, title, body, tags=None):
    """
    Resolver for createPost mutation

    Args:
        parent: None (Mutation root)
        info: Execution info (includes context with user)
        title: Post title
        body: Post body
        tags: Optional list of tags

    Returns:
        Created Post object
    """
    # Get authenticated user from context
    user = info.context.get("user")
    if not user:
        raise Exception("Authentication required")

    # Validate input
    if len(title) < 5:
        raise Exception("Title must be at least 5 characters")

    # Create post in database
    post = database.create_post(
        title=title,
        body=body,
        author_id=user.id,
        tags=tags or [],
        published_at=datetime.now(timezone.utc)
    )

    # Return created post (GraphQL will resolve nested fields)
    return post


def resolve_update_post(parent, info, id, title=None, body=None):
    """
    Resolver for updatePost mutation

    Args:
        parent: None (Mutation root)
        info: Execution info
        id: Post ID to update
        title: Optional new title
        body: Optional new body

    Returns:
        Updated Post object
    """
    # Get authenticated user
    user = info.context.get("user")
    if not user:
        raise Exception("Authentication required")

    # Fetch post
    post = database.get_post(id)
    if not post:
        raise Exception("Post not found")

    # Check authorization (only author can update)
    if post.author_id != user.id:
        raise Exception("Not authorized to update this post")

    # Update fields
    update_data = {}
    if title is not None:
        update_data["title"] = title
    if body is not None:
        update_data["body"] = body

    if update_data:
        update_data["updated_at"] = datetime.now(timezone.utc)
        post = database.update_post(id, **update_data)

    return post


def resolve_delete_post(parent, info, id):
    """
    Resolver for deletePost mutation

    Args:
        parent: None (Mutation root)
        info: Execution info
        id: Post ID to delete

    Returns:
        Boolean (True if deleted)
    """
    # Get authenticated user
    user = info.context.get("user")
    if not user:
        raise Exception("Authentication required")

    # Fetch post
    post = database.get_post(id)
    if not post:
        raise Exception("Post not found")

    # Check authorization
    if post.author_id != user.id and not user.is_admin:
        raise Exception("Not authorized to delete this post")

    # Delete post
    database.delete_post(id)

    return True


# Error handling
# GraphQL returns errors in response:
{
  "data": {
    "createPost": null
  },
  "errors": [
    {
      "message": "Title must be at least 5 characters",
      "path": ["createPost"]
    }
  ]
}

pip install strawberry-graphql[fastapi]

# database.py
import threading
import uuid
from typing import Dict, List, Optional, Any
from datetime import datetime, timezone


class InMemoryDatabase:
    """
    A thread-safe In-Memory Database for relational data.
    Designed to be dropped into Strawberry GraphQL resolvers.
    """

    def __init__(self):
        # The 'tables'
        self._users: Dict[str, Any] = {}
        self._posts: Dict[str, Any] = {}
        self._comments: Dict[str, Any] = {}

        # Thread safety lock
        self._lock = threading.Lock()

        # Seed initial data
        self._bootstrap()

    def _bootstrap(self):
        """Initialize with some starting records."""
        u1_id = self.create_user("alice", "alice@example.com")["id"]
        u2_id = self.create_user("bob", "bob@example.com")["id"]

        p1 = self.create_post(
            "GraphQL vs REST",
            "A deep dive into API design.",
            u1_id,
            datetime.now(timezone.utc).isoformat()
        )

        self.create_comment("Great post!", u2_id, p1["id"])

    # --- User Logic ---
    def create_user(self, username: str, email: str) -> Dict:
        with self._lock:
            user_id = f"user_{uuid.uuid4().hex[:8]}"
            user = {"id": user_id, "username": username, "email": email}
            self._users[user_id] = user
            return user

    def get_user(self, user_id: str) -> Optional[Dict]:
        return self._users.get(user_id)

    def get_users(self, limit: int = 10) -> List[Dict]:
        return list(self._users.values())[:limit]

    # --- Post Logic ---
    def create_post(self, title: str, body: str, author_id: str, published_at: str) -> Dict:
        with self._lock:
            post_id = f"post_{uuid.uuid4().hex[:8]}"
            post = {
                "id": post_id,
                "title": title,
                "body": body,
                "author_id": author_id,
                "published_at": published_at
            }
            self._posts[post_id] = post
            return post

    def get_post(self, post_id: str) -> Optional[Dict]:
        return self._posts.get(post_id)

    def get_posts(self, limit: int = 10, offset: int = 0) -> List[Dict]:
        all_posts = list(self._posts.values())
        return all_posts[offset: offset + limit]

    def get_posts_by_user(self, author_id: str) -> List[Dict]:
        return [p for p in self._posts.values() if p["author_id"] == author_id]

    def update_post(self, post_id: str, **kwargs) -> Optional[Dict]:
        with self._lock:
            if post_id in self._posts:
                self._posts[post_id].update(kwargs)
                return self._posts[post_id]
            return None

    def delete_post(self, post_id: str) -> bool:
        with self._lock:
            if post_id in self._posts:
                del self._posts[post_id]
                return True
            return False

    # --- Comment Logic ---
    def create_comment(self, text: str, author_id: str, post_id: str) -> Dict:
        with self._lock:
            comment_id = f"comment_{uuid.uuid4().hex[:8]}"
            comment = {
                "id": comment_id,
                "text": text,
                "author_id": author_id,
                "post_id": post_id
            }
            self._comments[comment_id] = comment
            return comment

    def get_comments_by_post(self, post_id: str) -> List[Dict]:
        return [c for c in self._comments.values() if c["post_id"] == post_id]


# Initialize database
database = InMemoryDatabase()

# schema.py
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
import strawberry
from typing import List, Optional
from datetime import datetime, timezone
from database import database


# Define types using Strawberry decorators
@strawberry.type
class User:
    """User type with fields"""
    id: strawberry.ID
    username: str
    email: str

    @strawberry.field
    def posts(self) -> List["Post"]:
        """Resolver for posts field"""
        posts_data = database.get_posts_by_user(self.id)
        return [Post(**p) for p in posts_data]


@strawberry.type
class Post:
    """Post type"""
    id: strawberry.ID
    title: str
    body: str
    author_id: strawberry.ID
    published_at: Optional[str] = None

    @strawberry.field
    def author(self) -> User:
        """Resolver for author field"""
        user_data = database.get_user(self.author_id)
        if user_data:
            return User(**user_data)
        return None

    @strawberry.field
    def comments(self) -> List["Comment"]:
        """Resolver for comments field"""
        comments_data = database.get_comments_by_post(self.id)
        return [Comment(**c) for c in comments_data]


@strawberry.type
class Comment:
    """Comment type"""
    id: strawberry.ID
    text: str
    author_id: strawberry.ID
    post_id: strawberry.ID

    @strawberry.field
    def author(self) -> User:
        user_data = database.get_user(self.author_id)
        if user_data:
            return User(**user_data)
        return None


# Define Query type
@strawberry.type
class Query:
    """
    Root query type.

    Important: Database returns dictionaries, but Strawberry types expect objects.
    Use **dict unpacking to convert: User(**user_data)
    """

    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        """Get user by ID"""
        user_data = database.get_user(id)
        if user_data:
            return User(**user_data)  # Convert dict to User object
        return None

    @strawberry.field
    def users(self, limit: int = 10) -> List[User]:
        """Get all users"""
        users_data = database.get_users(limit=limit)
        return [User(**u) for u in users_data]  # Convert each dict to User object

    @strawberry.field
    def post(self, id: strawberry.ID) -> Optional[Post]:
        """Get post by ID"""
        post_data = database.get_post(id)
        if post_data:
            return Post(**post_data)  # Convert dict to Post object
        return None

    @strawberry.field
    def posts(self, limit: int = 10, offset: int = 0) -> List[Post]:
        """Get all posts"""
        posts_data = database.get_posts(limit=limit, offset=offset)
        return [Post(**p) for p in posts_data]  # Convert each dict to Post object


# Define Input types for mutations
@strawberry.input
class CreatePostInput:
    title: str
    body: str
    tags: List[str] = strawberry.field(default_factory=list)


# Define Mutation type
@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_post(self, input: CreatePostInput, info: strawberry.Info) -> Post:
        """Create new post"""
        # Get user from context
        user = info.context.get("user")
        if not user:
            raise Exception("Authentication required")

        # Create post
        post_data = database.create_post(
            title=input.title,
            body=input.body,
            author_id=user.id,
            published_at=datetime.now(timezone.utc).isoformat()
        )

        return Post(**post_data)

    @strawberry.mutation
    def update_post(
        self,
        id: strawberry.ID,
        title: Optional[str] = None,
        body: Optional[str] = None,
        info: strawberry.Info = None
    ) -> Post:
        """Update post"""
        user = info.context.get("user")
        if not user:
            raise Exception("Authentication required")

        post_data = database.get_post(id)
        if not post_data:
            raise Exception("Post not found")

        if post_data["author_id"] != user.id:
            raise Exception("Not authorized")

        update_data = {}
        if title:
            update_data["title"] = title
        if body:
            update_data["body"] = body

        updated_post = database.update_post(id, **update_data)
        return Post(**updated_post)

    @strawberry.mutation
    def delete_post(self, id: strawberry.ID, info: strawberry.Info) -> bool:
        """Delete post"""
        user = info.context.get("user")
        if not user:
            raise Exception("Authentication required")

        post_data = database.get_post(id)
        if not post_data:
            raise Exception("Post not found")

        if post_data["author_id"] != user.id:
            raise Exception("Not authorized")

        database.delete_post(id)
        return True


# Create schema
schema = strawberry.Schema(query=Query, mutation=Mutation)

# Create FastAPI app
app = FastAPI()

# Add GraphQL router
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")

uvicorn schema:app --reload

pip install PyJWT

# auth.py
from fastapi import Request, Depends
from strawberry.fastapi import GraphQLRouter
import jwt
from database import database

SECRET_KEY = "your-secret-key"


def get_current_user(request: Request):
    """
    Extract user from JWT token in Authorization header.

    Returns User object or None if not authenticated.
    """
    auth_header = request.headers.get("Authorization")

    if not auth_header:
        return None

    try:
        # Extract token from "Bearer <token>"
        token = auth_header.replace("Bearer ", "")

        # Decode JWT
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])

        # Get user from database
        user = database.get_user(payload["user_id"])
        return user

    except (jwt.InvalidTokenError, KeyError):
        return None


async def get_context(
    request: Request,
    user = Depends(get_current_user)
):
    """
    Build GraphQL context for each request.

    Context is passed to all resolvers via info.context
    """
    return {
        "request": request,
        "user": user
    }


# Create GraphQL router with context
graphql_app = GraphQLRouter(
    schema,
    context_getter=get_context
)

app.include_router(graphql_app, prefix="/graphql")


# Now resolvers can access user:
@strawberry.mutation
def create_post(self, input: CreatePostInput, info: strawberry.Info) -> Post:
    user = info.context["user"]  # Get authenticated user
    if not user:
        raise Exception("Authentication required")

    # Create post for authenticated user
    post = database.create_post(
        title=input.title,
        body=input.body,
        author_id=user.id
    )

    return post


# Client sends requests with Authorization header:
# curl -X POST http://localhost:8000/graphql #   -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." #   -H "Content-Type: application/json" #   -d '{"query": "mutation { createPost(input: {title: "Test"}) { id } }"}'

# Query: Get posts with authors
query {
  posts(limit: 100) {
    title
    author {
      username
    }
  }
}


# Naive resolver implementation:
@strawberry.field
def posts(self, limit: int = 10) -> List[Post]:
    # Query 1: Fetch all posts
    posts = database.execute("SELECT * FROM posts LIMIT %s", [limit])
    return posts


@strawberry.field
def author(self) -> User:
    # Called for EACH post!
    # Query 2-101: Fetch author for each post
    author = database.execute("SELECT * FROM users WHERE id = %s", [self.author_id])
    return author


# Total queries:
# 1 query for posts
# + 100 queries for authors (one per post)
# = 101 queries!

# Database logs:
SELECT * FROM posts LIMIT 100
SELECT * FROM users WHERE id = 1
SELECT * FROM users WHERE id = 2
SELECT * FROM users WHERE id = 3
...
SELECT * FROM users WHERE id = 100

# Problem:
# - 101 database queries for simple request
# - Each query has latency (1-10ms)
# - Total time: 100-1000ms instead of 10-20ms
# - Database connection pool exhaustion
# - Scales horribly (1000 posts = 1001 queries!)

pip install strawberry-graphql[dataloader]

from strawberry.dataloader import DataLoader
from typing import List
from database import database

# Create DataLoader for users
async def load_users(keys: List[int]) -> List[User]:
    """
    Batch function - receives list of user IDs,
    returns list of users in same order.

    Called once per request with all accumulated IDs.
    """
    print(f"Loading users: {keys}")

    # Single query with IN clause
    query = "SELECT * FROM users WHERE id IN (%s)" % ",".join(str(k) for k in keys)
    users = database.execute(query)

    # Create mapping
    user_map = {user.id: user for user in users}

    # Return in same order as keys (important!)
    return [user_map.get(key) for key in keys]


# Create DataLoader instance
user_loader = DataLoader(load_fn=load_users)


# Use DataLoader in resolver
@strawberry.field
async def author(self, info: strawberry.Info) -> User:
    """
    Fetch author using DataLoader.

    Instead of immediate database query,
    DataLoader accumulates all author_ids,
    then batches them into single query.
    """
    # Get loader from context
    loader = info.context["user_loader"]

    # Load user (batched automatically)
    user = await loader.load(self.author_id)

    return user


# Add DataLoader to context
async def get_context(request: Request):
    return {
        "user_loader": DataLoader(load_fn=load_users)
    }


# Execution flow with 100 posts:
# 1. Query posts
# 2. For each post, resolver calls loader.load(author_id)
# 3. DataLoader accumulates: [1, 2, 3, ..., 100]
# 4. After all resolvers enqueued, DataLoader calls load_users([1,2,3,...,100])
# 5. Single query: SELECT * FROM users WHERE id IN (1,2,3,...,100)
# 6. Results distributed to waiting resolvers

# Database logs with DataLoader:
SELECT * FROM posts LIMIT 100
SELECT * FROM users WHERE id IN (1,2,3,4,5,...,100)

# Total: 2 queries instead of 101!

# DataLoaders also cache within a request

# Query: Posts with authors AND comments with authors
query {
  posts {
    title
    author {      # Author loaded here
      name
    }
    comments {
      text
      author {    # Same author? Cached, no duplicate fetch!
        name
      }
    }
  }
}


# Without cache:
# 1. Fetch posts
# 2. Batch fetch post authors
# 3. Fetch comments for each post
# 4. Batch fetch comment authors (duplicates post authors!)

# With DataLoader cache:
# 1. Fetch posts
# 2. Batch fetch post authors (cached)
# 3. Fetch comments
# 4. Batch fetch comment authors
#    - Author ID 1 already cached from step 2 → skip!
#    - Author ID 5 already cached from step 2 → skip!
#    - Only fetch new authors

# Result:
# - No duplicate fetches for same user
# - Further reduces queries
# - Cache is per-request (doesn't leak between requests)

import strawberry
from strawberry.permission import BasePermission
from typing import Any

# Custom permission classes
class IsAuthenticated(BasePermission):
    """Require authentication"""

    message = "User is not authenticated"

    def has_permission(self, source: Any, info: strawberry.Info, **kwargs) -> bool:
        return info.context.get("user") is not None


class IsAdmin(BasePermission):
    """Require admin role"""

    message = "Admin access required"

    def has_permission(self, source: Any, info: strawberry.Info, **kwargs) -> bool:
        user = info.context.get("user")
        return user and user.role == "admin"


class IsOwner(BasePermission):
    """Require ownership of resource"""

    message = "You don't have permission to access this resource"

    def has_permission(self, source: Any, info: strawberry.Info, **kwargs) -> bool:
        user = info.context.get("user")
        if not user:
            return False

        # source is the parent object (e.g., Post)
        # Check if user owns it
        return source.author_id == user.id


# Apply permissions to fields
@strawberry.type
class User:
    id: strawberry.ID
    username: str

    @strawberry.field(permission_classes=[IsAuthenticated])
    def email(self) -> str:
        """Email only visible to authenticated users"""
        return self._email

    @strawberry.field(permission_classes=[IsOwner])
    def private_notes(self) -> str:
        """Private notes only visible to owner"""
        return self._private_notes


# Apply permissions to mutations
@strawberry.type
class Mutation:
    @strawberry.mutation(permission_classes=[IsAuthenticated])
    def create_post(self, input: CreatePostInput, info: strawberry.Info) -> Post:
        """Only authenticated users can create posts"""
        user = info.context["user"]

        post = database.create_post(
            title=input.title,
            body=input.body,
            author_id=user.id
        )

        return post

    @strawberry.mutation(permission_classes=[IsAdmin])
    def delete_user(self, id: strawberry.ID) -> bool:
        """Only admins can delete users"""
        database.delete_user(id)
        return True


# Query with permission check
@strawberry.type
class Query:
    @strawberry.field(permission_classes=[IsAdmin])
    def all_users(self) -> List[User]:
        """Admin-only query"""
        return database.get_all_users()


# When permission denied, GraphQL returns error:
{
  "data": {
    "createPost": null
  },
  "errors": [
    {
      "message": "User is not authenticated",
      "path": ["createPost"]
    }
  ]
}

@strawberry.mutation
def update_post(
    self,
    id: strawberry.ID,
    title: Optional[str],
    info: strawberry.Info
) -> Post:
    """Update post with custom authorization logic"""

    # Get authenticated user
    user = info.context.get("user")
    if not user:
        raise Exception("Authentication required")

    # Fetch post
    post = database.get_post(id)
    if not post:
        raise Exception("Post not found")

    # Check authorization - owner or admin can update
    is_owner = post.author_id == user.id
    is_admin = user.role == "admin"

    if not (is_owner or is_admin):
        raise Exception("Not authorized to update this post")

    # Update post
    if title:
        post = database.update_post(id, title=title)

    return post


# Hide sensitive fields based on context
@strawberry.type
class User:
    id: strawberry.ID
    username: str

    @strawberry.field
    def email(self, info: strawberry.Info) -> Optional[str]:
        """
        Email visibility rules:
        - Own email: always visible
        - Other's email: only if authenticated and friends
        - Public: not visible
        """
        current_user = info.context.get("user")

        # Not authenticated - hide email
        if not current_user:
            return None

        # Own email - show
        if current_user.id == self.id:
            return self._email

        # Friend's email - show
        if database.are_friends(current_user.id, self.id):
            return self._email

        # Otherwise - hide
        return None


# Rate limiting per user
from collections import defaultdict
import time

request_counts = defaultdict(list)

def check_rate_limit(user_id: str, limit: int = 100, window: int = 60) -> bool:
    """Check if user exceeded rate limit"""
    now = time.time()
    window_start = now - window

    # Remove old timestamps
    timestamps = request_counts[user_id]
    timestamps[:] = [ts for ts in timestamps if ts > window_start]

    # Check limit
    if len(timestamps) >= limit:
        return False

    timestamps.append(now)
    return True


@strawberry.type
class Query:
    @strawberry.field
    def expensive_query(self, info: strawberry.Info) -> str:
        """Rate-limited expensive query"""
        user = info.context.get("user")
        if not user:
            raise Exception("Authentication required")

        # Check rate limit
        if not check_rate_limit(user.id, limit=10, window=60):
            raise Exception("Rate limit exceeded. Max 10 requests per minute.")

        # Execute expensive operation
        result = perform_expensive_computation()
        return result

# Problem: Malicious or accidental expensive queries

# Deeply nested query (DoS attack)
query MaliciousQuery {
  users {
    posts {
      comments {
        author {
          posts {
            comments {
              author {
                posts {
                  # ... infinite nesting possible!
                }
              }
            }
          }
        }
      }
    }
  }
}

# Result: Massive database load, server crash


# Solution: Query depth limiting
from strawberry.extensions import QueryDepthLimiter

schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    extensions=[
        QueryDepthLimiter(max_depth=5)  # Limit to 5 levels deep
    ]
)

# Query depth > 5 rejected:
{
  "errors": [
    {
      "message": "Query exceeds maximum depth of 5"
    }
  ]
}


# Query complexity analysis
# Assign cost to each field, reject if total > limit

from strawberry.extensions import QueryCostAnalyzer

@strawberry.type
class Query:
    @strawberry.field
    def users(self, limit: int = 10) -> List[User]:
        """Cost: 1 + limit (fetching N users)"""
        return database.get_users(limit)

    @strawberry.field
    def expensive_report(self) -> Report:
        """Cost: 100 (very expensive operation)"""
        return generate_report()


# Extension to calculate and limit cost
schema = strawberry.Schema(
    query=Query,
    extensions=[
        QueryCostAnalyzer(
            max_cost=1000  # Reject queries with cost > 1000
        )
    ]
)


# Timeout protection
import asyncio

async def timeout_wrapper(resolver, timeout=5):
    """Wrap resolver with timeout"""
    try:
        return await asyncio.wait_for(resolver(), timeout=timeout)
    except asyncio.TimeoutError:
        raise Exception(f"Query timeout after {timeout}s")

# Registration:
POST /graphql/register
{
  "query": "query GetUser($id: ID!) { user(id: $id) { name email } }",
  "id": "abc123"
}

# Usage:
POST /graphql
{
  "queryId": "abc123",
  "variables": {"id": "123"}
}

# Benefits:
# - Smaller request size (ID instead of full query)
# - Can cache responses by ID + variables
# - Prevents arbitrary queries (security)

pip install redis

import redis
from functools import wraps
import json

redis_client = redis.Redis()


def cached_field(ttl: int = 300):
    """Cache field resolver results"""
    def decorator(func):
        @wraps(func)
        async def wrapper(self, info: strawberry.Info, **kwargs):
            # Generate cache key from field name + arguments
            cache_key = f"gql:{func.__name__}:{self.id}:{kwargs}"

            # Try cache
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)

            # Compute
            result = await func(self, info, **kwargs)

            # Cache
            redis_client.setex(cache_key, ttl, json.dumps(result))

            return result
        return wrapper
    return decorator


@strawberry.type
class User:
    id: strawberry.ID

    @strawberry.field
    @cached_field(ttl=300)  # Cache for 5 minutes
    async def posts(self, limit: int = 10) -> List[Post]:
        """Cached field - expensive to compute"""
        return await database.get_posts_by_user(self.id, limit)

# GraphQL GET request
GET /graphql?query={user(id:"123"){name}}&variables={}

# Response with Cache-Control header:
HTTP/1.1 200 OK
Cache-Control: public, max-age=300
Content-Type: application/json

{
  "data": {
    "user": {
      "name": "Alice"
    }
  }
}

# CDN/browser can cache this response for 5 minutes!

from prometheus_client import Counter, Histogram

# GraphQL-specific metrics
GRAPHQL_REQUESTS = Counter(
    'graphql_requests_total',
    'GraphQL requests',
    ['operation_type', 'operation_name']
)

GRAPHQL_ERRORS = Counter(
    'graphql_errors_total',
    'GraphQL errors',
    ['error_type']
)

GRAPHQL_RESOLVER_DURATION = Histogram(
    'graphql_resolver_duration_seconds',
    'Resolver execution time',
    ['type_name', 'field_name']
)

GRAPHQL_QUERY_DEPTH = Histogram(
    'graphql_query_depth',
    'Query depth',
    buckets=[1, 2, 3, 5, 10, 20]
)


# Extension to collect metrics
from strawberry.extensions import Extension

class MetricsExtension(Extension):
    def on_operation(self):
        # Track operation
        operation_type = self.execution_context.operation_type
        operation_name = self.execution_context.operation_name

        GRAPHQL_REQUESTS.labels(
            operation_type=operation_type,
            operation_name=operation_name
        ).inc()

    def on_error(self, error):
        # Track errors
        GRAPHQL_ERRORS.labels(error_type=type(error).__name__).inc()


# Add to schema
schema = strawberry.Schema(
    query=Query,
    extensions=[MetricsExtension]
)


# Log slow queries
import logging

logger = logging.getLogger(__name__)

class SlowQueryLogger(Extension):
    def on_operation_end(self):
        duration = self.execution_context.duration

        if duration > 1.0:  # Slower than 1 second
            query = self.execution_context.query
            variables = self.execution_context.variables

            logger.warning(
                f"Slow query detected: {duration:.2f}s\n"
                f"Query: {query}\n"
                f"Variables: {variables}"
            )