Loading…
We have reached the summit. These final five questions are less about "how to write code" and more about "how to architect systems." These are the types of problems you will face in a Senior Developer interview or on your first week as a Lead.
We will tackle implementing a robust permission system, handling file uploads securely (not on the server!), breaking a monolith into microservices-ready modules, handling high-traffic real-time features, and finally, the art of debugging a live production server.
We touched on this in Q39, but let's build a production-grade dependency. The goal is to have a reusable way to protect endpoints based on user roles (e.g., Admin, Manager, User).
from fastapi import FastAPI, Depends, HTTPException, status
from typing import List
app = FastAPI()
# 1. Mock Database & User
fake_users_db = {
"alice": {"username": "alice", "role": "admin"},
"bob": {"username": "bob", "role": "user"},
}
# 2. Authentication Dependency (Simplified)
def get_current_user(token: str):
# In reality, decode JWT here
user = fake_users_db.get(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid Token")
return user
# 3. The RBAC Dependency Factory
class RoleChecker:
def __init__(self, allowed_roles: List[str]):
self.allowed_roles = allowed_roles
def __call__(self, user: Annotated[dict, Depends(get_current_user)]):
if user["role"] not in self.allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Operation not permitted"
)
return user
# 4. Usage
allow_admin = RoleChecker(["admin"])
allow_user = RoleChecker(["admin", "user"])
@app.delete("/users/{id}", dependencies=[Depends(allow_admin)])
def delete_user(id: int):
return {"msg": "User deleted"}
@app.get("/items/", dependencies=[Depends(allow_user)])
def read_items():
return {"msg": "Here are items"}
Sample Output:
Request DELETE /users/1 with token "bob" (role: user):
{
"detail": "Operation not permitted"
}Storing files on the server disk is bad practice (server crashes = files lost). We need to validate the file (ensure it's an image) and upload it to AWS S3 using boto3.
from fastapi import FastAPI, UploadFile, File, HTTPException
import boto3
from botocore.exceptions import NoCredentialsError
app = FastAPI()
s3 = boto3.client("s3", aws_access_key_id="KEY", aws_secret_access_key="SECRET")
BUCKET_NAME = "my-bucket"
@app.post("/upload/")
async def upload_image(file: UploadFile = File(...)):
# 1. Validation
if file.content_type not in ["image/jpeg", "image/png"]:
raise HTTPException(400, detail="Only JPEG/PNG images allowed")
try:
# 2. Upload to S3
# We use file.file (the actual Python file object)
s3.upload_fileobj(file.file, BUCKET_NAME, file.filename)
# 3. Generate URL
url = f"https://{BUCKET_NAME}.s3.amazonaws.com/{file.filename}"
return {"url": url}
except NoCredentialsError:
raise HTTPException(500, detail="AWS Credentials missing")
except Exception as e:
raise HTTPException(500, detail=str(e))
You have a main.py with 5,000 lines of code. It's a nightmare. The interviewer wants to see how you refactor it using APIRouter.
The Strategy:
1. Create a folder routers/.
2. Move endpoints into routers/users.py, routers/items.py.
3. Define an APIRouter in each file.
4. Import and mount them in main.py.
# --- routers/users.py ---
from fastapi import APIRouter
router = APIRouter(
prefix="/users",
tags=["users"],
responses={404: {"description": "Not found"}},
)
@router.get("/")
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
# --- main.py ---
from fastapi import FastAPI
from routers import users # Import the module
app = FastAPI()
# Connect the router
app.include_router(users.router)
@app.get("/")
async def root():
return {"message": "Hello Application!"}
Note: The prefix /users means the endpoint is accessible at http://localhost:8000/users/.
We need to send notifications to thousands of connected users. A simple list of WebSockets works for a single server, but managing connections efficiently is key. This is a "Connection Manager" pattern.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
app = FastAPI()
# 1. Connection Manager Class
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left")
Your manager says "The server is returning 500 errors." What do you do? This tests your operational knowledge.
traceback.Common Fix (Middleware):Often, you want to see the error ID in the response so you can find it in logs easily.
import uuid
import logging
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
app = FastAPI()
logger = logging.getLogger("app")
@app.middleware("http")
async def catch_exceptions_middleware(request: Request, call_next):
try:
return await call_next(request)
except Exception as e:
# Generate a unique ID for this error
error_id = uuid.uuid4()
logger.error(f"Error ID: {error_id} - Message: {str(e)}", exc_info=True)
# Return a safe message to the user with the ID
return JSONResponse(
status_code=500,
content={"detail": "Internal Server Error", "error_id": str(error_id)}
)
@app.get("/crash")
def crash():
raise ValueError("Something went wrong!")
Sample Output (User sees):
{
"detail": "Internal Server Error",
"error_id": "a1b2c3d4-..."
}The developer can then grep the logs for a1b2c3d4-... to find the exact traceback.
3 questions · no sign-up, nothing stored
Progress is stored in this browser only - no sign-up, nothing sent anywhere.