将框架从 Flask 迁移到 FastAPI

主要改动:
- 更新依赖:使用 fastapi、uvicorn 替代 Flask
- 数据库层:从 Flask-SQLAlchemy 迁移到纯 SQLAlchemy
- 新增 database.py 管理数据库连接和会话
- 路由层:从 Blueprint 迁移到 APIRouter,添加 Pydantic 模型验证
- 应用层:使用 FastAPI 中间件替代 Flask 插件
- 启动方式:使用 uvicorn 替代 Flask 开发服务器
- 更新 Docker 配置以支持 FastAPI

优势:
- 更高的性能和异步支持
- 自动生成 OpenAPI 文档
- 更好的类型安全和数据验证
- 所有 API 端点保持向后兼容

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-12-31 10:12:22 +00:00
parent ebd31e2716
commit 6ecd95ad5d
10 changed files with 423 additions and 352 deletions

View File

@@ -1,18 +1,19 @@
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime
from sqlalchemy import func
from werkzeug.security import generate_password_hash, check_password_hash
db = SQLAlchemy()
Base = declarative_base()
class User(db.Model):
class User(Base):
"""用户模型"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password_hash = db.Column(db.String(200), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
id = Column(Integer, primary_key=True)
username = Column(String(80), unique=True, nullable=False)
password_hash = Column(String(200), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
def set_password(self, password):
"""设置密码(哈希加密)"""
@@ -29,23 +30,23 @@ class User(db.Model):
'created_at': self.created_at.isoformat() if self.created_at else None
}
class Task(db.Model):
class Task(Base):
"""任务模型"""
__tablename__ = 'tasks'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
description = db.Column(db.Text)
polished_description = db.Column(db.Text) # AI润色后的描述
status = db.Column(db.String(20), default='pending') # pending, in_progress, completed
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
id = Column(Integer, primary_key=True)
title = Column(String(200), nullable=False)
description = Column(Text)
polished_description = Column(Text) # AI润色后的描述
status = Column(String(20), default='pending') # pending, in_progress, completed
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 关联关系
time_records = db.relationship('TimeRecord', backref='task', lazy=True, cascade='all, delete-orphan')
def to_dict(self):
return {
time_records = relationship('TimeRecord', back_populates='task', cascade='all, delete-orphan')
def to_dict(self, db_session=None):
result = {
'id': self.id,
'title': self.title,
'description': self.description,
@@ -53,27 +54,34 @@ class Task(db.Model):
'status': self.status,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
'total_time': self.get_total_time()
}
def get_total_time(self):
if db_session:
result['total_time'] = self.get_total_time(db_session)
else:
result['total_time'] = 0
return result
def get_total_time(self, db_session):
"""获取任务总时长(秒)"""
total_seconds = db.session.query(func.sum(TimeRecord.duration)).filter(
total_seconds = db_session.query(func.sum(TimeRecord.duration)).filter(
TimeRecord.task_id == self.id
).scalar() or 0
return int(total_seconds)
class TimeRecord(db.Model):
class TimeRecord(Base):
"""时间记录模型"""
__tablename__ = 'time_records'
id = db.Column(db.Integer, primary_key=True)
task_id = db.Column(db.Integer, db.ForeignKey('tasks.id'), nullable=False)
start_time = db.Column(db.DateTime, nullable=False)
end_time = db.Column(db.DateTime)
duration = db.Column(db.Integer) # 时长(秒)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
id = Column(Integer, primary_key=True)
task_id = Column(Integer, ForeignKey('tasks.id'), nullable=False)
start_time = Column(DateTime, nullable=False)
end_time = Column(DateTime)
duration = Column(Integer) # 时长(秒)
created_at = Column(DateTime, default=datetime.utcnow)
# 关联关系
task = relationship('Task', back_populates='time_records')
def to_dict(self):
return {
'id': self.id,
@@ -83,7 +91,7 @@ class TimeRecord(db.Model):
'duration': self.duration,
'created_at': self.created_at.isoformat() if self.created_at else None
}
def calculate_duration(self):
"""计算时长"""
if self.start_time and self.end_time: