Web Development with Python

This module covers building web applications using Python with Flask and FastAPI frameworks.

Table of Contents

1. •Introduction to Web Development
2. •Flask Framework
3. •FastAPI Framework
4. •Templates and Static Files
5. •Databases and ORMs
6. •REST API Design
7. •Authentication and Security
8. •Deployment

---

Introduction to Web Development

Python excels at web development with its simple syntax and powerful frameworks.

Web Development Concepts

Client (Browser)                    Server (Python)
      |                                    |
      |  ------- HTTP Request -------->   |
      |                                    |
      |  <------ HTTP Response --------   |
      |                                    |

HTTP Methods

Common Status Codes

---

Flask Framework

Flask is a lightweight, flexible web framework.

Installation

pip install flask

Basic Flask Application

from flask import Flask, request, jsonify

app = Flask(__name__)

# Basic route
@app.route('/')
def home():
    return 'Hello, World!'

# Route with variable
@app.route('/user/<username>')
def show_user(username):
    return f'User: {username}'

# Route with type conversion
@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'Post ID: {post_id}'

# Multiple HTTP methods
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        return f'Logging in: {username}'
    return '''
        <form method="post">
            <input name="username">
            <button type="submit">Login</button>
        </form>
    '''

if __name__ == '__main__':
    app.run(debug=True)

Request Handling

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/data', methods=['POST'])
def handle_data():
    # Get JSON data
    json_data = request.get_json()

    # Get form data
    form_value = request.form.get('key')

    # Get query parameters
    page = request.args.get('page', default=1, type=int)

    # Get headers
    auth_header = request.headers.get('Authorization')

    # Get files
    file = request.files.get('upload')

    return jsonify({
        'json': json_data,
        'page': page
    })

Response Handling

from flask import Flask, jsonify, make_response, redirect, url_for

app = Flask(__name__)

@app.route('/api/user')
def get_user():
    # Return JSON
    return jsonify({'name': 'Alice', 'age': 30})

@app.route('/custom')
def custom_response():
    # Custom response with headers
    response = make_response('Custom Response')
    response.headers['X-Custom-Header'] = 'value'
    response.status_code = 201
    return response

@app.route('/go-home')
def go_home():
    # Redirect
    return redirect(url_for('home'))

@app.route('/error')
def error_example():
    # Error response
    return jsonify({'error': 'Not found'}), 404

Flask Blueprints (Modular Apps)

# users.py
from flask import Blueprint, jsonify

users_bp = Blueprint('users', __name__, url_prefix='/users')

@users_bp.route('/')
def list_users():
    return jsonify([{'name': 'Alice'}, {'name': 'Bob'}])

@users_bp.route('/<int:user_id>')
def get_user(user_id):
    return jsonify({'id': user_id, 'name': 'Alice'})

# app.py
from flask import Flask
from users import users_bp

app = Flask(__name__)
app.register_blueprint(users_bp)

---

FastAPI Framework

FastAPI is a modern, high-performance framework with automatic API documentation.

Installation

pip install fastapi uvicorn

Basic FastAPI Application

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# Basic route
@app.get("/")
def read_root():
    return {"message": "Hello World"}

# Path parameters
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "query": q}

# Request body with Pydantic
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

@app.post("/items/")
def create_item(item: Item):
    return {"item": item, "price_with_tax": item.price * 1.1}

# Run with: uvicorn main:app --reload

Request Validation with Pydantic

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from datetime import datetime

app = FastAPI()

class Address(BaseModel):
    street: str
    city: str
    zip_code: str = Field(..., regex=r'^\d{5}$')

class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=120)
    address: Optional[Address] = None
    tags: List[str] = []
    created_at: datetime = Field(default_factory=datetime.now)

    class Config:
        schema_extra = {
            "example": {
                "username": "johndoe",
                "email": "john@example.com",
                "age": 30,
                "tags": ["developer", "python"]
            }
        }

@app.post("/users/")
def create_user(user: User):
    return user

