第13章 Web開發¶

前言¶

本章將帶領讀者深入學習Python Web開發的各個方面，從基礎的Web開發概念到主流框架的實際應用。我們將學習Flask輕量級框架、Django全棧框架和FastAPI現代框架，掌握數據庫操作、前端集成、API設計等核心技能。通過大量的實際代碼示例和項目實踐，讀者將能夠獨立開發完整的Web應用程序。

13.1 Web開發基礎¶

Web應用架構概述¶

Web應用程序採用客戶端-服務器架構模式，客戶端（通常是瀏覽器）通過HTTP協議向服務器發送請求，服務器處理請求後返回響應。現代Web應用通常分爲三層架構：

表示層（Presentation Layer）：負責用戶界面和用戶交互
業務邏輯層（Business Logic Layer）：處理業務規則和數據處理
數據訪問層（Data Access Layer）：負責數據存儲和檢索

HTTP協議基礎¶

HTTP（HyperText Transfer Protocol）是Web通信的基礎協議。讓我們通過一個簡單的Python腳本來理解HTTP請求和響應：

# http_demo.py
import requests

# 發送GET請求
response = requests.get('https://httpbin.org/get')
print(f"狀態碼: {response.status_code}")
print(f"響應頭: {response.headers}")
print(f"響應內容: {response.json()}")

# 發送POST請求
data = {'name': 'Python學習者', 'skill': 'Web開發'}
response = requests.post('https://httpbin.org/post', json=data)
print(f"POST響應: {response.json()}")

運行上述代碼，輸出類似如下：

狀態碼: 200
響應頭: {'Date': 'Mon, 15 Jan 2024 10:30:00 GMT', 'Content-Type': 'application/json', 'Content-Length': '312'}
響應內容: {'args': {}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'httpbin.org', 'User-Agent': 'python-requests/2.28.1'}, 'origin': '192.168.1.100', 'url': 'https://httpbin.org/get'}
POST響應: {'args': {}, 'data': '{"name": "Python學習者", "skill": "Web開發"}', 'files': {}, 'form': {}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '52', 'Content-Type': 'application/json', 'Host': 'httpbin.org', 'User-Agent': 'python-requests/2.28.1'}, 'json': {'name': 'Python學習者', 'skill': 'Web開發'}, 'origin': '192.168.1.100', 'url': 'https://httpbin.org/post'}

Python Web開發生態¶

Python擁有豐富的Web開發生態系統，主要框架對比如下：

框架	特點	適用場景	學習難度
Flask	輕量級、靈活	小型項目、API開發	低
Django	全棧、功能完整	大型項目、快速開發	中
FastAPI	現代、高性能	API開發、微服務	中
Tornado	異步、高併發	即時應用	高

13.2 Flask輕量級框架¶

Flask安裝和環境配置¶

首先創建虛擬環境並安裝Flask：

# 創建虛擬環境
python -m venv flask_env

# 激活虛擬環境（Windows）
flask_env\Scripts\activate

# 安裝Flask
pip install Flask

安裝成功後，輸出類似如下：

Collecting Flask
  Downloading Flask-2.3.3-py3-none-any.whl (96 kB)
Collecting Werkzeug>=2.3.7
  Downloading Werkzeug-2.3.7-py3-none-any.whl (242 kB)
Collecting Jinja2>=3.1.2
  Downloading Jinja2-3.1.2-py3-none-any.whl (133 kB)
Collecting itsdangerous>=2.1.2
  Downloading itsdangerous-2.1.2-py3-none-any.whl (15 kB)
Collecting click>=8.1.3
  Downloading click-8.1.7-py3-none-any.whl (97 kB)
Installing collected packages: Werkzeug, Jinja2, itsdangerous, click, Flask
Successfully installed Flask-2.3.3 Jinja2-3.1.2 Werkzeug-2.3.7 click-8.1.7 itsdangerous-2.1.2

第一個Flask應用¶

創建一個簡單的Flask應用：

# app.py
from flask import Flask, render_template, request, jsonify

app = Flask(__name__)

@app.route('/api/users', methods=['GET', 'POST'])
def users():
    if request.method == 'POST':
        try:
            user = user_schema.load(request.json)
            db.session.add(user)
            db.session.commit()
            return user_schema.jsonify(user), 201
        except ValidationError as err:
            return jsonify({'errors': err.messages}), 400

    # GET請求
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)

    users = User.query.paginate(
        page=page, per_page=per_page, error_out=False
    )

    return jsonify({
        'users': users_schema.dump(users.items),
        'pagination': {
            'page': users.page,
            'pages': users.pages,
            'per_page': users.per_page,
            'total': users.total
        }
    })

@app.route('/api/users/<int:user_id>', methods=['GET', 'PUT', 'DELETE'])
def user_detail(user_id):
    user = User.query.get_or_404(user_id)

    if request.method == 'GET':
        return user_schema.jsonify(user)

    elif request.method == 'PUT':
        try:
            user = user_schema.load(request.json, instance=user, partial=True)
            db.session.commit()
            return user_schema.jsonify(user)
        except ValidationError as err:
            return jsonify({'errors': err.messages}), 400

    elif request.method == 'DELETE':
        db.session.delete(user)
        db.session.commit()
        return '', 204

@app.route('/api/posts', methods=['GET', 'POST'])
def posts():
    if request.method == 'POST':
        try:
            post = post_schema.load(request.json)
            db.session.add(post)
            db.session.commit()
            return post_schema.jsonify(post), 201
        except ValidationError as err:
            return jsonify({'errors': err.messages}), 400

    # GET請求
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    category = request.args.get('category')

    query = Post.query
    if category:
        query = query.filter(Post.category == category)

    posts = query.paginate(
        page=page, per_page=per_page, error_out=False
    )

    return jsonify({
        'posts': posts_schema.dump(posts.items),
        'pagination': {
            'page': posts.page,
            'pages': posts.pages,
            'per_page': posts.per_page,
            'total': posts.total
        }
    })

@app.route('/api/posts/<int:post_id>', methods=['GET', 'PUT', 'DELETE'])
def post_detail(post_id):
    post = Post.query.get_or_404(post_id)

    if request.method == 'GET':
        return post_schema.jsonify(post)

    elif request.method == 'PUT':
        try:
            post = post_schema.load(request.json, instance=post, partial=True)
            post.updated_at = datetime.utcnow()
            db.session.commit()
            return post_schema.jsonify(post)
        except ValidationError as err:
            return jsonify({'errors': err.messages}), 400

    elif request.method == 'DELETE':
        db.session.delete(post)
        db.session.commit()
        return '', 204

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

運行RESTful API服務：

python restful_api.py

輸出類似如下：

 * Running on http://127.0.0.1:5000
 * Debug mode: on
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 345-678-901

API測試示例¶

使用curl命令測試API：

# 創建用戶
curl -X POST http://localhost:5000/api/users \
  -H "Content-Type: application/json" \
  -d '{"username": "testuser", "email": "test@example.com"}'

# 獲取用戶列表
curl http://localhost:5000/api/users

# 創建文章
curl -X POST http://localhost:5000/api/posts \
  -H "Content-Type: application/json" \
  -d '{"title": "測試文章", "content": "這是一篇測試文章", "user_id": 1, "category": "測試"}'

# 獲取文章列表
curl http://localhost:5000/api/posts

13.8 部署和運維¶

生產環境部署¶

使用Gunicorn部署Flask應用¶

安裝Gunicorn：

pip install gunicorn

創建WSGI入口文件：

# wsgi.py
from app import app

if __name__ == "__main__":
    app.run()

使用Gunicorn啓動應用：

gunicorn --bind 0.0.0.0:8000 --workers 4 wsgi:app

輸出類似如下：

[2024-01-15 15:30:00 +0000] [12345] [INFO] Starting gunicorn 20.1.0
[2024-01-15 15:30:00 +0000] [12345] [INFO] Listening at: http://0.0.0.0:8000 (12345)
[2024-01-15 15:30:00 +0000] [12345] [INFO] Using worker: sync
[2024-01-15 15:30:00 +0000] [12346] [INFO] Booting worker with pid: 12346
[2024-01-15 15:30:00 +0000] [12347] [INFO] Booting worker with pid: 12347
[2024-01-15 15:30:00 +0000] [12348] [INFO] Booting worker with pid: 12348
[2024-01-15 15:30:00 +0000] [12349] [INFO] Booting worker with pid: 12349

Nginx反向代理配置¶

創建Nginx配置文件：

# /etc/nginx/sites-available/flask_app
server {
    listen 80;
    server_name your-domain.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    location /static {
        alias /path/to/your/app/static;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }

    # 啓用gzip壓縮
    gzip on;
    gzip_vary on;
    gzip_min_length 1024;
    gzip_types text/plain text/css text/xml text/javascript application/javascript application/xml+rss application/json;
}

Docker容器化部署¶

創建Dockerfile：

# Dockerfile
FROM python:3.9-slim

# 設置工作目錄
WORKDIR /app

# 複製依賴文件
COPY requirements.txt .

# 安裝依賴
RUN pip install --no-cache-dir -r requirements.txt

# 複製應用代碼
COPY . .

# 暴露端口
EXPOSE 8000

# 創建非root用戶
RUN useradd --create-home --shell /bin/bash app \
    && chown -R app:app /app
USER app

# 啓動命令
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "wsgi:app"]

創建docker-compose.yml：

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLASK_ENV=production
      - DATABASE_URL=postgresql://user:password@db:5432/blogdb
    depends_on:
      - db
    volumes:
      - ./logs:/app/logs

  db:
    image: postgres:13
    environment:
      - POSTGRES_DB=blogdb
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - web

volumes:
  postgres_data:

構建和運行容器：

# 構建鏡像
docker-compose build

# 啓動服務
docker-compose up -d

# 查看日誌
docker-compose logs -f web

輸出類似如下：

Building web
Step 1/8 : FROM python:3.9-slim
 ---> 1234567890ab
Step 2/8 : WORKDIR /app
 ---> Running in abcdef123456
Removing intermediate container abcdef123456
 ---> 234567890abc
...
Successfully built 345678901bcd
Successfully tagged flask_app_web:latest
Creating flask_app_db_1 ... done
Creating flask_app_web_1 ... done
Creating flask_app_nginx_1 ... done

13.9 性能優化¶

應用性能優化¶

數據庫查詢優化¶

# 優化前：N+1查詢問題
def get_posts_with_authors_slow():
    posts = Post.query.all()
    result = []
    for post in posts:
        result.append({
            'title': post.title,
            'author': post.author.username  # 每次都查詢數據庫
        })
    return result

# 優化後：使用join預加載
def get_posts_with_authors_fast():
    posts = Post.query.options(db.joinedload(Post.author)).all()
    result = []
    for post in posts:
        result.append({
            'title': post.title,
            'author': post.author.username  # 已預加載，無需額外查詢
        })
    return result

# 使用原生SQL進行復雜查詢
def get_post_statistics():
    result = db.session.execute("""
        SELECT 
            u.username,
            COUNT(p.id) as post_count,
            AVG(LENGTH(p.content)) as avg_content_length
        FROM users u
        LEFT JOIN posts p ON u.id = p.user_id
        GROUP BY u.id, u.username
        ORDER BY post_count DESC
    """)

    return [{
        'username': row.username,
        'post_count': row.post_count,
        'avg_content_length': row.avg_content_length
    } for row in result]

緩存策略實現¶

# cache.py
from flask import Flask
from flask_caching import Cache
import redis
import json
from functools import wraps

app = Flask(__name__)

# 配置緩存
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379/0'
cache = Cache(app)

# Redis連接
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 自定義緩存裝飾器
def cache_result(timeout=300, key_prefix=''):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 生成緩存鍵
            cache_key = f"{key_prefix}:{f.__name__}:{hash(str(args) + str(kwargs))}"

            # 嘗試從緩存獲取
            cached_result = redis_client.get(cache_key)
            if cached_result:
                return json.loads(cached_result)

            # 執行函數並緩存結果
            result = f(*args, **kwargs)
            redis_client.setex(cache_key, timeout, json.dumps(result, default=str))

            return result
        return decorated_function
    return decorator

# 使用緩存的API端點
@app.route('/api/posts/popular')
@cache_result(timeout=600, key_prefix='popular_posts')
def get_popular_posts():
    # 模擬複雜的查詢
    posts = db.session.execute("""
        SELECT p.*, u.username, COUNT(c.id) as comment_count
        FROM posts p
        JOIN users u ON p.user_id = u.id
        LEFT JOIN comments c ON p.id = c.post_id
        GROUP BY p.id
        ORDER BY comment_count DESC, p.created_at DESC
        LIMIT 10
    """).fetchall()

    return [{
        'id': post.id,
        'title': post.title,
        'author': post.username,
        'comment_count': post.comment_count
    } for post in posts]

