Code Standards

  # app/api/v1/endpoints/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.api import deps
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService
from app.core.logging import get_logger

logger = get_logger(__name__)
router = APIRouter()


@router.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(
    user_in: UserCreate,
    db: Session = Depends(deps.get_db),
    current_user: User = Depends(deps.get_current_user),
) -> UserResponse:
    """
    Create a new user.

    This endpoint creates a new user in the system with the provided details.

    Args:
        user_in: User creation data containing email, full_name, and password
        db: Database session (injected)
        current_user: Currently authenticated user (injected)

    Returns:
        UserResponse: The newly created user object

    Raises:
        HTTPException 403: If current user is not an admin
        HTTPException 400: If email already exists

    Example:
        ```python
        POST /api/v1/users
        {
            "email": "john@example.com",
            "full_name": "John Doe",
            "password": "securepassword123"
        }
        ```
    """
    logger.info(f"Creating user with email: {user_in.email}")

    # Authorisation check
    if not current_user.is_admin:
        logger.warning(f"Non-admin user {current_user.id} attempted to create user")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only administrators can create users"
        )

    # Call service layer
    try:
        user = UserService.create_user(db, user_in)
        logger.info(f"User created successfully: {user.id}")
        return user
    except ValueError as e:
        # Service layer raises ValueError for business logic errors
        logger.error(f"Failed to create user: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e)
        )
  

  # app/services/user_service.py
from typing import Optional
from sqlalchemy.orm import Session

from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
from app.core.security import get_password_hash
from app.core.logging import get_logger

logger = get_logger(__name__)


class UserService:
    """Service for managing user-related business logic."""

    @staticmethod
    def create_user(db: Session, user_in: UserCreate) -> User:
        """
        Create a new user.

        Args:
            db: Database session
            user_in: User creation schema

        Returns:
            User: The newly created user

        Raises:
            ValueError: If email already exists
        """
        # Check if user already exists
        existing_user = db.query(User).filter(User.email == user_in.email).first()
        if existing_user:
            logger.warning(f"Attempted to create duplicate user: {user_in.email}")
            raise ValueError(f"User with email {user_in.email} already exists")

        # Create user
        user = User(
            email=user_in.email,
            full_name=user_in.full_name,
            hashed_password=get_password_hash(user_in.password),
            is_active=True,
            is_admin=False,
        )

        db.add(user)
        db.commit()
        db.refresh(user)

        logger.info(f"User created with ID: {user.id}")
        return user

    @staticmethod
    def get_user_by_email(db: Session, email: str) -> Optional[User]:
        """
        Retrieve a user by email.

        Args:
            db: Database session
            email: User's email address

        Returns:
            Optional[User]: User if found, None otherwise
        """
        return db.query(User).filter(User.email == email).first()

    @staticmethod
    def update_user(db: Session, user_id: int, user_in: UserUpdate) -> User:
        """
        Update user details.

        Args:
            db: Database session
            user_id: ID of user to update
            user_in: Update data

        Returns:
            User: Updated user object

        Raises:
            ValueError: If user not found
        """
        user = db.query(User).filter(User.id == user_id).first()
        if not user:
            raise ValueError(f"User {user_id} not found")

        # Update only provided fields
        update_data = user_in.dict(exclude_unset=True)
        for field, value in update_data.items():
            setattr(user, field, value)

        db.commit()
        db.refresh(user)

        logger.info(f"User {user_id} updated")
        return user
  

  # app/api/deps.py
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from sqlalchemy.orm import Session

from app.db.session import SessionLocal
from app.core.config import settings
from app.models.user import User
from app.core.logging import get_logger

logger = get_logger(__name__)
security = HTTPBearer()


