跳到主要内容

第13章:项目实战

作为 Java 开发者,你已经熟悉了如何构建 Java 项目。本章将通过一个实际的 Python 项目,帮助你将所学的 Python 知识应用到实践中。我们将创建一个简单的 Web 应用,展示 Java 开发者如何快速上手 Python 项目开发。

13.1 项目概述

我们将创建一个 个人任务管理系统,具有以下功能:

  • 用户注册和登录
  • 任务的创建、编辑、删除和查询
  • 任务状态管理(待办、进行中、已完成)
  • 任务分类
  • 数据持久化(使用 SQLite 数据库)

技术栈

  • 后端:Python + FastAPI
  • 数据库:SQLite
  • ORM:SQLAlchemy
  • 前端:HTML + CSS + JavaScript

13.2 项目结构

task-manager/
├── app/
│ ├── __init__.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ └── tasks.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── task.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── task.py
│ ├── core/
│ │ ├── __init__.py
│ │ ├── config.py
│ │ └── security.py
│ └── db/
│ ├── __init__.py
│ └── session.py
├── static/
│ ├── css/
│ │ └── style.css
│ └── js/
│ └── script.js
├── templates/
│ ├── base.html
│ ├── login.html
│ ├── register.html
│ └── tasks.html
├── main.py
├── requirements.txt
└── README.md

13.3 环境搭建

1. 创建项目目录

mkdir task-manager
cd task-manager

2. 创建虚拟环境

# Windows
python -m venv venv
venv\Scripts\activate

# macOS/Linux
python3 -m venv venv
source venv/bin/activate

3. 安装依赖

创建 requirements.txt 文件:

fastapi
uvicorn[standard]
sqlalchemy
pydantic
pydantic-settings
python-jose[cryptography]
python-multipart
passlib[bcrypt]
alembic

安装依赖:

pip install -r requirements.txt

13.4 数据库模型

1. 数据库配置

创建 app/core/config.py

from pydantic_settings import BaseSettings
from typing import Optional


class Settings(BaseSettings):
"""应用配置"""
PROJECT_NAME: str = "Task Manager"
API_V1_STR: str = "/api/v1"
SECRET_KEY: str = "your-secret-key-here"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

# 数据库配置
DATABASE_URL: str = "sqlite:///./task.db"

class Config:
case_sensitive = True


settings = Settings()

2. 数据库会话

创建 app/db/session.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from app.core.config import settings

