undefined

    query = """
        query GetUser($id: ID!) {
            user(id: $id) {
                id
                email
                name
            }
        }
    """

    # Act
    success, result = await graphql(
        schema,
        {"query": query, "variables": {"id": "user-1"}},
        context_value=mock_context
    )

    # Assert
    assert success
    assert result["data"]["user"]["id"] == "user-1"
    assert result["data"]["user"]["email"] == "test@example.com"
    mock_context["loaders"]["user_loader"].load.assert_called_once_with("user-1")

@pytest.mark.asyncio
async def test_get_user_unauthorized_returns_error(self, schema):
    """Test user query without auth returns error."""
    # Arrange - no user in context
    context = {"user": None, "loaders": {}}

    query = """
        query GetUser($id: ID!) {
            user(id: $id) {
                id
                email
            }
        }
    """

    # Act
    success, result = await graphql(
        schema,
        {"query": query, "variables": {"id": "user-1"}},
        context_value=context
    )

    # Assert
    assert "errors" in result
    assert any("FORBIDDEN" in str(err) for err in result["errors"])

    mutation = """
        mutation CreatePost($input: CreatePostInput!) {
            createPost(input: $input) {
                post {
                    id
                    title
                    content
                }
                errors {
                    message
                    code
                }
            }
        }
    """

    variables = {
        "input": {
            "title": "Test Post",
            "content": "Test content"
        }
    }

    # Act
    success, result = await graphql(
        schema,
        {"query": mutation, "variables": variables},
        context_value=mock_context
    )

    # Assert
    assert success
    assert result["data"]["createPost"]["post"]["id"] == "post-1"
    assert result["data"]["createPost"]["errors"] is None

@pytest.mark.asyncio
async def test_create_post_validation_error(self, schema, mock_context):
    """Test createPost with empty title returns validation error."""
    mutation = """
        mutation CreatePost($input: CreatePostInput!) {
            createPost(input: $input) {
                post {
                    id
                }
                errors {
                    message
                    field
                    code
                }
            }
        }
    """

    variables = {
        "input": {
            "title": "",  # Invalid - empty title
            "content": "Test content"
        }
    }

    # Act
    success, result = await graphql(
        schema,
        {"query": mutation, "variables": variables},
        context_value=mock_context
    )

    # Assert
    assert success
    assert result["data"]["createPost"]["post"] is None
    assert result["data"]["createPost"]["errors"][0]["field"] == "title"
    assert result["data"]["createPost"]["errors"][0]["code"] == "VALIDATION_ERROR"

    # Track how many times batch function is called
    batch_calls = []

    async def batch_load_users(user_ids):
        batch_calls.append(list(user_ids))
        return [{"id": uid, "name": f"User {uid}"} for uid in user_ids]

    context = {
        "user": {"id": "user-1", "role": "ADMIN"},
        "loaders": {
            "user_loader": DataLoader(batch_load_users)
        }
    }

    query = """
        query GetPosts {
            posts(first: 3) {
                edges {
                    node {
                        id
                        author {
                            id
                            name
                        }
                    }
                }
            }
        }
    """

    # Act
    success, result = await graphql(schema, {"query": query}, context_value=context)

    # Assert - should batch all author loads into single call
    assert success
    assert len(batch_calls) == 1  # Only one batch call, not N calls

undefined

undefined

# Validation
if not input.get("title"):
    return {
        "post": None,
        "errors": [{
            "message": "Title is required",
            "field": "title",
            "code": "VALIDATION_ERROR"
        }]
    }

# Create post
post = await context["db"].create_post({
    **input,
    "authorId": context["user"]["id"]
})

return {"post": post, "errors": None}

undefined

undefined

---

undefined

**Good - Batched Loading:**
```python

undefined

undefined

**Good - Complexity and Depth Limits:**
```python

undefined

undefined

**Good - Cached Responses:**
```python

async def get_or_set(self, key, fetch_func):
    now = datetime.utcnow()
    if key in self._cache:
        if now - self._timestamps[key] < self._ttl:
            return self._cache[key]

    value = await fetch_func()
    self._cache[key] = value
    self._timestamps[key] = now
    return value

undefined

undefined

**Good - Cursor-Based Pagination:**
```python

if after:
    cursor_id = decode_cursor(after)
    query += " WHERE id > ?"
    params.append(cursor_id)