# Query parameter validation
@app.get("/items/")
def list_items(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    search: Optional[str] = Query(None, min_length=3)
):
    return {"skip": skip, "limit": limit, "search": search}

Async Support

from fastapi import FastAPI
import asyncio
import httpx

app = FastAPI()

# Async route
@app.get("/async-items/")
async def read_items():
    # Async database call or HTTP request
    await asyncio.sleep(0.1)
    return {"items": ["item1", "item2"]}

# Async HTTP client
async def fetch_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

@app.get("/external-data/")
async def get_external():
    data = await fetch_data("https://api.example.com/data")
    return data

Dependencies (Dependency Injection)

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional

app = FastAPI()

# Simple dependency
def get_db():
    db = DatabaseConnection()
    try:
        yield db
    finally:
        db.close()

# Dependency that uses another dependency
def get_current_user(
    db = Depends(get_db),
    token: str = Header(...)
):
    user = db.get_user_by_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

@app.get("/users/me")
def read_current_user(user = Depends(get_current_user)):
    return user

# Class-based dependency
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items/")
def list_items(pagination: Pagination = Depends()):
    return {"skip": pagination.skip, "limit": pagination.limit}

---

Templates and Static Files

Flask with Jinja2 Templates

# app.py
from flask import Flask, render_template

app = Flask(__name__)

@app.route('/hello/<name>')
def hello(name):
    return render_template('hello.html', name=name)

<!-- templates/hello.html -->
<!DOCTYPE html>
<html>
  <head>
    <title>Hello</title>
  </head>
  <body>
    <h1>Hello, {{ name }}!</h1>

    {% if name == 'admin' %}
    <p>Welcome, administrator!</p>
    {% endif %}

    <ul>
      {% for item in items %}
      <li>{{ item }}</li>
      {% endfor %}
    </ul>
  </body>
</html>

Template Inheritance

<!-- templates/base.html -->
<!DOCTYPE html>
<html>
  <head>
    <title>{% block title %}{% endblock %} - My Site</title>
    <link
      rel="stylesheet"
      href="{{ url_for('static', filename='style.css') }}"
    />
  </head>
  <body>
    <nav>
      <a href="/">Home</a>
      <a href="/about">About</a>
    </nav>

    <main>{% block content %}{% endblock %}</main>

    <footer>&copy; 2024 My Site</footer>
  </body>
</html>

<!-- templates/home.html -->
{% extends "base.html" %} {% block title %}Home{% endblock %} {% block content
%}
<h1>Welcome Home</h1>
<p>This is the home page.</p>
{% endblock %}

Static Files

# Flask static files
from flask import Flask, url_for

app = Flask(__name__)

# Static files go in 'static' folder
# Access: /static/css/style.css
# In template: {{ url_for('static', filename='css/style.css') }}

# FastAPI static files
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

---

Databases and ORMs

SQLAlchemy with Flask

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

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

# Models
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    posts = db.relationship('Post', backref='author', lazy=True)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

# CRUD operations
@app.route('/users', methods=['POST'])
def create_user():
    user = User(username='alice', email='alice@example.com')
    db.session.add(user)
    db.session.commit()
    return {'id': user.id}

@app.route('/users/<int:user_id>')
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return {'username': user.username, 'email': user.email}

@app.route('/users')
def list_users():
    users = User.query.filter(User.email.endswith('@example.com')).all()
    return {'users': [u.username for u in users]}

SQLAlchemy with FastAPI

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

# Database setup
DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Model
class UserDB(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True)

Base.metadata.create_all(bind=engine)

# Pydantic schemas
class UserCreate(BaseModel):
    username: str
    email: str

class UserResponse(BaseModel):
    id: int
    username: str
    email: str

    class Config:
        orm_mode = True

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