def get_db() -> Generator:
    """
    Get database session.

    Yields:
        Session: SQLAlchemy database session
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def get_current_user(
    db: Session = Depends(get_db),
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> User:
    """
    Get current authenticated user from JWT token.

    Args:
        db: Database session
        credentials: HTTP Bearer token credentials

    Returns:
        User: Currently authenticated user

    Raises:
        HTTPException 401: If token is invalid or user not found
    """
    token = credentials.credentials

    try:
        payload = jwt.decode(
            token,
            settings.JWT_SECRET_KEY,
            algorithms=[settings.JWT_ALGORITHM]
        )
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials"
            )
    except JWTError:
        logger.warning("Invalid JWT token provided")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials"
        )

    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        logger.warning(f"User {user_id} from token not found in database")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found"
        )

    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Inactive user"
        )

    return user
  

  # Example combining all error types
from fastapi import HTTPException, status

# 401 - Authentication failed
raise HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Invalid credentials"
)

# 403 - Authorisation failed
raise HTTPException(
    status_code=status.HTTP_403_FORBIDDEN,
    detail="Only admins can perform this action"
)

# 422 - Validation error (automatic with Pydantic)
# No code needed - FastAPI handles this

# 400 - Business logic error
raise HTTPException(
    status_code=status.HTTP_400_BAD_REQUEST,
    detail="Email already exists"
)
  

  # app/core/logging.py
import logging
import sys
from typing import Any
import json

class JSONFormatter(logging.Formatter):
    """Custom JSON formatter for structured logging."""

    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }

        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data)


def get_logger(name: str) -> logging.Logger:
    """
    Get a configured logger instance.

    Args:
        name: Logger name (usually __name__)

    Returns:
        logging.Logger: Configured logger
    """
    logger = logging.getLogger(name)

    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)

    return logger


# Usage in code
logger = get_logger(__name__)

logger.info("User created", extra={"user_id": user.id, "email": user.email})
logger.error("Failed to process order", extra={"order_id": order_id, "error": str(e)})
  

  # app/schemas/user.py
from typing import Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field


class UserBase(BaseModel):
    """Base user schema with common fields."""

    email: EmailStr = Field(..., description="User's email address")
    full_name: str = Field(..., min_length=1, max_length=100, description="User's full name")


class UserCreate(UserBase):
    """Schema for creating a new user."""

    password: str = Field(..., min_length=8, description="User's password (min 8 characters)")


class UserUpdate(BaseModel):
    """Schema for updating user details."""

    full_name: Optional[str] = Field(None, min_length=1, max_length=100)
    email: Optional[EmailStr] = None


class UserResponse(UserBase):
    """Schema for user response (excludes sensitive data)."""

    id: int
    is_active: bool
    is_admin: bool
    created_at: datetime
    updated_at: Optional[datetime]

    class Config:
        from_attributes = True  # Allows loading from ORM models
  

  def process_shopify_order(order_id: str, db: Session) -> Order:
    """
    Process a Shopify order and update inventory.

    This function fetches order details from Shopify, validates the products,
    updates inventory levels, and creates a fulfillment record.

    Args:
        order_id: The Shopify order ID
        db: Database session for database operations

    Returns:
        Order: The processed order object

    Raises:
        ValueError: If order_id is invalid or order not found
        HTTPException: If Shopify API call fails

    Example:
        ```python
        order = process_shopify_order("12345", db)
        print(f"Order {order.id} processed successfully")
        ```

    Note:
        This function makes external API calls and may take 2-5 seconds.
        It should be called from a background worker, not directly in API endpoints.
    """
    # Implementation here
    pass
  

  // src/components/UserProfile.tsx
import React, { useState, useEffect } from 'react';
import { useQuery, useMutation } from '@tanstack/react-query';
import { userApi } from '../api/users';

/**
 * User profile component displaying and editing user information.
 *
 * @component
 * @example
 * ```tsx
 * <UserProfile userId={123} />
 * ```
 */
interface UserProfileProps {
  /** The ID of the user to display */
  userId: number;
  /** Callback when user is updated */
  onUpdate?: (user: User) => void;
}

interface User {
  id: number;
  email: string;
  full_name: string;
  is_active: boolean;
  created_at: string;
}

export const UserProfile: React.FC<UserProfileProps> = ({ userId, onUpdate }) => {
  const [isEditing, setIsEditing] = useState(false);
  const [formData, setFormData] = useState({ full_name: '' });

  // Fetch user data
  const {
    data: user,
    isLoading,
    error,
    refetch
  } = useQuery({
    queryKey: ['user', userId],
    queryFn: () => userApi.getUser(userId),
  });

  // Update user mutation
  const updateMutation = useMutation({
    mutationFn: (data: { full_name: string }) => userApi.updateUser(userId, data),
    onSuccess: (updatedUser) => {
      setIsEditing(false);
      refetch();
      onUpdate?.(updatedUser);
    },
  });

  // Initialize form when user data loads
  useEffect(() => {
    if (user) {
      setFormData({ full_name: user.full_name });
    }
  }, [user]);

  // Event handlers
  const handleEdit = () => {
    setIsEditing(true);
  };

  const handleCancel = () => {
    setIsEditing(false);
    if (user) {
      setFormData({ full_name: user.full_name });
    }
  };

  const handleSubmit = (e: React.FormEvent) => {
    e.preventDefault();
    updateMutation.mutate(formData);
  };

  const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>) => {
    setFormData({ ...formData, [e.target.name]: e.target.value });
  };

  // Loading state
  if (isLoading) {
    return (
      <div className="flex items-center justify-center p-8">
        <div className="animate-spin rounded-full h-8 w-8 border-b-2 border-blue-600"></div>
      </div>
    );
  }

  // Error state
  if (error) {
    return (
      <div className="bg-red-50 border border-red-200 rounded p-4">
        <p className="text-red-800">Failed to load user: {error.message}</p>
      </div>
    );
  }

  // No data state
  if (!user) {
    return <div>User not found</div>;
  }

  return (
    <div className="bg-white shadow rounded-lg p-6">
      <h2 className="text-2xl font-bold mb-4">User Profile</h2>

      {!isEditing ? (
        // View mode
        <div className="space-y-3">
          <div>
            <label className="text-sm font-medium text-gray-600">Name</label>
            <p className="text-gray-900">{user.full_name}</p>
          </div>
          <div>
            <label className="text-sm font-medium text-gray-600">Email</label>
            <p className="text-gray-900">{user.email}</p>
          </div>
          <div>
            <label className="text-sm font-medium text-gray-600">Status</label>
            <p className="text-gray-900">{user.is_active ? 'Active' : 'Inactive'}</p>
          </div>
          <button
            onClick={handleEdit}
            className="mt-4 bg-blue-600 text-white px-4 py-2 rounded hover:bg-blue-700"
          >
            Edit Profile
          </button>
        </div>
      ) : (
        // Edit mode
        <form onSubmit={handleSubmit} className="space-y-4">
          <div>
            <label htmlFor="full_name" className="block text-sm font-medium text-gray-700">
              Full Name
            </label>
            <input
              type="text"
              id="full_name"
              name="full_name"
              value={formData.full_name}
              onChange={handleInputChange}
              className="mt-1 block w-full rounded border-gray-300 shadow-sm focus:border-blue-500 focus:ring-blue-500"
              required
            />
          </div>

          <div className="flex gap-2">
            <button
              type="submit"
              disabled={updateMutation.isPending}
              className="bg-blue-600 text-white px-4 py-2 rounded hover:bg-blue-700 disabled:opacity-50"
            >
              {updateMutation.isPending ? 'Saving...' : 'Save Changes'}
            </button>
            <button
              type="button"
              onClick={handleCancel}
              className="bg-gray-300 text-gray-700 px-4 py-2 rounded hover:bg-gray-400"
            >
              Cancel
            </button>
          </div>

          {updateMutation.isError && (
            <div className="bg-red-50 border border-red-200 rounded p-3">
              <p className="text-red-800 text-sm">
                Failed to update: {updateMutation.error.message}
              </p>
            </div>
          )}
        </form>
      )}
    </div>
  );
};
  

  /**
 * Props for the OrderList component.
 */
interface OrderListProps {
  /** Array of orders to display */
  orders: Order[];
  /** Callback when an order is selected */
  onOrderSelect: (orderId: number) => void;
  /** Optional CSS class name */
  className?: string;
  /** Whether to show the action buttons */
  showActions?: boolean;
}
  

  // Component state
const [searchQuery, setSearchQuery] = useState('');
const [selectedIds, setSelectedIds] = useState<number[]>([]);

// Server state with React Query
const { data, isLoading, error } = useQuery({
  queryKey: ['orders', { status: 'pending', search: searchQuery }],
  queryFn: () => orderApi.fetchOrders({ status: 'pending', search: searchQuery }),
});
  

  # Standard library imports
import json
import logging
from datetime import datetime
from typing import Optional, List

# Third-party imports
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel

# Local application imports
from app.core.config import settings
from app.models.user import User
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService
  

  // React imports
import React, { useState, useEffect } from 'react';

// Third-party library imports
import { useQuery } from '@tanstack/react-query';
import axios from 'axios';

// Local component imports
import { Button } from '../components/Button';
import { Modal } from '../components/Modal';

// Local utility/API imports
import { userApi } from '../api/users';
import { formatDate } from '../utils/date';

// Type imports
import type { User, Order } from '../types';
  

  # GOOD - Explains why
# We cache this for 1 hour because the Shopify API rate limit is 2 calls/second
cache.set(key, value, timeout=3600)

# BAD - Explains what (code already shows this)
# Set cache value
cache.set(key, value, timeout=3600)