# 緩存失效策略
@app.route('/api/posts', methods=['POST'])
def create_post():
    # 創建文章邏輯...
    post = create_new_post(request.json)

    # 清除相關緩存
    cache.delete_memoized(get_popular_posts)
    redis_client.delete('popular_posts:*')

    return post_schema.jsonify(post), 201

異步處理和後臺任務¶

# tasks.py
from celery import Celery
from flask import Flask
from flask_mail import Mail, Message
import time

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

# 配置郵件
app.config['MAIL_SERVER'] = 'smtp.gmail.com'
app.config['MAIL_PORT'] = 587
app.config['MAIL_USE_TLS'] = True
app.config['MAIL_USERNAME'] = 'your-email@gmail.com'
app.config['MAIL_PASSWORD'] = 'your-password'

mail = Mail(app)

# 初始化Celery
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

@celery.task
def send_email_async(subject, recipients, body):
    """異步發送郵件"""
    with app.app_context():
        msg = Message(
            subject=subject,
            recipients=recipients,
            body=body
        )
        mail.send(msg)
    return f"郵件已發送給: {', '.join(recipients)}"

@celery.task
def generate_report_async(user_id):
    """異步生成用戶報告"""
    # 模擬耗時操作
    time.sleep(10)

    # 生成報告邏輯
    user = User.query.get(user_id)
    posts = Post.query.filter_by(user_id=user_id).all()

    report = {
        'user': user.username,
        'total_posts': len(posts),
        'total_words': sum(len(post.content.split()) for post in posts),
        'generated_at': datetime.utcnow().isoformat()
    }

    # 發送報告郵件
    send_email_async.delay(
        subject=f"用戶報告 - {user.username}",
        recipients=[user.email],
        body=f"您的報告已生成：\n{json.dumps(report, indent=2, ensure_ascii=False)}"
    )

    return report

# API端點
@app.route('/api/users/<int:user_id>/report', methods=['POST'])
def request_user_report(user_id):
    # 啓動異步任務
    task = generate_report_async.delay(user_id)

    return jsonify({
        'message': '報告生成任務已啓動',
        'task_id': task.id,
        'status_url': f'/api/tasks/{task.id}'
    }), 202

@app.route('/api/tasks/<task_id>')
def get_task_status(task_id):
    task = generate_report_async.AsyncResult(task_id)

    if task.state == 'PENDING':
        response = {
            'state': task.state,
            'status': '任務等待中...'
        }
    elif task.state == 'PROGRESS':
        response = {
            'state': task.state,
            'status': task.info.get('status', ''),
            'progress': task.info.get('progress', 0)
        }
    elif task.state == 'SUCCESS':
        response = {
            'state': task.state,
            'result': task.result
        }
    else:
        response = {
            'state': task.state,
            'error': str(task.info)
        }

    return jsonify(response)

啓動Celery worker：

celery -A tasks.celery worker --loglevel=info

輸出類似如下：

 -------------- celery@DESKTOP-ABC123 v5.2.0 (dawn-chorus)
--- ***** ----- 
-- ******* ---- Windows-10-10.0.19041-SP0 2024-01-15 16:00:00
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x1234567890ab
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     redis://localhost:6379/0
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

[tasks]
  . tasks.generate_report_async
  . tasks.send_email_async

[2024-01-15 16:00:00,123: INFO/MainProcess] Connected to redis://localhost:6379/0
[2024-01-15 16:00:00,124: INFO/MainProcess] mingle: searching for available workers
[2024-01-15 16:00:00,125: INFO/MainProcess] mingle: all alone
[2024-01-15 16:00:00,126: INFO/MainProcess] celery@DESKTOP-ABC123 ready.

監控和日誌¶

應用監控配置¶

# monitoring.py
from flask import Flask, request, g
from prometheus_flask_exporter import PrometheusMetrics
import logging
import time
import psutil
import os

app = Flask(__name__)

# 配置Prometheus監控
metrics = PrometheusMetrics(app)
metrics.info('flask_app_info', 'Application info', version='1.0.0')

# 配置日誌
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 請求日誌中間件
@app.before_request
def before_request():
    g.start_time = time.time()
    logger.info(f"Request started: {request.method} {request.url}")

@app.after_request
def after_request(response):
    duration = time.time() - g.start_time
    logger.info(
        f"Request completed: {request.method} {request.url} "
        f"Status: {response.status_code} Duration: {duration:.3f}s"
    )
    return response

# 系統監控端點
@app.route('/health')
def health_check():
    return jsonify({
        'status': 'healthy',
        'timestamp': datetime.utcnow().isoformat(),
        'version': '1.0.0'
    })

@app.route('/metrics/system')
def system_metrics():
    return jsonify({
        'cpu_percent': psutil.cpu_percent(),
        'memory_percent': psutil.virtual_memory().percent,
        'disk_percent': psutil.disk_usage('/').percent,
        'process_count': len(psutil.pids()),
        'load_average': os.getloadavg() if hasattr(os, 'getloadavg') else None
    })

# 自定義監控裝飾器
def monitor_performance(operation_name):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            start_time = time.time()
            try:
                result = f(*args, **kwargs)
                duration = time.time() - start_time
                logger.info(f"Operation {operation_name} completed in {duration:.3f}s")
                return result
            except Exception as e:
                duration = time.time() - start_time
                logger.error(f"Operation {operation_name} failed after {duration:.3f}s: {str(e)}")
                raise
        return decorated_function
    return decorator

# 使用監控裝飾器
@app.route('/api/posts')
@monitor_performance('get_posts')
def get_posts():
    posts = Post.query.all()
    return posts_schema.jsonify(posts)

總結¶

本章全面介紹了Python Web開發的各個方面，從基礎概念到實際應用。我們學習了：