engine = create_engine(
settings.DATABASE_URL,
connect_args={"check_same_thread": False} # SQLite 特定配置
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


def get_db():
"""获取数据库会话"""
db = SessionLocal()
try:
yield db
finally:
db.close()

3. 用户模型

创建 app/models/user.py

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship

from app.db.session import Base


class User(Base):
"""用户模型"""
__tablename__ = "users"

id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
email = Column(String, unique=True, index=True, nullable=False)
password_hash = Column(String, nullable=False)

# 关系
tasks = relationship("Task", back_populates="user")

4. 任务模型

创建 app/models/task.py

from sqlalchemy import Column, Integer, String, Text, ForeignKey, Enum
from sqlalchemy.orm import relationship
import enum

from app.db.session import Base


class TaskStatus(str, enum.Enum):
"""任务状态"""
TODO = "todo"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"


class Task(Base):
"""任务模型"""
__tablename__ = "tasks"

id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
description = Column(Text)
status = Column(Enum(TaskStatus), default=TaskStatus.TODO)
category = Column(String)
user_id = Column(Integer, ForeignKey("users.id"))

# 关系
user = relationship("User", back_populates="tasks")

5. 初始化数据库

创建 app/db/__init__.py

from app.db.session import Base, engine
from app.models import user, task


def init_db():
"""初始化数据库"""
# 创建所有表
Base.metadata.create_all(bind=engine)

13.5 数据验证模式

1. 用户模式

创建 app/schemas/user.py

from pydantic import BaseModel, EmailStr
from typing import Optional


class UserBase(BaseModel):
"""用户基础模式"""
username: str
email: EmailStr


class UserCreate(UserBase):
"""用户创建模式"""
password: str


class UserLogin(BaseModel):
"""用户登录模式"""
email: EmailStr
password: str


class User(UserBase):
"""用户响应模式"""
id: int

class Config:
from_attributes = True


class Token(BaseModel):
"""令牌模式"""
access_token: str
token_type: str


class TokenData(BaseModel):
"""令牌数据模式"""
user_id: Optional[int] = None

2. 任务模式

创建 app/schemas/task.py

from pydantic import BaseModel
from typing import Optional
from app.models.task import TaskStatus


class TaskBase(BaseModel):
"""任务基础模式"""
title: str
description: Optional[str] = None
status: TaskStatus = TaskStatus.TODO
category: Optional[str] = None


class TaskCreate(TaskBase):
"""任务创建模式"""
pass


class TaskUpdate(BaseModel):
"""任务更新模式"""
title: Optional[str] = None
description: Optional[str] = None
status: Optional[TaskStatus] = None
category: Optional[str] = None


class Task(TaskBase):
"""任务响应模式"""
id: int
user_id: int

class Config:
from_attributes = True

13.6 安全功能

创建 app/core/security.py

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext

from app.core.config import settings

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


def verify_password(plain_password: str, hashed_password: str) -> bool:
"""验证密码"""
return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
"""获取密码哈希值"""
return pwd_context.hash(password)


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""创建访问令牌"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt

13.7 API 路由

1. 认证路由

创建 app/api/auth.py

from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from jose import JWTError, jwt

from app.core.config import settings
from app.core.security import verify_password, get_password_hash, create_access_token
from app.db.session import get_db
from app.models.user import User
from app.schemas.user import UserCreate, User as UserSchema, Token, TokenData

router = APIRouter()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")


def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)) -> User:
"""获取当前用户"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
token_data = TokenData(user_id=int(user_id))
except JWTError:
raise credentials_exception
user = db.query(User).filter(User.id == token_data.user_id).first()
if user is None:
raise credentials_exception
return user


@router.post("/register", response_model=UserSchema)
def register(user_in: UserCreate, db: Session = Depends(get_db)):
"""用户注册"""
# 检查用户名是否已存在
db_user = db.query(User).filter(User.username == user_in.username).first()
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered"
)
# 检查邮箱是否已存在
db_user = db.query(User).filter(User.email == user_in.email).first()
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# 创建新用户
hashed_password = get_password_hash(user_in.password)
db_user = User(
username=user_in.username,
email=user_in.email,
password_hash=hashed_password
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user


@router.post("/login", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
"""用户登录"""
# 查找用户
user = db.query(User).filter(User.email == form_data.username).first()
if not user or not verify_password(form_data.password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
# 创建访问令牌
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": str(user.id)},
expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}

2. 任务路由

创建 app/api/tasks.py

from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.models.task import Task
from app.models.user import User
from app.schemas.task import TaskCreate, TaskUpdate, Task as TaskSchema
from app.api.auth import get_current_user

router = APIRouter()


@router.post("/", response_model=TaskSchema)
def create_task(task_in: TaskCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
"""创建任务"""
db_task = Task(
title=task_in.title,
description=task_in.description,
status=task_in.status,
category=task_in.category,
user_id=current_user.id
)
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task


@router.get("/", response_model=List[TaskSchema])
def get_tasks(
category: str = None,
status: str = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""获取任务列表"""
query = db.query(Task).filter(Task.user_id == current_user.id)

if category:
query = query.filter(Task.category == category)
if status:
query = query.filter(Task.status == status)

tasks = query.all()
return tasks


@router.get("/{task_id}", response_model=TaskSchema)
def get_task(task_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
"""获取单个任务"""
task = db.query(Task).filter(Task.id == task_id, Task.user_id == current_user.id).first()
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)
return task


