Bulk operations in FastAPI - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ component_behavior

intermediate

Bulk Create Endpoint Behavior

Consider a FastAPI endpoint designed to create multiple user records at once using a list of user data. What will be the response if one of the user entries in the list is missing a required field?

FastAPI

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str

@app.post('/users/bulk')
async def create_users(users: list[User]):
    return {'created': len(users)}

A The endpoint creates all users except the one missing the field and returns the count of created users.

B The endpoint returns a 422 Unprocessable Entity error and no users are created.

C The endpoint ignores validation and creates all users, filling missing fields with null.

D The endpoint returns a 500 Internal Server Error due to missing data.

💡 Hint

Think about how Pydantic validates input data before the endpoint logic runs.

FastAPI uses Pydantic to validate input data. If any item in the list is missing a required field, validation fails and FastAPI returns a 422 error before entering the endpoint function.

📝 Syntax

intermediate

Correct Bulk Update Syntax

Which of the following FastAPI endpoint definitions correctly accepts a list of items for bulk update using Pydantic models?

FastAPI

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    id: int
    value: str

A

@app.put('/items/bulk')
async def update_items(items: list[Item]):
    return {'updated': len(items)}

B

@app.put('/items/bulk')
async def update_items(items: Item):
    return {'updated': 1}

C

@app.put('/items/bulk')
async def update_items(items: dict[int, Item]):
    return {'updated': len(items)}

D

@app.put('/items/bulk')
async def update_items(items: Item[]):
    return {'updated': len(items)}

💡 Hint

Remember how Python type hints express lists and arrays.

In Python, a list of Pydantic models is typed as list[Model]. Option A uses this correct syntax. Option A expects a single item, C uses a dict which is not expected, and D uses invalid syntax.

❓ state_output

advanced

Bulk Delete Endpoint Output

Given a FastAPI endpoint that deletes multiple records by their IDs, what will be the output if some IDs do not exist in the database?

FastAPI

from fastapi import FastAPI

app = FastAPI()

fake_db = {1: 'a', 2: 'b', 3: 'c'}

@app.delete('/items/bulk')
async def delete_items(ids: list[int]):
    deleted = 0
    for id_ in ids:
        if id_ in fake_db:
            del fake_db[id_]
            deleted += 1
    return {'deleted': deleted, 'requested': len(ids)}

A {'deleted': number_of_existing_ids, 'requested': total_ids_passed}

B {'deleted': total_ids_passed, 'requested': total_ids_passed}

C Raises a KeyError when a non-existing ID is deleted.

D {'deleted': 0, 'requested': total_ids_passed}

💡 Hint

Look at how the code checks if an ID exists before deleting.

The code only deletes IDs found in the fake_db and counts them. It returns how many were deleted and how many were requested. It does not raise errors for missing IDs.

🔧 Debug

advanced

Debugging Bulk Insert with Async DB

Why does this FastAPI bulk insert endpoint fail to insert all items into the database?

FastAPI

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Record(BaseModel):
    id: int
    data: str

fake_db = {}

@app.post('/records/bulk')
async def bulk_insert(records: list[Record]):
    for record in records:
        fake_db[record.id] = record.data
    return {'inserted': len(records)}

A The endpoint fails because the fake_db dictionary is not thread-safe for async operations.

B The endpoint fails because the records parameter should be a dict, not a list.

C The endpoint fails because the loop should use await when inserting each record.

D The endpoint works correctly; no failure occurs.

💡 Hint

Consider that fake_db is a simple dictionary and the code is synchronous inside the async function.

The code runs synchronously inside the async function and updates a simple dictionary. No async database calls are made, so no concurrency issues arise. The endpoint works as expected.

🧠 Conceptual

expert

Best Practice for Bulk Operations in FastAPI

Which approach is best to handle large bulk operations in FastAPI to avoid blocking the server and ensure responsiveness?

A Process all items synchronously in the request handler, returning only after completion.

B Split the bulk list into single-item requests and process them one by one synchronously.

C Use background tasks or external workers to process bulk operations asynchronously after responding.

D Reject bulk requests and force clients to send single-item requests.

💡 Hint

Think about server load and user experience when processing large data.

Handling large bulk operations synchronously can block the server and degrade performance. Using background tasks or external workers allows the server to respond quickly and process data asynchronously, improving responsiveness and scalability.

Practice

1. What is the main benefit of using bulk operations in FastAPI?

easy

A. They limit the number of items processed to one per request.

B. They automatically generate HTML forms for data input.

C. They encrypt data before sending it to the client.

D. They allow handling many items in a single request, reducing server calls.

Solution

1. Step 1: Understand bulk operations purpose

   Bulk operations let you send many items at once instead of one by one.

2. Step 2: Identify the main benefit

   This reduces the number of server calls and speeds up processing.