Web開發基礎：理解了Web應用架構、HTTP協議和Python Web生態系統
Flask框架：掌握了輕量級框架的使用，包括路由、模板、數據庫集成
Django框架：學習了全棧框架的MVC架構和完整的開發流程
FastAPI框架：瞭解了現代API框架的特性和異步編程
數據庫操作：掌握了SQLite基礎操作和SQLAlchemy ORM的使用
前端集成：學習了AJAX技術和前後端數據交互
API設計：理解了RESTful API的設計原則和最佳實踐
部署運維：掌握了生產環境部署和容器化技術
性能優化：學習了緩存策略、異步處理和監控技術

通過本章的學習，讀者應該能夠：
- 選擇合適的Web框架進行項目開發
- 設計和實現完整的Web應用程序
- 處理數據庫操作和前後端交互
- 部署應用到生產環境
- 優化應用性能和監控系統狀態

Web開發是一個不斷發展的領域，建議讀者繼續關注新技術和最佳實踐，在實際項目中不斷提升技能。route(‘/’)
def home():
return ‘

歡迎來到Python Web開發世界！

‘

@app.route(‘/user/‘)
def user_profile(name):
return f’

用戶：{name}

歡迎訪問您的個人主頁！

‘

@app.route(‘/api/data’, methods=[‘GET’, ‘POST’])
def api_data():
if request.method == ‘POST’:
data = request.get_json()
return jsonify({
‘message’: ‘數據接收成功’,
‘received_data’: data,
‘status’: ‘success’
})
else:
return jsonify({
‘message’: ‘Hello from Flask API’,
‘version’: ‘1.0’,
‘endpoints’: [‘/api/data’, ‘/user/‘]
})

if name == ‘main’:
app.run(debug=True, host=‘0.0.0.0’, port=5000)