app = FastAPI()

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = UserDB(username=user.username, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

---

REST API Design

RESTful API Best Practices

from fastapi import FastAPI, HTTPException, Query, status
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI()

# Resource models
class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None

class Item(ItemBase):
    id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        orm_mode = True

class ItemList(BaseModel):
    items: List[Item]
    total: int
    page: int
    per_page: int

# In-memory storage
items_db = {}
item_counter = 0

# CRUD Endpoints
@app.get("/items", response_model=ItemList)
def list_items(
    page: int = Query(1, ge=1),
    per_page: int = Query(10, ge=1, le=100),
    search: Optional[str] = None
):
    """List all items with pagination."""
    all_items = list(items_db.values())

    if search:
        all_items = [i for i in all_items if search.lower() in i['name'].lower()]

    start = (page - 1) * per_page
    end = start + per_page

    return {
        "items": all_items[start:end],
        "total": len(all_items),
        "page": page,
        "per_page": per_page
    }

@app.post("/items", response_model=Item, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate):
    """Create a new item."""
    global item_counter
    item_counter += 1
    now = datetime.now()

    new_item = {
        "id": item_counter,
        **item.dict(),
        "created_at": now,
        "updated_at": now
    }
    items_db[item_counter] = new_item
    return new_item

@app.get("/items/{item_id}", response_model=Item)
def get_item(item_id: int):
    """Get a specific item."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, item: ItemCreate):
    """Full update of an item."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")

    items_db[item_id] = {
        **items_db[item_id],
        **item.dict(),
        "updated_at": datetime.now()
    }
    return items_db[item_id]

@app.patch("/items/{item_id}", response_model=Item)
def partial_update_item(item_id: int, item: ItemUpdate):
    """Partial update of an item."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")

    update_data = item.dict(exclude_unset=True)
    items_db[item_id] = {
        **items_db[item_id],
        **update_data,
        "updated_at": datetime.now()
    }
    return items_db[item_id]

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
    """Delete an item."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    del items_db[item_id]

---

Authentication and Security

JWT Authentication with FastAPI

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta
from typing import Optional

# Configuration
SECRET_KEY = "your-secret-key-here"  # Use environment variable!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

# Models
class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    email: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

# Fake database
fake_users_db = {
    "alice": {
        "username": "alice",
        "email": "alice@example.com",
        "hashed_password": pwd_context.hash("secret"),
        "disabled": False,
    }
}

# Helper functions
def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db: dict, username: str) -> Optional[UserInDB]:
    if username in db:
        return UserInDB(**db[username])
    return None

def authenticate_user(db: dict, username: str, password: str):
    user = get_user(db, username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = get_user(fake_users_db, username)
    if user is None:
        raise credentials_exception
    return user

# Endpoints
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

Flask Session Authentication

from flask import Flask, session, request, redirect, url_for
from werkzeug.security import generate_password_hash, check_password_hash
from functools import wraps

app = Flask(__name__)
app.secret_key = 'your-secret-key'  # Use environment variable!

# Login required decorator
def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user_id' not in session:
            return redirect(url_for('login'))
        return f(*args, **kwargs)
    return decorated_function

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']

        user = get_user_by_username(username)
        if user and check_password_hash(user['password'], password):
            session['user_id'] = user['id']
            return redirect(url_for('dashboard'))

        return 'Invalid credentials', 401

    return render_template('login.html')

@app.route('/logout')
def logout():
    session.pop('user_id', None)
    return redirect(url_for('home'))

@app.route('/dashboard')
@login_required
def dashboard():
    return 'Welcome to dashboard!'

---

Deployment

Running in Production

# Flask with Gunicorn
# gunicorn -w 4 -b 0.0.0.0:8000 app:app

# FastAPI with Uvicorn
# uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

Docker Deployment

# Dockerfile for FastAPI
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - '8000:8000'
    environment:
      - DATABASE_URL=postgresql://user:pass@db/app
    depends_on:
      - db

  db:
    image: postgres:15
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=app
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

---

Quick Reference

Flask vs FastAPI

Common Packages

---

Next Steps

1. •Frontend Integration: React, Vue.js, or HTMX
2. •GraphQL: Strawberry or Ariadne
3. •WebSockets: Real-time communication
4. •Caching: Redis integration
5. •Message Queues: Celery, RabbitMQ
6. •Monitoring: Prometheus, Grafana