query += " ORDER BY id LIMIT ?"
params.append(first + 1)  # Fetch one extra to check hasNextPage

posts = await db.query(query, *params)

has_next = len(posts) > first
if has_next:
    posts = posts[:first]

return {
    "edges": [
        {"node": post, "cursor": encode_cursor(post["id"])}
        for post in posts
    ],
    "pageInfo": {
        "hasNextPage": has_next,
        "endCursor": encode_cursor(posts[-1]["id"]) if posts else None
    }
}

undefined

undefined

**Good - Proper Async Operations:**
```python

    user, posts, stats = await asyncio.gather(
        user_task, posts_task, stats_task
    )

    return {
        "user": user.json(),
        "posts": posts.json(),
        "stats": stats.json()
    }

---

undefined

```typescript
// codegen.ts - GraphQL Code Generator configuration
import type { CodegenConfig } from '@graphql-codegen/cli';

const config: CodegenConfig = {
  schema: './schema.graphql',
  generates: {
    './src/types/graphql.ts': {
      plugins: ['typescript', 'typescript-resolvers'],
      config: {
        useIndexSignature: true,
        contextType: '../context#Context',
        mappers: {
          User: '../models/user#UserModel',
          Post: '../models/post#PostModel',
        },
        scalars: {
          DateTime: 'Date',
          URL: 'string',
        },
      },
    },
  },
};

export default config;

import DataLoader from 'dataloader';
import { User, Post } from './models';

// ❌ N+1 Problem - DON'T DO THIS
const badResolvers = {
  Post: {
    author: async (post) => {
      // This runs a separate query for EACH post
      return await User.findById(post.authorId);
    },
  },
};

// ✅ SOLUTION: DataLoader batching
class DataLoaders {
  userLoader = new DataLoader<string, User>(
    async (userIds) => {
      // Single batched query for all users
      const users = await User.findMany({
        where: { id: { in: [...userIds] } },
      });

      // Return users in the same order as requested IDs
      const userMap = new Map(users.map(u => [u.id, u]));
      return userIds.map(id => userMap.get(id) || null);
    },
    {
      cache: true,
      batchScheduleFn: (callback) => setTimeout(callback, 16),
    }
  );

  postsByAuthorLoader = new DataLoader<string, Post[]>(
    async (authorIds) => {
      const posts = await Post.findMany({
        where: { authorId: { in: [...authorIds] } },
      });

      const postsByAuthor = new Map<string, Post[]>();
      authorIds.forEach(id => postsByAuthor.set(id, []));
      posts.forEach(post => {
        const authorPosts = postsByAuthor.get(post.authorId) || [];
        authorPosts.push(post);
        postsByAuthor.set(post.authorId, authorPosts);
      });

      return authorIds.map(id => postsByAuthor.get(id) || []);
    }
  );
}

// Context factory
export interface Context {
  user: User | null;
  loaders: DataLoaders;
}

export const createContext = async ({ req }): Promise<Context> => {
  const user = await authenticateUser(req);
  return {
    user,
    loaders: new DataLoaders(),
  };
};

// Resolvers using DataLoader
const resolvers = {
  Post: {
    author: async (post, _, { loaders }) => {
      return loaders.userLoader.load(post.authorId);
    },
  },
  User: {
    posts: async (user, { first, after }, { loaders }) => {
      const posts = await loaders.postsByAuthorLoader.load(user.id);
      return paginatePosts(posts, first, after);
    },
  },
};

import { GraphQLError } from 'graphql';
import { shield, rule, and, or } from 'graphql-shield';

// ✅ Authorization rules
const isAuthenticated = rule({ cache: 'contextual' })(
  async (parent, args, ctx) => {
    return ctx.user !== null;
  }
);

const isAdmin = rule({ cache: 'contextual' })(
  async (parent, args, ctx) => {
    return ctx.user?.role === 'ADMIN';
  }
);

const isPostOwner = rule({ cache: 'strict' })(
  async (parent, args, ctx) => {
    const post = await ctx.loaders.postLoader.load(args.id);
    return post?.authorId === ctx.user?.id;
  }
);