運行Flask應用：
```shell
python app.py

輸出類似如下：

 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:5000
 * Running on http://192.168.1.100:5000
 * Debug mode: on
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 123-456-789

Flask模板系統¶

Flask使用Jinja2模板引擎。首先創建模板目錄結構：

flask_project/
├── app.py
├── templates/
│   ├── base.html
│   ├── index.html
│   └── user.html
└── static/
    ├── css/
    │   └── style.css
    └── js/
        └── main.js

創建基礎模板：

<!-- templates/base.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{% block title %}Flask Web應用{% endblock %}</title>
    <link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
</head>
<body>
    <nav class="navbar">
        <div class="nav-container">
            <h1><a href="{{ url_for('home') }}">Flask Demo</a></h1>
            <ul class="nav-menu">
                <li><a href="{{ url_for('home') }}">首頁</a></li>
                <li><a href="{{ url_for('about') }}">關於</a></li>
            </ul>
        </div>
    </nav>

    <main class="container">
        {% block content %}{% endblock %}
    </main>

    <footer>
        <p>&copy; 2024 Python Web開發教程</p>
    </footer>

    <script src="{{ url_for('static', filename='js/main.js') }}"></script>
</body>
</html>

創建首頁模板：

<!-- templates/index.html -->
{% extends "base.html" %}

{% block title %}首頁 - Flask Web應用{% endblock %}

{% block content %}
<div class="hero">
    <h1>歡迎來到Flask Web開發</h1>
    <p>這是一個完整的Flask應用示例</p>

    <div class="features">
        <div class="feature">
            <h3>路由系統</h3>
            <p>靈活的URL路由配置</p>
        </div>
        <div class="feature">
            <h3>模板引擎</h3>
            <p>強大的Jinja2模板系統</p>
        </div>
        <div class="feature">
            <h3>數據庫集成</h3>
            <p>SQLAlchemy ORM支持</p>
        </div>
    </div>

    <form class="demo-form" method="POST" action="{{ url_for('submit_form') }}">
        <h3>用戶信息表單</h3>
        <input type="text" name="username" placeholder="用戶名" required>
        <input type="email" name="email" placeholder="郵箱" required>
        <textarea name="message" placeholder="留言" rows="4"></textarea>
        <button type="submit">提交</button>
    </form>
</div>
{% endblock %}

Flask數據庫集成¶

安裝Flask-SQLAlchemy：

pip install Flask-SQLAlchemy

創建數據庫模型：

# models.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'your-secret-key-here'

db = SQLAlchemy(app)

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)

    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

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

# 路由和視圖函數
@app.route('/')
def index():
    posts = Post.query.order_by(Post.created_at.desc()).all()
    return render_template('index.html', posts=posts)

@app.route('/user/<username>')
def user_posts(username):
    user = User.query.filter_by(username=username).first_or_404()
    posts = Post.query.filter_by(author=user).order_by(Post.created_at.desc()).all()
    return render_template('user_posts.html', user=user, posts=posts)

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

        # 查找或創建用戶
        user = User.query.filter_by(username=username).first()
        if not user:
            user = User(username=username, email=f"{username}@example.com")
            db.session.add(user)
            db.session.commit()

        # 創建文章
        post = Post(title=title, content=content, author=user)
        db.session.add(post)
        db.session.commit()

        return redirect(url_for('index'))

    return render_template('create_post.html')

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

運行應用並創建數據庫：

python models.py

輸出類似如下：

 * Running on http://127.0.0.1:5000
 * Debug mode: on
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 234-567-890

13.3 Django全棧框架¶

Django安裝和項目創建¶

安裝Django：

pip install Django

創建Django項目：

django-admin startproject myblog
cd myblog
python manage.py startapp blog

輸出類似如下：

C:\Users\Python\myblog>django-admin startproject myblog
C:\Users\Python\myblog>cd myblog
C:\Users\Python\myblog\myblog>python manage.py startapp blog

項目結構如下：

myblog/
├── manage.py
├── myblog/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── blog/
    ├── __init__.py
    ├── admin.py
    ├── apps.py
    ├── models.py
    ├── tests.py
    └── views.py

Django模型定義¶

在blog/models.py中定義數據模型：

# blog/models.py
from django.db import models
from django.contrib.auth.models import User
from django.urls import reverse

class Category(models.Model):
    name = models.CharField(max_length=100, unique=True)
    description = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name_plural = "Categories"

    def __str__(self):
        return self.name

class Post(models.Model):
    STATUS_CHOICES = [
        ('draft', '草稿'),
        ('published', '已發佈'),
    ]

    title = models.CharField(max_length=200)
    slug = models.SlugField(max_length=200, unique=True)
    content = models.TextField()
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
    category = models.ForeignKey(Category, on_delete=models.SET_NULL, null=True, blank=True)
    status = models.CharField(max_length=10, choices=STATUS_CHOICES, default='draft')
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        ordering = ['-created_at']

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        return reverse('blog:post_detail', kwargs={'slug': self.slug})

class Comment(models.Model):
    post = models.ForeignKey(Post, on_delete=models.CASCADE, related_name='comments')
    author_name = models.CharField(max_length=100)
    author_email = models.EmailField()
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    is_approved = models.BooleanField(default=False)

    class Meta:
        ordering = ['created_at']

    def __str__(self):
        return f'Comment by {self.author_name} on {self.post.title}'

Django視圖和URL配置¶

創建視圖函數：

# blog/views.py
from django.shortcuts import render, get_object_or_404, redirect
from django.http import JsonResponse
from django.core.paginator import Paginator
from django.contrib import messages
from .models import Post, Category, Comment
from .forms import CommentForm

def post_list(request):
    posts = Post.objects.filter(status='published')
    category_id = request.GET.get('category')

    if category_id:
        posts = posts.filter(category_id=category_id)

    paginator = Paginator(posts, 5)  # 每頁顯示5篇文章
    page_number = request.GET.get('page')
    page_obj = paginator.get_page(page_number)

    categories = Category.objects.all()

    context = {
        'page_obj': page_obj,
        'categories': categories,
        'current_category': category_id
    }
    return render(request, 'blog/post_list.html', context)

def post_detail(request, slug):
    post = get_object_or_404(Post, slug=slug, status='published')
    comments = post.comments.filter(is_approved=True)

    if request.method == 'POST':
        form = CommentForm(request.POST)
        if form.is_valid():
            comment = form.save(commit=False)
            comment.post = post
            comment.save()
            messages.success(request, '評論提交成功，等待審覈！')
            return redirect('blog:post_detail', slug=slug)
    else:
        form = CommentForm()

    context = {
        'post': post,
        'comments': comments,
        'form': form
    }
    return render(request, 'blog/post_detail.html', context)

def category_posts(request, category_id):
    category = get_object_or_404(Category, id=category_id)
    posts = Post.objects.filter(category=category, status='published')

    paginator = Paginator(posts, 5)
    page_number = request.GET.get('page')
    page_obj = paginator.get_page(page_number)

    context = {
        'category': category,
        'page_obj': page_obj
    }
    return render(request, 'blog/category_posts.html', context)

配置URL路由：

# blog/urls.py
from django.urls import path
from . import views

app_name = 'blog'

urlpatterns = [
    path('', views.post_list, name='post_list'),
    path('post/<slug:slug>/', views.post_detail, name='post_detail'),
    path('category/<int:category_id>/', views.category_posts, name='category_posts'),
]

# myblog/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('blog.urls')),
]

數據庫遷移¶

執行數據庫遷移：

python manage.py makemigrations
python manage.py migrate

輸出類似如下：

Migrations for 'blog':
  blog\migrations\0001_initial.py
    - Create model Category
    - Create model Post
    - Create model Comment
Operations to perform:
  Apply all migrations: admin, auth, blog, contenttypes, sessions
Running migrations:
  Applying contenttypes.0001_initial... OK
  Applying auth.0001_initial... OK
  Applying admin.0001_initial... OK
  Applying admin.0002_logentry_remove_auto_add... OK
  Applying admin.0003_logentry_add_action_flag_choices... OK
  Applying contenttypes.0002_remove_content_type_name... OK
  Applying auth.0002_alter_permission_name_max_length... OK
  Applying auth.0003_alter_user_email_max_length... OK
  Applying auth.0004_alter_user_username_opts... OK
  Applying auth.0005_alter_user_last_login_null... OK
  Applying auth.0006_require_contenttypes_0002... OK
  Applying auth.0007_alter_validators_add_error_messages... OK
  Applying auth.0008_alter_user_username_max_length... OK
  Applying auth.0009_alter_user_last_name_max_length... OK
  Applying auth.0010_alter_group_name_max_length... OK
  Applying auth.0011_update_proxy_permissions... OK
  Applying auth.0012_alter_user_first_name_max_length... OK
  Applying blog.0001_initial... OK
  Applying sessions.0001_initial... OK

創建超級用戶：

python manage.py createsuperuser

輸出類似如下：

Username (leave blank to use 'admin'): admin
Email address: admin@example.com
Password: 
Password (again): 
Superuser created successfully.

13.4 FastAPI現代框架¶

FastAPI安裝和基礎應用¶

安裝FastAPI和相關依賴：

pip install fastapi uvicorn python-multipart

創建第一個FastAPI應用：

# main.py
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, EmailStr
from typing import List, Optional
from datetime import datetime
import uvicorn

app = FastAPI(
    title="Python Web開發API",
    description="這是一個完整的FastAPI應用示例",
    version="1.0.0"
)

security = HTTPBearer()

# 數據模型
class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True

class PostBase(BaseModel):
    title: str
    content: str
    category: Optional[str] = None

class PostCreate(PostBase):
    pass

class Post(PostBase):
    id: int
    author_id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

# 模擬數據庫
fake_users_db = []
fake_posts_db = []
user_id_counter = 1
post_id_counter = 1

# 依賴注入
def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    # 這裏應該驗證JWT token，簡化處理
    token = credentials.credentials
    if token != "valid-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"user_id": 1, "username": "testuser"}

# API路由
@app.get("/")
async def root():
    return {
        "message": "歡迎使用FastAPI",
        "version": "1.0.0",
        "docs": "/docs",
        "redoc": "/redoc"
    }

@app.post("/users/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    global user_id_counter

    # 檢查用戶名是否已存在
    for existing_user in fake_users_db:
        if existing_user["username"] == user.username:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="用戶名已存在"
            )

    new_user = {
        "id": user_id_counter,
        "username": user.username,
        "email": user.email,
        "full_name": user.full_name,
        "is_active": True,
        "created_at": datetime.now()
    }
    fake_users_db.append(new_user)
    user_id_counter += 1

    return new_user

@app.get("/users/", response_model=List[User])
async def get_users(skip: int = 0, limit: int = 10):
    return fake_users_db[skip : skip + limit]

@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    for user in fake_users_db:
        if user["id"] == user_id:
            return user
    raise HTTPException(status_code=404, detail="用戶不存在")

@app.post("/posts/", response_model=Post, status_code=status.HTTP_201_CREATED)
async def create_post(post: PostCreate, current_user: dict = Depends(get_current_user)):
    global post_id_counter

    new_post = {
        "id": post_id_counter,
        "title": post.title,
        "content": post.content,
        "category": post.category,
        "author_id": current_user["user_id"],
        "created_at": datetime.now(),
        "updated_at": datetime.now()
    }
    fake_posts_db.append(new_post)
    post_id_counter += 1

    return new_post

@app.get("/posts/", response_model=List[Post])
async def get_posts(skip: int = 0, limit: int = 10, category: Optional[str] = None):
    posts = fake_posts_db
    if category:
        posts = [post for post in posts if post.get("category") == category]
    return posts[skip : skip + limit]

@app.get("/posts/{post_id}", response_model=Post)
async def get_post(post_id: int):
    for post in fake_posts_db:
        if post["id"] == post_id:
            return post
    raise HTTPException(status_code=404, detail="文章不存在")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

運行FastAPI應用：

uvicorn main:app --reload

輸出類似如下：

INFO:     Will watch for changes in these directories: ['C:\\Users\\Python\\fastapi_demo']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [12345] using StatReload
INFO:     Started server process [12346]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

訪問 http://127.0.0.1:8000/docs 可以看到自動生成的API文檔。

FastAPI數據庫集成¶

安裝SQLAlchemy和數據庫驅動：

pip install sqlalchemy databases[sqlite] alembic

創建數據庫配置：

# database.py
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from databases import Database

DATABASE_URL = "sqlite:///./fastapi_blog.db"

# SQLAlchemy
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Databases query builder
database = Database(DATABASE_URL)
metadata = MetaData()

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

創建數據庫模型：

# models.py
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True, index=True)
    full_name = Column(String(100))
    hashed_password = Column(String(100))
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), index=True)
    content = Column(Text)
    category = Column(String(50))
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    author = relationship("User", back_populates="posts")

13.5 數據庫操作¶

SQLite基礎操作¶

創建一個完整的數據庫操作示例：

# sqlite_demo.py
import sqlite3
from datetime import datetime
import json

class BlogDatabase:
    def __init__(self, db_name="blog.db"):
        self.db_name = db_name
        self.init_database()

    def init_database(self):
        """初始化數據庫表"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()

        # 創建用戶表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                password_hash TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        # 創建文章表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS posts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                content TEXT NOT NULL,
                author_id INTEGER,
                category TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (author_id) REFERENCES users (id)
            )
        ''')

        conn.commit()
        conn.close()
        print("數據庫初始化完成")

    def create_user(self, username, email, password_hash):
        """創建用戶"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()

        try:
            cursor.execute(
                "INSERT INTO users (username, email, password_hash) VALUES (?, ?, ?)",
                (username, email, password_hash)
            )
            conn.commit()
            user_id = cursor.lastrowid
            print(f"用戶創建成功，ID: {user_id}")
            return user_id
        except sqlite3.IntegrityError as e:
            print(f"用戶創建失敗: {e}")
            return None
        finally:
            conn.close()

    def create_post(self, title, content, author_id, category=None):
        """創建文章"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()

        cursor.execute(
            "INSERT INTO posts (title, content, author_id, category) VALUES (?, ?, ?, ?)",
            (title, content, author_id, category)
        )
        conn.commit()
        post_id = cursor.lastrowid
        conn.close()

        print(f"文章創建成功，ID: {post_id}")
        return post_id

    def get_posts_with_authors(self):
        """獲取文章列表（包含作者信息）"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()

        cursor.execute('''
            SELECT p.id, p.title, p.content, p.category, p.created_at,
                   u.username, u.email
            FROM posts p
            JOIN users u ON p.author_id = u.id
            ORDER BY p.created_at DESC
        ''')

        posts = cursor.fetchall()
        conn.close()

        return posts

    def search_posts(self, keyword):
        """搜索文章"""
        conn = sqlite3.connect(self.db_name)
        cursor = conn.cursor()

        cursor.execute(
            "SELECT * FROM posts WHERE title LIKE ? OR content LIKE ?",
            (f"%{keyword}%", f"%{keyword}%")
        )

        results = cursor.fetchall()
        conn.close()

        return results

# 使用示例
if __name__ == "__main__":
    db = BlogDatabase()

    # 創建用戶
    user1_id = db.create_user("張三", "zhangsan@example.com", "hashed_password_123")
    user2_id = db.create_user("李四", "lisi@example.com", "hashed_password_456")

    # 創建文章
    if user1_id:
        db.create_post("Python Web開發入門", "這是一篇關於Python Web開發的文章...", user1_id, "技術")
        db.create_post("Flask框架詳解", "Flask是一個輕量級的Web框架...", user1_id, "技術")

    if user2_id:
        db.create_post("Django實戰項目", "使用Django開發一個完整的博客系統...", user2_id, "項目")

    # 查詢文章
    posts = db.get_posts_with_authors()
    print("\n所有文章:")
    for post in posts:
        print(f"ID: {post[0]}, 標題: {post[1]}, 作者: {post[5]}, 分類: {post[3]}")

    # 搜索文章
    search_results = db.search_posts("Flask")
    print("\n搜索結果:")
    for result in search_results:
        print(f"找到文章: {result[1]}")

運行數據庫操作示例：

python sqlite_demo.py

輸出類似如下：

數據庫初始化完成
用戶創建成功，ID: 1
用戶創建成功，ID: 2
文章創建成功，ID: 1
文章創建成功，ID: 2
文章創建成功，ID: 3

所有文章:
ID: 3, 標題: Django實戰項目, 作者: 李四, 分類: 項目
ID: 2, 標題: Flask框架詳解, 作者: 張三, 分類: 技術
ID: 1, 標題: Python Web開發入門, 作者: 張三, 分類: 技術

搜索結果:
找到文章: Flask框架詳解

SQLAlchemy ORM使用¶

創建一個完整的ORM示例：

# orm_demo.py
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime

# 數據庫配置
engine = create_engine('sqlite:///orm_blog.db', echo=True)
Base = declarative_base()
Session = sessionmaker(bind=engine)

# 模型定義
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship("Post", back_populates="author")

    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.utcnow)

    author = relationship("User", back_populates="posts")

    def __repr__(self):
        return f"<Post(title='{self.title}', author='{self.author.username if self.author else None}')>"

# 創建表
Base.metadata.create_all(engine)

# 數據操作函數
def create_sample_data():
    session = Session()

    # 創建用戶
    user1 = User(username="python_dev", email="dev@python.com")
    user2 = User(username="web_master", email="master@web.com")

    session.add_all([user1, user2])
    session.commit()

    # 創建文章
    post1 = Post(
        title="SQLAlchemy ORM教程",
        content="這是一篇詳細的SQLAlchemy ORM使用教程...",
        author=user1
    )

    post2 = Post(
        title="Web開發最佳實踐",
        content="分享一些Web開發的最佳實踐和經驗...",
        author=user1
    )

    post3 = Post(
        title="數據庫設計原則",
        content="介紹關係型數據庫設計的基本原則...",
        author=user2
    )

    session.add_all([post1, post2, post3])
    session.commit()

    print("示例數據創建完成")
    session.close()

def query_data():
    session = Session()

    # 查詢所有用戶
    print("\n所有用戶:")
    users = session.query(User).all()
    for user in users:
        print(f"  {user}")

    # 查詢特定用戶的文章
    print("\npython_dev的文章:")
    user = session.query(User).filter_by(username="python_dev").first()
    if user:
        for post in user.posts:
            print(f"  {post.title} - {post.created_at}")

    # 聯合查詢
    print("\n所有文章（包含作者信息）:")
    posts = session.query(Post).join(User).all()
    for post in posts:
        print(f"  《{post.title}》 by {post.author.username}")

    # 條件查詢
    print("\n標題包含'教程'的文章:")
    tutorial_posts = session.query(Post).filter(Post.title.contains('教程')).all()
    for post in tutorial_posts:
        print(f"  {post.title}")

    session.close()

if __name__ == "__main__":
    create_sample_data()
    query_data()

運行ORM示例：

python orm_demo.py

輸出類似如下：

2024-01-15 14:30:00,123 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-01-15 14:30:00,124 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("users")
2024-01-15 14:30:00,124 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-01-15 14:30:00,125 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("users")
2024-01-15 14:30:00,125 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-01-15 14:30:00,126 INFO sqlalchemy.engine.Engine PRAGMA main.table_info("posts")
2024-01-15 14:30:00,126 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-01-15 14:30:00,127 INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("posts")
2024-01-15 14:30:00,127 INFO sqlalchemy.engine.Engine [raw sql] ()
2024-01-15 14:30:00,128 INFO sqlalchemy.engine.Engine 
CREATE TABLE users (
    id INTEGER NOT NULL, 
    username VARCHAR(50) NOT NULL, 
    email VARCHAR(100) NOT NULL, 
    created_at DATETIME, 
    PRIMARY KEY (id), 
    UNIQUE (username), 
    UNIQUE (email)
)

2024-01-15 14:30:00,128 INFO sqlalchemy.engine.Engine [no key 0.00000s] ()
2024-01-15 14:30:00,129 INFO sqlalchemy.engine.Engine 
CREATE TABLE posts (
    id INTEGER NOT NULL, 
    title VARCHAR(200) NOT NULL, 
    content TEXT NOT NULL, 
    author_id INTEGER, 
    created_at DATETIME, 
    PRIMARY KEY (id), 
    FOREIGN KEY(author_id) REFERENCES users (id)
)

2024-01-15 14:30:00,129 INFO sqlalchemy.engine.Engine [no key 0.00000s] ()
2024-01-15 14:30:00,130 INFO sqlalchemy.engine.Engine COMMIT
示例數據創建完成

所有用戶:
  <User(username='python_dev', email='dev@python.com')>
  <User(username='web_master', email='master@web.com')>

python_dev的文章:
  SQLAlchemy ORM教程 - 2024-01-15 14:30:00.135000
  Web開發最佳實踐 - 2024-01-15 14:30:00.136000

所有文章（包含作者信息）:
  《SQLAlchemy ORM教程》 by python_dev
  《Web開發最佳實踐》 by python_dev
  《數據庫設計原則》 by web_master

標題包含'教程'的文章:
  SQLAlchemy ORM教程

13.6 前端集成¶

AJAX和API調用¶

創建一個完整的前後端交互示例：

<!-- templates/ajax_demo.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>AJAX API調用示例</title>
    <style>
        body {
            font-family: 'Microsoft YaHei', sans-serif;
            max-width: 800px;
            margin: 0 auto;
            padding: 20px;
            background-color: #f5f5f5;
        }
        .container {
            background: white;
            padding: 30px;
            border-radius: 10px;
            box-shadow: 0 2px 10px rgba(0,0,0,0.1);
        }
        .form-group {
            margin-bottom: 20px;
        }
        label {
            display: block;
            margin-bottom: 5px;
            font-weight: bold;
        }
        input, textarea {
            width: 100%;
            padding: 10px;
            border: 1px solid #ddd;
            border-radius: 5px;
            font-size: 14px;
        }
        button {
            background: #007bff;
            color: white;
            padding: 10px 20px;
            border: none;
            border-radius: 5px;
            cursor: pointer;
            margin-right: 10px;
        }
        button:hover {
            background: #0056b3;
        }
        .posts {
            margin-top: 30px;
        }
        .post {
            background: #f8f9fa;
            padding: 15px;
            margin-bottom: 15px;
            border-radius: 5px;
            border-left: 4px solid #007bff;
        }
        .loading {
            color: #666;
            font-style: italic;
        }
        .error {
            color: #dc3545;
            background: #f8d7da;
            padding: 10px;
            border-radius: 5px;
            margin: 10px 0;
        }
        .success {
            color: #155724;
            background: #d4edda;
            padding: 10px;
            border-radius: 5px;
            margin: 10px 0;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>Python Web API 交互示例</h1>

        <!-- 創建文章表單 -->
        <form id="postForm">
            <div class="form-group">
                <label for="title">文章標題:</label>
                <input type="text" id="title" name="title" required>
            </div>
            <div class="form-group">
                <label for="content">文章內容:</label>
                <textarea id="content" name="content" rows="5" required></textarea>
            </div>
            <div class="form-group">
                <label for="category">分類:</label>
                <input type="text" id="category" name="category">
            </div>
            <button type="submit">發佈文章</button>
            <button type="button" onclick="loadPosts()">刷新文章列表</button>
        </form>

        <div id="message"></div>

        <!-- 文章列表 -->
        <div class="posts">
            <h2>文章列表</h2>
            <div id="postsList">
                <div class="loading">加載中...</div>
            </div>
        </div>
    </div>

    <script>
        // API基礎URL
        const API_BASE = 'http://localhost:8000';

        // 顯示消息
        function showMessage(message, type = 'success') {
            const messageDiv = document.getElementById('message');
            messageDiv.innerHTML = `<div class="${type}">${message}</div>`;
            setTimeout(() => {
                messageDiv.innerHTML = '';
            }, 3000);
        }

        // 發送POST請求創建文章
        document.getElementById('postForm').addEventListener('submit', async function(e) {
            e.preventDefault();

            const formData = new FormData(this);
            const postData = {
                title: formData.get('title'),
                content: formData.get('content'),
                category: formData.get('category') || null
            };

            try {
                const response = await fetch(`${API_BASE}/posts/`, {
                    method: 'POST',
                    headers: {
                        'Content-Type': 'application/json',
                        'Authorization': 'Bearer valid-token'
                    },
                    body: JSON.stringify(postData)
                });

                if (response.ok) {
                    const result = await response.json();
                    showMessage('文章發佈成功！', 'success');
                    this.reset();
                    loadPosts(); // 刷新文章列表
                } else {
                    const error = await response.json();
                    showMessage(`發佈失敗: ${error.detail}`, 'error');
                }
            } catch (error) {
                showMessage(`網絡錯誤: ${error.message}`, 'error');
            }
        });

        // 加載文章列表
        async function loadPosts() {
            const postsDiv = document.getElementById('postsList');
            postsDiv.innerHTML = '<div class="loading">加載中...</div>';

            try {
                const response = await fetch(`${API_BASE}/posts/`);

                if (response.ok) {
                    const posts = await response.json();

                    if (posts.length === 0) {
                        postsDiv.innerHTML = '<p>暫無文章</p>';
                        return;
                    }

                    let html = '';
                    posts.forEach(post => {
                        html += `
                            <div class="post">
                                <h3>${post.title}</h3>
                                <p>${post.content}</p>
                                <small>分類: ${post.category || '未分類'} | 創建時間: ${new Date(post.created_at).toLocaleString()}</small>
                            </div>
                        `;
                    });

                    postsDiv.innerHTML = html;
                } else {
                    postsDiv.innerHTML = '<div class="error">加載文章失敗</div>';
                }
            } catch (error) {
                postsDiv.innerHTML = `<div class="error">網絡錯誤: ${error.message}</div>`;
            }
        }

        // 頁面加載時獲取文章列表
        document.addEventListener('DOMContentLoaded', loadPosts);

        // 使用Fetch API的高級示例
        async function advancedApiCall() {
            try {
                // 併發請求多個API
                const [usersResponse, postsResponse] = await Promise.all([
                    fetch(`${API_BASE}/users/`),
                    fetch(`${API_BASE}/posts/`)
                ]);

                const users = await usersResponse.json();
                const posts = await postsResponse.json();

                console.log('用戶數據:', users);
                console.log('文章數據:', posts);

                // 處理數據...
            } catch (error) {
                console.error('API調用失敗:', error);
            }
        }

        // 帶有重試機制的API調用
        async function apiCallWithRetry(url, options = {}, maxRetries = 3) {
            for (let i = 0; i < maxRetries; i++) {
                try {
                    const response = await fetch(url, options);
                    if (response.ok) {
                        return await response.json();
                    }
                    throw new Error(`HTTP ${response.status}: ${response.statusText}`);
                } catch (error) {
                    if (i === maxRetries - 1) throw error;
                    console.log(`請求失敗，${1000 * (i + 1)}ms後重試...`);
                    await new Promise(resolve => setTimeout(resolve, 1000 * (i + 1)));
                }
            }
        }
    </script>
</body>
</html>

13.7 Web服務和API設計¶

RESTful API設計實踐¶

創建一個完整的RESTful API示例：
```python

