REST APIs in Python

1. Resources and URLs
   - Everything is a resource (user, post, product)
   - Each resource has a unique URL

   GET  /api/users          → List all users
   GET  /api/users/1        → Get user with ID 1
   POST /api/users          → Create new user
   PUT  /api/users/1        → Update user 1
   DELETE /api/users/1      → Delete user 1

2. HTTP Methods (Verbs)
   GET     → Retrieve data (safe, idempotent)
   POST    → Create new resource
   PUT     → Update entire resource (idempotent)
   PATCH   → Update partial resource
   DELETE  → Remove resource (idempotent)

3. Status Codes
   200 OK              → Success
   201 Created         → Resource created
   204 No Content      → Success, no response body
   400 Bad Request     → Invalid input
   401 Unauthorized    → Authentication required
   403 Forbidden       → Authenticated but no permission
   404 Not Found       → Resource doesn't exist
   500 Server Error    → Something broke on server

4. Stateless
   - Each request contains all needed information
   - Server doesn't store client state
   - Use tokens (JWT) for authentication

5. JSON Format (standard)
   {
     "id": 1,
     "username": "alice",
     "email": "alice@example.com"
   }

# Install Django REST Framework
pip install djangorestframework

# settings.py
INSTALLED_APPS = [
    'rest_framework',
]

REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 10
}

# models.py
from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.CharField(max_length=100)
    isbn = models.CharField(max_length=13, unique=True)
    published_date = models.DateField()
    price = models.DecimalField(max_digits=6, decimal_places=2)

    def __str__(self):
        return self.title

# serializers.py (convert models to/from JSON)
from rest_framework import serializers
from .models import Book

class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = ['id', 'title', 'author', 'isbn', 'published_date', 'price']

    # Custom validation
    def validate_price(self, value):
        if value <= 0:
            raise serializers.ValidationError("Price must be positive")
        return value

# views.py (using ViewSets - most powerful)
from rest_framework import viewsets
from .models import Book
from .serializers import BookSerializer

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer

    # Custom action
    @action(detail=False, methods=['get'])
    def recent(self, request):
        recent_books = Book.objects.order_by('-published_date')[:5]
        serializer = self.get_serializer(recent_books, many=True)
        return Response(serializer.data)

# urls.py
from rest_framework.routers import DefaultRouter
from .views import BookViewSet

router = DefaultRouter()
router.register(r'books', BookViewSet)

urlpatterns = [
    path('api/', include(router.urls)),
]

# This creates these endpoints automatically:
# GET    /api/books/          → List books
# POST   /api/books/          → Create book
# GET    /api/books/1/        → Get book 1
# PUT    /api/books/1/        → Update book 1
# PATCH  /api/books/1/        → Partial update
# DELETE /api/books/1/        → Delete book 1
# GET    /api/books/recent/   → Custom action

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.TokenAuthentication',
        'rest_framework.authentication.SessionAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ]
}

# Generate token for users
from rest_framework.authtoken.models import Token
token = Token.objects.create(user=user)
print(token.key)

# Client sends token in header:
# Authorization: Token 9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b

# views.py - Custom permissions
from rest_framework import permissions

class IsOwnerOrReadOnly(permissions.BasePermission):
    def has_object_permission(self, request, view, obj):
        # Read permissions for everyone
        if request.method in permissions.SAFE_METHODS:
            return True
        # Write permissions only for owner
        return obj.owner == request.user

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    permission_classes = [IsOwnerOrReadOnly]

# JWT Authentication (more modern)
pip install djangorestframework-simplejwt

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework_simplejwt.authentication.JWTAuthentication',
    ],
}

# urls.py
from rest_framework_simplejwt.views import (
    TokenObtainPairView,
    TokenRefreshView,
)

urlpatterns = [
    path('api/token/', TokenObtainPairView.as_view()),
    path('api/token/refresh/', TokenRefreshView.as_view()),
]

# Client workflow:
# 1. POST /api/token/ with username/password
#    Returns: access token + refresh token
# 2. Use access token in requests:
#    Authorization: Bearer 
# 3. When access expires, POST to /api/token/refresh/
#    with refresh token to get new access token

# Install
pip install flask-restful flask-sqlalchemy

# app.py
from flask import Flask
from flask_restful import Resource, Api, reqparse, fields, marshal_with
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///books.db'
db = SQLAlchemy(app)
api = Api(app)

# Model
class Book(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    author = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)

    def __repr__(self):
        return f'<Book {self.title}>'

# Serialization format
book_fields = {
    'id': fields.Integer,
    'title': fields.String,
    'author': fields.String,
    'price': fields.Float
}

# Request parser (validation)
book_parser = reqparse.RequestParser()
book_parser.add_argument('title', type=str, required=True, help='Title is required')
book_parser.add_argument('author', type=str, required=True)
book_parser.add_argument('price', type=float, required=True)

# Resources (endpoints)
class BookList(Resource):
    @marshal_with(book_fields)
    def get(self):
        return Book.query.all()

    @marshal_with(book_fields)
    def post(self):
        args = book_parser.parse_args()
        book = Book(
            title=args['title'],
            author=args['author'],
            price=args['price']
        )
        db.session.add(book)
        db.session.commit()
        return book, 201