3. Final Answer:

   They allow handling many items in a single request, reducing server calls. -> Option D

4. Quick Check:

   Bulk operations = fewer server calls [OK]

Hint: Bulk means many items at once, not one by one [OK]

Common Mistakes:

• Thinking bulk operations encrypt data
• Confusing bulk with form generation
• Assuming bulk limits to one item

2. Which of the following is the correct way to accept multiple items in a FastAPI endpoint using Pydantic models?

easy

A. def create_items(items: ItemModel[]):

B. def create_items(item: ItemModel):

C. def create_items(items: List[ItemModel]):

D. def create_items(items: Dict[ItemModel]):

Solution

1. Step 1: Recall FastAPI typing for multiple items

   FastAPI uses Python typing List[Model] to accept multiple objects.

2. Step 2: Check each option's syntax

   def create_items(items: List[ItemModel]): uses List[ItemModel], which is correct. def create_items(item: ItemModel): accepts one item only. def create_items(items: ItemModel[]): uses invalid syntax. def create_items(items: Dict[ItemModel]): uses Dict which is not for lists.

3. Final Answer:

   def create_items(items: List[ItemModel]): -> Option C

4. Quick Check:

   Multiple items = List[Model] [OK]

Hint: Use List[Model] to accept many items in FastAPI [OK]

Common Mistakes:

• Using ItemModel[] which is invalid syntax
• Using Dict instead of List for multiple items
• Forgetting to import List from typing

3. Given this FastAPI endpoint:

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post('/items/')
async def create_items(items: List[Item]):
    return {'count': len(items), 'names': [item.name for item in items]}

What will be the response if the client sends this JSON body?

[{"name": "apple", "price": 1.2}, {"name": "banana", "price": 0.5}]

medium

A. Validation error due to missing fields

B. {"count": 2, "names": ["apple", "banana"]}

C. {"count": 1, "names": ["apple"]}

D. {"count": 0, "names": []}

Solution

1. Step 1: Understand input and endpoint behavior

   The endpoint accepts a list of Item objects and returns count and names list.

2. Step 2: Analyze input JSON

   The input has two items with names 'apple' and 'banana'.

3. Step 3: Predict output

   Count is 2, names list contains both names.

4. Final Answer:

   {"count": 2, "names": ["apple", "banana"]} -> Option B

5. Quick Check:

   List length = 2, names extracted correctly [OK]

Hint: Count list length and extract names from each item [OK]

Common Mistakes:

• Assuming only one item processed
• Expecting validation error without reason
• Returning empty list mistakenly

4. Identify the error in this FastAPI bulk operation endpoint:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    age: int

@app.post('/users/')
async def create_users(users: User):
    return {'usernames': [user.username for user in users]}

medium

A. The parameter 'users' should be a list type like List[User].

B. The return statement should not use list comprehension.

C. The User model is missing required fields.

D. The endpoint path should be '/user/' not '/users/'.

Solution

1. Step 1: Check parameter type for bulk input

   The parameter 'users' is typed as User, which accepts one object, not multiple.

2. Step 2: Identify correct typing for multiple users

   To accept many users, it should be List[User] from typing module.

3. Final Answer:

   The parameter 'users' should be a list type like List[User]. -> Option A

4. Quick Check:

   Bulk input needs List[Model] type [OK]

Hint: Bulk input must be List[Model], not single Model [OK]

Common Mistakes:

• Using single model instead of list for bulk
• Thinking list comprehension causes error
• Confusing endpoint path naming

5. You want to create a FastAPI endpoint that accepts a bulk list of products but only saves those with price greater than 10. Which code snippet correctly implements this filtering inside the endpoint?

hard

A. async def create_products(products: List[Product]):
filtered = [p for p in products if p.price > 10]
return {'saved': len(filtered)}

B. async def create_products(products: List[Product]):
filtered = [p for p in products if p.price < 10]
return {'saved': len(filtered)}

C. async def create_products(products: Product):
filtered = [p for p in products if p.price > 10]
return {'saved': len(filtered)}

D. async def create_products(products: List[Product]):
filtered = filter(lambda p: p.price < 10, products)
return {'saved': len(filtered)}

Solution

1. Step 1: Confirm parameter type for bulk products

   Parameter must be List[Product] to accept multiple products.

2. Step 2: Check filtering condition

   We want products with price greater than 10, so filter with p.price > 10.

3. Step 3: Validate return value

   Return count of filtered list using len(filtered).

4. Final Answer:

   async def create_products(products: List[Product]): filtered = [p for p in products if p.price > 10] return {'saved': len(filtered)} -> Option A

5. Quick Check:

   Filter price > 10 and count saved [OK]

Hint: Filter with list comprehension and correct condition [OK]

Common Mistakes:

• Filtering with wrong condition (price < 10)
• Using single Product instead of List[Product]
• Using filter but forgetting to convert to list before len()