// ✅ Permission layer
const permissions = shield(
  {
    Query: {
      me: isAuthenticated,
      user: isAuthenticated,
      posts: true, // Public
    },
    Mutation: {
      createPost: isAuthenticated,
      updatePost: and(isAuthenticated, or(isPostOwner, isAdmin)),
      deletePost: and(isAuthenticated, or(isPostOwner, isAdmin)),
    },
    User: {
      email: isAuthenticated, // Only authenticated users see emails
      posts: true, // Public field
    },
  },
  {
    allowExternalErrors: false,
    fallbackError: new GraphQLError('Not authorized', {
      extensions: { code: 'FORBIDDEN' },
    }),
  }
);

// ❌ DON'T - Causes N+1 queries
const resolvers = {
  Post: {
    author: (post) => db.query('SELECT * FROM users WHERE id = ?', [post.authorId]),
  },
};

// ✅ DO - Use DataLoader
const resolvers = {
  Post: {
    author: (post, _, { loaders }) => loaders.userLoader.load(post.authorId),
  },
};

// ❌ DON'T - Allow unlimited queries
const server = new ApolloServer({ typeDefs, resolvers });

// ✅ DO - Add complexity limits
const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [depthLimit(7), complexityLimit(1000)],
});

// ❌ DON'T - Public access to all fields
type User {
  email: String!
  socialSecurityNumber: String!
}

// ✅ DO - Field-level authorization
type User {
  email: String! @auth
  socialSecurityNumber: String! @auth(requires: ADMIN)
}

undefined

    success, result = await graphql(
        schema, {"query": query}, context_value=auth_context
    )

    assert success
    assert result["data"]["me"]["id"] == "user-1"

@pytest.mark.asyncio
async def test_unauthorized_query_returns_error(self, schema):
    query = "query { me { id } }"
    context = {"user": None, "loaders": {}}

    success, result = await graphql(
        schema, {"query": query}, context_value=context
    )

    assert "errors" in result

    success, result = await graphql(
        schema, {"query": mutation}, context_value=auth_context
    )

    assert result["data"]["createPost"]["errors"][0]["field"] == "title"

undefined

undefined

    assert response.status_code == 200
    data = response.json()
    assert "data" in data
    assert "posts" in data["data"]

@pytest.mark.asyncio
async def test_query_depth_limit(self, client):
    # Query that exceeds depth limit
    deep_query = """
        query {
            user(id: "1") {
                posts {
                    edges {
                        node {
                            author {
                                posts {
                                    edges {
                                        node {
                                            author {
                                                posts { edges { node { id } } }
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    """

    response = await client.post("/graphql", json={"query": deep_query})
    data = response.json()

    assert "errors" in data
    assert any("depth" in str(err).lower() for err in data["errors"])

@pytest.mark.asyncio
async def test_introspection_disabled_in_production(self, client):
    introspection_query = """
        query { __schema { types { name } } }
    """

    response = await client.post(
        "/graphql",
        json={"query": introspection_query}
    )
    data = response.json()

    # Should be blocked in production
    assert "errors" in data

undefined

undefined

    async def mock_batch(ids):
        batch_calls.append(list(ids))
        return [{"id": id, "name": f"User {id}"} for id in ids]

    loader = DataLoader(mock_batch)

    # Load multiple users
    results = await asyncio.gather(
        loader.load("1"),
        loader.load("2"),
        loader.load("3")
    )

    # Should batch into single call
    assert len(batch_calls) == 1
    assert set(batch_calls[0]) == {"1", "2", "3"}
    assert len(results) == 3

@pytest.mark.asyncio
async def test_user_loader_caches_results(self):
    call_count = 0

    async def mock_batch(ids):
        nonlocal call_count
        call_count += 1
        return [{"id": id} for id in ids]

    loader = DataLoader(mock_batch)

    # Load same user twice
    await loader.load("1")
    await loader.load("1")

    # Should only call batch once due to caching
    assert call_count == 1

undefined

undefined

schema = build_schema(type_defs)
errors = validate_schema(schema)

assert len(errors) == 0, f"Schema errors: {errors}"

schema = build_schema(type_defs)
type_map = schema.type_map

required_types = ["User", "Post", "Query", "Mutation"]
for type_name in required_types:
    assert type_name in type_map, f"Missing type: {type_name}"

schema = build_schema(type_defs)
type_map = schema.type_map

# Verify pagination types
assert "PageInfo" in type_map
assert "PostConnection" in type_map
assert "PostEdge" in type_map

undefined

undefined

---