class BookResource(Resource):
    @marshal_with(book_fields)
    def get(self, book_id):
        return Book.query.get_or_404(book_id)

    @marshal_with(book_fields)
    def put(self, book_id):
        book = Book.query.get_or_404(book_id)
        args = book_parser.parse_args()
        book.title = args['title']
        book.author = args['author']
        book.price = args['price']
        db.session.commit()
        return book

    def delete(self, book_id):
        book = Book.query.get_or_404(book_id)
        db.session.delete(book)
        db.session.commit()
        return '', 204

# Register routes
api.add_resource(BookList, '/api/books')
api.add_resource(BookResource, '/api/books/')

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

# Install
pip install fastapi uvicorn sqlalchemy pydantic

# main.py
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker, Session, declarative_base
from pydantic import BaseModel, validator

# Database setup
DATABASE_URL = "sqlite:///./books.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

# SQLAlchemy Model
class BookDB(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    author = Column(String, nullable=False)
    price = Column(Float, nullable=False)

Base.metadata.create_all(bind=engine)

# Pydantic models (for validation and serialization)
class BookBase(BaseModel):
    title: str
    author: str
    price: float

    @validator('price')
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('Price must be positive')
        return v

class BookCreate(BookBase):
    pass

class Book(BookBase):
    id: int

    class Config:
        from_attributes = True

# FastAPI app
app = FastAPI(title="Book API", version="1.0.0")

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Endpoints
@app.get("/api/books", response_model=list[Book])
def get_books(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    """Get all books with pagination"""
    books = db.query(BookDB).offset(skip).limit(limit).all()
    return books

@app.get("/api/books/{book_id}", response_model=Book)
def get_book(book_id: int, db: Session = Depends(get_db)):
    """Get a specific book by ID"""
    book = db.query(BookDB).filter(BookDB.id == book_id).first()
    if not book:
        raise HTTPException(status_code=404, detail="Book not found")
    return book

@app.post("/api/books", response_model=Book, status_code=201)
def create_book(book: BookCreate, db: Session = Depends(get_db)):
    """Create a new book"""
    db_book = BookDB(**book.dict())
    db.add(db_book)
    db.commit()
    db.refresh(db_book)
    return db_book

@app.put("/api/books/{book_id}", response_model=Book)
def update_book(book_id: int, book: BookCreate, db: Session = Depends(get_db)):
    """Update a book"""
    db_book = db.query(BookDB).filter(BookDB.id == book_id).first()
    if not db_book:
        raise HTTPException(status_code=404, detail="Book not found")

    for key, value in book.dict().items():
        setattr(db_book, key, value)

    db.commit()
    db.refresh(db_book)
    return db_book

@app.delete("/api/books/{book_id}", status_code=204)
def delete_book(book_id: int, db: Session = Depends(get_db)):
    """Delete a book"""
    db_book = db.query(BookDB).filter(BookDB.id == book_id).first()
    if not db_book:
        raise HTTPException(status_code=404, detail="Book not found")
    db.delete(db_book)
    db.commit()
    return None

# Run with: uvicorn main:app --reload
# Visit http://127.0.0.1:8000/docs for auto-generated docs!
# Visit http://127.0.0.1:8000/redoc for alternative docs

# Authentication with JWT
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from datetime import datetime, timedelta
import jwt

security = HTTPBearer()
SECRET_KEY = "your-secret-key"

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(hours=1)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/api/login")
def login(username: str, password: str):
    # Validate credentials (simplified)
    if username == "admin" and password == "secret":
        token = create_access_token({"sub": username})
        return {"access_token": token, "token_type": "bearer"}
    raise HTTPException(status_code=401, detail="Invalid credentials")

@app.get("/api/protected")
def protected_route(user = Depends(verify_token)):
    return {"message": f"Hello {user['sub']}"}

# Background tasks
from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # Simulate sending email
    print(f"Sending email to {email}: {message}")

@app.post("/api/send-notification")
def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Your book order confirmed!")
    return {"message": "Notification will be sent"}

# File uploads
from fastapi import File, UploadFile

@app.post("/api/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # Process file
    return {"filename": file.filename, "size": len(contents)}

# CORS (for frontend integration)
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # React app
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Using pytest with FastAPI
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_get_books():
    response = client.get("/api/books")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

def test_create_book():
    response = client.post("/api/books", json={
        "title": "Test Book",
        "author": "Test Author",
        "price": 19.99
    })
    assert response.status_code == 201
    assert response.json()["title"] == "Test Book"

# Using pytest with DRF
from rest_framework.test import APITestCase

class BookAPITest(APITestCase):
    def test_get_books(self):
        response = self.client.get('/api/books/')
        self.assertEqual(response.status_code, 200)

    def test_create_book(self):
        data = {'title': 'Test Book', 'author': 'Author', 'price': 19.99}
        response = self.client.post('/api/books/', data)
        self.assertEqual(response.status_code, 201)

# Manual testing with curl
curl http://localhost:8000/api/books

curl -X POST http://localhost:8000/api/books \
  -H "Content-Type: application/json" \
  -d '{"title": "New Book", "author": "Alice", "price": 29.99}'

# Or use Postman, Insomnia, or HTTPie for easier testing