十五天Python系统学习教程第十五天

task-manager/  
├── app/  
│   ├── __init__.py          # 应用初始化  
│   ├── models.py           # 数据模型（类似Java的Entity）  
│   ├── routes/             # 路由模块（类似Controller）  
│   ├── services/           # 业务逻辑层（类似Service）  
│   ├── utils/              # 工具类（类似Java的Utils）  
│   └── extensions.py       # 扩展组件（数据库/邮件等）  
├── tests/                  # 测试套件  
├── migrations/             # 数据库迁移脚本  
├── requirements.txt        # 依赖清单  
├── Dockerfile              # 容器化配置  
└── celery_worker.py        # 异步任务处理  

# models.py  
from datetime import datetime  
from app.extensions import db  

class User(db.Model):  
    id = db.Column(db.Integer, primary_key=True)  
    username = db.Column(db.String(80), unique=True)  
    tasks = db.relationship('Task', backref='owner')  

class Task(db.Model):  
    id = db.Column(db.Integer, primary_key=True)  
    title = db.Column(db.String(200), nullable=False)  
    status = db.Column(db.Enum('todo', 'doing', 'done'))  
    priority = db.Column(db.Integer, default=3)  
    created_at = db.Column(db.DateTime, default=datetime.utcnow)  
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))  

# services/task_service.py  
from app.models import Task  

class TaskService:  
    @staticmethod  
    def create_task(user_id, title, priority=3):  
        new_task = Task(title=title, priority=priority, user_id=user_id)  
        db.session.add(new_task)  
        db.session.commit()  
        return new_task  

    @staticmethod  
    def get_user_tasks(user_id, status=None):  
        query = Task.query.filter_by(user_id=user_id)  
        if status:  
            query = query.filter_by(status=status)  
        return query.order_by(Task.priority.desc()).all()  

from flask_jwt_extended import JWTManager, create_access_token  

jwt = JWTManager()  

@jwt.user_identity_loader  
def user_identity_lookup(user):  
    return user.id  

@jwt.user_lookup_loader  
def user_lookup_callback(_jwt_header, jwt_data):  
    identity = jwt_data["sub"]  
    return User.query.get(identity)  

# 登录接口  
@app.route("/login", methods=["POST"])  
def login():  
    username = request.json.get("username")  
    user = User.query.filter_by(username=username).first()  
    access_token = create_access_token(identity=user)  
    return {"access_token": access_token}  

# celery_worker.py  
from celery import Celery  
from app.extensions import mail  

celery = Celery(__name__, broker="redis://localhost:6379/0")  

@celery.task  
def send_notification_email(recipient, message):  
    mail.send_message(  
        subject="任务提醒",  
        recipients=[recipient],  
        body=message  
    )  

# 在服务层调用  
TaskService.create_task(...)  
send_notification_email.delay(user.email, "新任务已创建")  

# Dockerfile  
FROM python:3.10-slim  

WORKDIR /app  
COPY requirements.txt .  
RUN pip install -r requirements.txt  

COPY . .  

CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:create_app()"]  

# task-manager.conf  
server {  
    listen 80;  
    location / {  
        proxy_pass http://web:5000;  
        proxy_set_header Host $host;  
    }  
    location /static {  
        alias /app/static;  
    }  
}  

# 错误方式  
def add_item(item, items=[]):  
    items.append(item)  

# 正确方式  
def add_item(item, items=None):  
    items = items or []