restful_api.py¶

from flask import Flask, request, jsonify, abort
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from marshmallow import ValidationError
from datetime import datetime
import os

app = Flask(name)
app.config[‘SQLALCHEMY_DATABASE_URI’] = ‘sqlite:///restful_blog.db’
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False

db = SQLAlchemy(app)
ma = Marshmallow(app)

數據模型¶

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, cascade=’all, delete-orphan’)

class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
category = db.Column(db.String(50))
user_id = db.Column(db.Integer, db.ForeignKey(‘user.id’), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

序列化Schema¶

class UserSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = User
load_instance = True
include_fk = True

class PostSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = Post
load_instance = True
include_fk = True

author = ma.Nested(UserSchema, only=['id', 'username'])

初始化Schema¶

user_schema = UserSchema()
users_schema = UserSchema(many=True)
post_schema = PostSchema()
posts_schema = PostSchema(many=True)

錯誤處理¶

@app.errorhandler(404)
def not_found(error):
return jsonify({‘error’: ‘Resource not found’}), 404

@app.errorhandler(400)
def bad_request(error):
return jsonify({‘error’: ‘Bad request’}), 400

@app.

第13章 Web開發¶

前言¶

13.1 Web開發基礎¶

Web應用架構概述¶

HTTP協議基礎¶

Python Web開發生態¶

13.2 Flask輕量級框架¶

Flask安裝和環境配置¶

第一個Flask應用¶

API測試示例¶

13.8 部署和運維¶

生產環境部署¶

使用Gunicorn部署Flask應用¶

Nginx反向代理配置¶

Docker容器化部署¶

13.9 性能優化¶

應用性能優化¶

數據庫查詢優化¶

緩存策略實現¶

異步處理和後臺任務¶

監控和日誌¶

應用監控配置¶

總結¶

歡迎來到Python Web開發世界！

用戶：{name}

Flask模板系統¶

Flask數據庫集成¶

13.3 Django全棧框架¶

Django安裝和項目創建¶

Django模型定義¶

Django視圖和URL配置¶

數據庫遷移¶

13.4 FastAPI現代框架¶

FastAPI安裝和基礎應用¶

FastAPI數據庫集成¶

13.5 數據庫操作¶

SQLite基礎操作¶

SQLAlchemy ORM使用¶

13.6 前端集成¶

AJAX和API調用¶

13.7 Web服務和API設計¶

RESTful API設計實踐¶

restful_api.py¶

數據模型¶

序列化Schema¶

初始化Schema¶

錯誤處理¶

相關文章