@router.put("/{task_id}", response_model=TaskSchema)
def update_task(
task_id: int,
task_in: TaskUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""更新任务"""
task = db.query(Task).filter(Task.id == task_id, Task.user_id == current_user.id).first()
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)

# 更新任务
update_data = task_in.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(task, field, value)

db.commit()
db.refresh(task)
return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
"""删除任务"""
task = db.query(Task).filter(Task.id == task_id, Task.user_id == current_user.id).first()
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)

db.delete(task)
db.commit()
return None

3. API 路由注册

创建 app/api/__init__.py

from fastapi import APIRouter
from app.api import auth, tasks

api_router = APIRouter()

api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(tasks.router, prefix="/tasks", tags=["tasks"])

13.8 主应用

创建 main.py

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from starlette.requests import Request

from app.core.config import settings
from app.api import api_router
from app.db import init_db

# 初始化数据库
init_db()

# 创建 FastAPI 应用
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

# 注册 API 路由
app.include_router(api_router, prefix=settings.API_V1_STR)

# 挂载静态文件
app.mount("/static", StaticFiles(directory="static"), name="static")

# 配置模板
templates = Jinja2Templates(directory="templates")

# 根路径
@app.get("/")
def read_root(request: Request):
return templates.TemplateResponse("login.html", {"request": request})

# 注册页面
@app.get("/register")
def register_page(request: Request):
return templates.TemplateResponse("register.html", {"request": request})

# 任务页面
@app.get("/tasks")
def tasks_page(request: Request):
return templates.TemplateResponse("tasks.html", {"request": request})

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

13.9 前端页面

1. 基础模板

创建 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 %}Task Manager{% endblock %}</title>
<link rel="stylesheet" href="/static/css/style.css">
</head>
<body>
<header>
<h1>Task Manager</h1>
<nav>
{% if user %}
<a href="/tasks">任务</a>
<a href="#" id="logout">退出</a>
{% else %}
<a href="/">登录</a>
<a href="/register">注册</a>
{% endif %}
</nav>
</header>
<main>
{% block content %}{% endblock %}
</main>
<script src="/static/js/script.js"></script>
</body>
</html>

2. 登录页面

创建 templates/login.html

{% extends "base.html" %}

{% block title %}登录 - Task Manager{% endblock %}

{% block content %}
<div class="container">
<h2>登录</h2>
<form id="login-form">
<div class="form-group">
<label for="email">邮箱</label>
<input type="email" id="email" name="email" required>
</div>
<div class="form-group">
<label for="password">密码</label>
<input type="password" id="password" name="password" required>
</div>
<button type="submit">登录</button>
</form>
<p class="message">还没有账号?<a href="/register">注册</a></p>
</div>
{% endblock %}

3. 注册页面

创建 templates/register.html

{% extends "base.html" %}

{% block title %}注册 - Task Manager{% endblock %}

{% block content %}
<div class="container">
<h2>注册</h2>
<form id="register-form">
<div class="form-group">
<label for="username">用户名</label>
<input type="text" id="username" name="username" required>
</div>
<div class="form-group">
<label for="email">邮箱</label>
<input type="email" id="email" name="email" required>
</div>
<div class="form-group">
<label for="password">密码</label>
<input type="password" id="password" name="password" required>
</div>
<button type="submit">注册</button>
</form>
<p class="message">已有账号?<a href="/">登录</a></p>
</div>
{% endblock %}

4. 任务页面

创建 templates/tasks.html

{% extends "base.html" %}

{% block title %}任务管理 - Task Manager{% endblock %}

{% block content %}
<div class="container">
<h2>任务管理</h2>

<!-- 创建任务表单 -->
<div class="task-form">
<h3>创建任务</h3>
<form id="create-task-form">
<div class="form-group">
<label for="title">标题</label>
<input type="text" id="title" name="title" required>
</div>
<div class="form-group">
<label for="description">描述</label>
<textarea id="description" name="description"></textarea>
</div>
<div class="form-group">
<label for="category">分类</label>
<input type="text" id="category" name="category">
</div>
<div class="form-group">
<label for="status">状态</label>
<select id="status" name="status">
<option value="todo">待办</option>
<option value="in_progress">进行中</option>
<option value="completed">已完成</option>
</select>
</div>
<button type="submit">创建任务</button>
</form>
</div>

<!-- 任务列表 -->
<div class="task-list">
<h3>任务列表</h3>
<div id="tasks-container">
<!-- 任务将通过 JavaScript 动态添加 -->
</div>
</div>
</div>
{% endblock %}

5. CSS 样式

创建 static/css/style.css

* {
box-sizing: border-box;
margin: 0;
padding: 0;
}

body {
font-family: Arial, sans-serif;
line-height: 1.6;
color: #333;
background-color: #f4f4f4;
}

header {
background-color: #333;
color: #fff;
padding: 1rem;
display: flex;
justify-content: space-between;
align-items: center;
}

nav a {
color: #fff;
text-decoration: none;
margin-left: 1rem;
}

nav a:hover {
text-decoration: underline;
}

.container {
max-width: 800px;
margin: 2rem auto;
padding: 2rem;
background-color: #fff;
border-radius: 5px;
box-shadow: 0 2px 5px rgba(0, 0, 0, 0.1);
}

h2 {
margin-bottom: 1.5rem;
color: #333;
}

h3 {
margin: 1.5rem 0 1rem;
color: #555;
}

.form-group {
margin-bottom: 1rem;
}

label {
display: block;
margin-bottom: 0.5rem;
font-weight: bold;
}

input[type="text"],
input[type="email"],
input[type="password"],
textarea,
select {
width: 100%;
padding: 0.5rem;
border: 1px solid #ddd;
border-radius: 4px;
font-size: 1rem;
}

textarea {
resize: vertical;
min-height: 100px;
}

button {
display: inline-block;
padding: 0.75rem 1.5rem;
background-color: #333;
color: #fff;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 1rem;
}

button:hover {
background-color: #555;
}

.message {
margin-top: 1rem;
text-align: center;
}

.message a {
color: #333;
text-decoration: none;
}

.message a:hover {
text-decoration: underline;
}

.task-form {
margin-bottom: 2rem;
padding-bottom: 1.5rem;
border-bottom: 1px solid #eee;
}

.task-item {
padding: 1rem;
border: 1px solid #ddd;
border-radius: 4px;
margin-bottom: 1rem;
background-color: #f9f9f9;
}

.task-item h4 {
margin-bottom: 0.5rem;
color: #333;
}

.task-item p {
margin-bottom: 0.5rem;
color: #666;
}

.task-meta {
font-size: 0.8rem;
color: #999;
margin-bottom: 1rem;
}

.task-actions {
display: flex;
gap: 0.5rem;
}

.task-actions button {
padding: 0.25rem 0.5rem;
font-size: 0.8rem;
}

.status-badge {
display: inline-block;
padding: 0.25rem 0.5rem;
border-radius: 12px;
font-size: 0.8rem;
font-weight: bold;
}

.status-todo {
background-color: #ffc107;
color: #000;
}

.status-in_progress {
background-color: #17a2b8;
color: #fff;
}

.status-completed {
background-color: #28a745;
color: #fff;
}

.error-message {
color: #dc3545;
margin-top: 0.5rem;
}

.success-message {
color: #28a745;
margin-top: 0.5rem;
}

6. JavaScript 脚本

创建 static/js/script.js

// API 基础 URL
const API_BASE_URL = "/api/v1";

// 获取本地存储的令牌
function getToken() {
return localStorage.getItem("token");
}

// 存储令牌到本地存储
function setToken(token) {
localStorage.setItem("token", token);
}

// 移除令牌
function removeToken() {
localStorage.removeItem("token");
}

// 检查是否已登录
function isLoggedIn() {
return !!getToken();
}

// 显示错误消息
function showError(element, message) {
const errorElement = document.createElement("div");
errorElement.className = "error-message";
errorElement.textContent = message;
element.appendChild(errorElement);
setTimeout(() => errorElement.remove(), 3000);
}

// 显示成功消息
function showSuccess(element, message) {
const successElement = document.createElement("div");
successElement.className = "success-message";
successElement.textContent = message;
element.appendChild(successElement);
setTimeout(() => successElement.remove(), 3000);
}

// 登录表单处理
document.addEventListener("DOMContentLoaded", function() {
const loginForm = document.getElementById("login-form");
if (loginForm) {
loginForm.addEventListener("submit", async function(e) {
e.preventDefault();

const email = document.getElementById("email").value;
const password = document.getElementById("password").value;

try {
const response = await fetch(`${API_BASE_URL}/auth/login`, {
method: "POST",
headers: {
"Content-Type": "application/x-www-form-urlencoded"
},
body: new URLSearchParams({
username: email,
password: password
})
});

if (!response.ok) {
const error = await response.json();
throw new Error(error.detail || "登录失败");
}

const data = await response.json();
setToken(data.access_token);
window.location.href = "/tasks";
} catch (error) {
showError(loginForm, error.message);
}
});
}

// 注册表单处理
const registerForm = document.getElementById("register-form");
if (registerForm) {
registerForm.addEventListener("submit", async function(e) {
e.preventDefault();

const username = document.getElementById("username").value;
const email = document.getElementById("email").value;
const password = document.getElementById("password").value;

try {
const response = await fetch(`${API_BASE_URL}/auth/register`, {
method: "POST",
headers: {
"Content-Type": "application/json"
},
body: JSON.stringify({ username, email, password })
});

if (!response.ok) {
const error = await response.json();
throw new Error(error.detail || "注册失败");
}

showSuccess(registerForm, "注册成功,请登录");
setTimeout(() => {
window.location.href = "/";
}, 2000);
} catch (error) {
showError(registerForm, error.message);
}
});
}

// 退出登录
const logoutButton = document.getElementById("logout");
if (logoutButton) {
logoutButton.addEventListener("click", function(e) {
e.preventDefault();
removeToken();
window.location.href = "/";
});
}

// 任务页面
const tasksContainer = document.getElementById("tasks-container");
const createTaskForm = document.getElementById("create-task-form");

if (tasksContainer && createTaskForm) {
// 加载任务
async function loadTasks() {
try {
const response = await fetch(`${API_BASE_URL}/tasks/`, {
headers: {
"Authorization": `Bearer ${getToken()}`
}
});

if (!response.ok) {
throw new Error("加载任务失败");
}

const tasks = await response.json();
renderTasks(tasks);
} catch (error) {
console.error(error);
}
}

// 渲染任务
function renderTasks(tasks) {
tasksContainer.innerHTML = "";

if (tasks.length === 0) {
tasksContainer.innerHTML = "<p>暂无任务</p>";
return;
}

tasks.forEach(task => {
const taskElement = document.createElement("div");
taskElement.className = "task-item";
taskElement.innerHTML = `
<h4>${task.title}</h4>
<p>${task.description || "无描述"}</p>
<div class="task-meta">
<span class="status-badge status-${task.status}">${getStatusText(task.status)}</span>
<span>${task.category || "未分类"}</span>
</div>
<div class="task-actions">
<button class="edit-btn" data-id="${task.id}">编辑</button>
<button class="delete-btn" data-id="${task.id}">删除</button>
</div>
`;
tasksContainer.appendChild(taskElement);
});

// 绑定编辑和删除事件
bindTaskActions();
}

// 获取状态文本
function getStatusText(status) {
const statusMap = {
todo: "待办",
in_progress: "进行中",
completed: "已完成"
};
return statusMap[status] || status;
}

// 绑定任务操作事件
function bindTaskActions() {
// 编辑按钮
document.querySelectorAll(".edit-btn").forEach(btn => {
btn.addEventListener("click", function() {
const taskId = this.getAttribute("data-id");
// 这里可以实现编辑功能
alert(`编辑任务 ${taskId}`);
});
});

// 删除按钮
document.querySelectorAll(".delete-btn").forEach(btn => {
btn.addEventListener("click", async function() {
const taskId = this.getAttribute("data-id");
if (confirm("确定要删除这个任务吗?")) {
try {
const response = await fetch(`${API_BASE_URL}/tasks/${taskId}`, {
method: "DELETE",
headers: {
"Authorization": `Bearer ${getToken()}`
}
});

if (!response.ok) {
throw new Error("删除任务失败");
}

loadTasks();
} catch (error) {
console.error(error);
}
}
});
});
}

// 创建任务表单处理
createTaskForm.addEventListener("submit", async function(e) {
e.preventDefault();

const title = document.getElementById("title").value;
const description = document.getElementById("description").value;
const category = document.getElementById("category").value;
const status = document.getElementById("status").value;

try {
const response = await fetch(`${API_BASE_URL}/tasks/`, {
method: "POST",
headers: {
"Content-Type": "application/json",
"Authorization": `Bearer ${getToken()}`
},
body: JSON.stringify({ title, description, category, status })
});

if (!response.ok) {
const error = await response.json();
throw new Error(error.detail || "创建任务失败");
}

showSuccess(createTaskForm, "任务创建成功");
createTaskForm.reset();
loadTasks();
} catch (error) {
showError(createTaskForm, error.message);
}
});

// 初始加载任务
loadTasks();
}

// 页面保护
if (window.location.pathname === "/tasks" && !isLoggedIn()) {
window.location.href = "/";
}
});

13.10 运行项目

1. 启动开发服务器

uvicorn main:app --reload

2. 访问项目

13.11 项目扩展

可以添加的功能:

  1. 任务优先级:为任务添加优先级属性
  2. 任务截止日期:添加任务截止日期
  3. 任务搜索:添加任务搜索功能
  4. 任务排序:按状态、创建时间等排序
  5. 任务统计:添加任务完成情况统计
  6. 用户配置:添加用户个人信息编辑
  7. 通知系统:任务状态变更通知
  8. 数据导出:导出任务为 CSV 或 JSON

技术扩展:

  1. 使用 PostgreSQL:替换 SQLite 为 PostgreSQL
  2. 添加缓存:使用 Redis 缓存
  3. 添加测试:编写单元测试和集成测试
  4. 部署:部署到 Heroku 或 AWS
  5. CI/CD:添加持续集成和部署

13.12 练习

  1. 扩展任务模型:添加任务优先级和截止日期字段
  2. 实现编辑功能:完善前端的任务编辑功能
  3. 添加任务搜索:实现任务搜索功能
  4. 添加任务统计:在任务页面添加统计信息
  5. 部署项目:尝试部署到本地或云服务器

13.13 小结

  • 本项目展示了如何使用 Python + FastAPI 创建一个完整的 Web 应用
  • 对比了 Java 和 Python 在 Web 开发中的不同实现方式
  • 演示了 Python 生态系统的强大之处,特别是 FastAPI 的简洁性和性能
  • 展示了如何使用 SQLAlchemy 进行数据库操作
  • 提供了一个完整的前端界面,实现了基本的 CRUD 操作

通过本项目的实践,你已经掌握了 Python Web 开发的基本流程和技术栈,能够将 Java 开发经验应用到 Python 项目中。