# lzweb **Repository Path**: wszyx0421/lzweb ## Basic Information - **Project Name**: lzweb - **Description**: No description available - **Primary Language**: Unknown - **License**: Apache-2.0 - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2025-11-13 - **Last Updated**: 2025-11-13 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README 针对当前项目的两种操作方案。 ## A方案 ## B方案 该方案是从零开始搭建一个全新的项目。 ### 2.1、预备项目 #### 2.1.1、新建仓库 在 `Gitee` 创建一个新的远程仓库，取名为 `lztbc-web` 。 将 远程仓库克隆到本地 `X:/bytecreek/mycodes` 目录下，比如: ```sh git clone git@gitee.com:azureroots/lztbc-web.git ``` #### 2.1.2、项目初始化 在 **命令提示符** 或 **终端** 中首先进入到 `lztbc-web` 本地仓库中: ```cmd X:\bytecreek\mycodes>cd lztbc-web X:\bytecreek\mycodes\lztbc-web> ``` 通过 `uv init` 来将当前的 `lztbc-web` 目录初始化为一个由 `uv` 所管理的项目。 项目初始化后会产生一些文件: - `.pytyon-version` - `main.py` - `pyproject.toml` - `README.md` ### 2.2、虚拟环境 #### 2.2.1、创建虚拟环境 首先要确保在 命令提示符 或 终端 中已经进入到 `lztbc-web` 目录中。 通过 `uv venv` 命令来创建新的虚拟环境: ```cmd X:\bytecreek\mycodes\lztbc-web>uv venv Using CPython 3.12.11 Creating virtual environment at: .venv Activate with: .venv\Scripts\activate X:\bytecreek\mycodes\lztbc-web> ``` #### 2.2.2、激活虚拟环境 在 Windows 系统的 命令提示符 中通过 以下形式激活虚拟环境: ```cmd .venv\Scripts\activate ``` 比如: ```cmd X:\bytecreek\mycodes\lztbc-web> .venv\Scripts\activate (lztbc-web) X:\bytecreek\mycodes\lztbc-web> ``` 在 Windows 的 **PowerShell** 或 Linux、macOS 的 **终端** 中通过以下形式激活虚拟环境: ```sh source .venv/bin/activate ``` > 如果想要退出激活状态，可以使用: > > ```cmd > .venv\Scripts\deactivate > ``` > > 该命令通常在 命令提示符 下使用。 ### 2.3、管理依赖 #### 2.3.1、查看依赖 查看项目的依赖关系树: ```sh uv tree ``` 查看当前虚拟环境下安装了哪些依赖: ```sh uv pip list ``` #### 2.3.2、安装依赖 向项目添加一个依赖项: ```sh uv add 依赖名称 ``` 也可以在 add 之后跟多个依赖名称从而一次安装多个依赖，比如： ```cmd (lztbc-web) X:\bytecreek\mycodes\lztbc-web>uv add fastapi uvicorn mysql-connector-python Resolved 18 packages in 1.47s Prepared 2 packages in 2m 00s Installed 16 packages in 200ms + annotated-doc==0.0.4 + annotated-types==0.7.0 + anyio==4.11.0 + click==8.3.0 + colorama==0.4.6 + fastapi==0.121.1 + h11==0.16.0 + idna==3.11 + mysql-connector-python==9.5.0 + pydantic==2.12.4 + pydantic-core==2.41.5 + sniffio==1.3.1 + starlette==0.49.3 + typing-extensions==4.15.0 + typing-inspection==0.4.2 + uvicorn==0.38.0 ``` 此处通过 `uv add ` 按装了三个依赖: - `fastapi` - `uvicorn` - `mysql-connector-python` 通过按装成功后的提示信息可知，除了安装这三个依赖外，还安装了它们所依赖的依赖。 > 如果需要从当前项目中删除一个已经按装过的依赖: > > ```cmd > uv remove 依赖名称 > ``` > > 该命令仅限当前项目。 ### 2.4、项目结构 整体项目结构图如下: ````cmd lztbc-web │ ├─.venv │ ├─app │ │ │ ├─config │ │ │ ├─controller │ │ │ ├─model │ │ │ ├─repository │ │ │ └─service │ ├─static │ │ │ ├─css │ │ │ └─js │ └─templates ```` #### 2.4.1、配置文件 在 `app/config/db_config.py` 中添加以下内容: ``` # 数据库连接配置 DB_CONFIG = { "host": "localhost", "user": "root", # 你的 MySQL 用户名 "password": "", # 你的 MySQL 密码 "database": "lanzhoutbc", "port": 3306, "charset": "utf8mb4" } ``` #### 2.4.2、模型和脚本 在 `app/model/course.sql` 中书写以下内容： ```sql use lanzhoutbc; -- 创建课程表（表名：t_courses，建议用英文小写+下划线命名规范） CREATE TABLE t_courses ( -- 编号：主键（唯一标识课程），自增整数（避免手动维护编号） id INT AUTO_INCREMENT COMMENT '课程编号（主键，唯一标识）', -- 名称：非空（课程必须有名称）， varchar(50) 足够存储课程名（如"Python 高级编程"） name VARCHAR(50) NOT NULL COMMENT '课程名称', -- 学分：非负小数（支持 0.5 学分场景），范围 0~10（根据实际需求调整） credit DECIMAL(3, 1) NOT NULL COMMENT '课程学分（0~10 之间，支持小数如 2.5）', PRIMARY KEY(id), CHECK (credit >= 0 AND credit <= 10) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='课程表'; ``` 在 `app/model/course_model.py` 中书写以下内容： ```py class Course: """课程领域模型""" def __init__(self, id=None, name=None, credit=None): """初始化Course对象""" self.id = id self.name = name self.credit = credit def validate(self): """验证课程信息的有效性""" if not self.name or self.name.strip() == "": raise ValueError("课程名称不能为空") if self.credit is None: raise ValueError("课程学分不能为空") if not isinstance(self.credit, (int, float)): raise ValueError("学分必须是数字") if not (0 <= self.credit <= 10): raise ValueError("学分必须在0到10之间") def __str__(self): return f"Course(id={self.id}, name={self.name}, credit={self.credit})" def __repr__(self): return self.__str__() ``` #### 2.4.3、数据访问层 在 `app/repository/course_repository.py` 中书写以下内容: ```python import mysql.connector from mysql.connector import Error from app.config.db_config import DB_CONFIG from app.model.course_model import Course class CourseRepository: """课程仓库类 - 负责所有与课程相关的数据库操作""" # 表名常量 TABLE_NAME = "t_courses" def __init__(self): """初始化课程仓库""" pass def get_connection(self): """获取数据库连接""" try: connection = mysql.connector.connect(**DB_CONFIG) return connection except Error as e: print(f"数据库连接失败: {e}") return None def save(self, course): """保存课程信息到数据库""" # 验证课程信息 course.validate() connection = self.get_connection() if not connection: return False try: cursor = connection.cursor() if course.id: # 更新操作 query = f"""UPDATE {self.TABLE_NAME} SET name = %s, credit = %s WHERE id = %s""" cursor.execute(query, (course.name, course.credit, course.id)) else: # 插入操作 query = f"""INSERT INTO {self.TABLE_NAME} (name, credit) VALUES (%s, %s)""" cursor.execute(query, (course.name, course.credit)) course.id = cursor.lastrowid connection.commit() return True except Error as e: print(f"保存课程失败: {e}") connection.rollback() return False finally: if connection.is_connected(): cursor.close() connection.close() def get_by_id(self, course_id): """根据ID获取课程""" connection = self.get_connection() if not connection: return None try: cursor = connection.cursor(dictionary=True) query = f"SELECT * FROM {self.TABLE_NAME} WHERE id = %s" cursor.execute(query, (course_id,)) row = cursor.fetchone() if row: return Course(**row) return None except Error as e: print(f"获取课程失败: {e}") return None finally: if connection.is_connected(): cursor.close() connection.close() def get_all(self): """获取所有课程""" connection = self.get_connection() if not connection: return [] try: cursor = connection.cursor(dictionary=True) query = f"SELECT * FROM {self.TABLE_NAME}" cursor.execute(query) rows = cursor.fetchall() return [Course(**row) for row in rows] except Error as e: print(f"获取所有课程失败: {e}") return [] finally: if connection.is_connected(): cursor.close() connection.close() def delete(self, course_id): """删除课程""" if not course_id: raise ValueError("课程ID不能为空") connection = self.get_connection() if not connection: return False try: cursor = connection.cursor() query = f"DELETE FROM {self.TABLE_NAME} WHERE id = %s" cursor.execute(query, (course_id,)) connection.commit() return cursor.rowcount > 0 except Error as e: print(f"删除课程失败: {e}") connection.rollback() return False finally: if connection.is_connected(): cursor.close() connection.close() def update(self, course_id, name=None, credit=None): """更新课程信息""" if not course_id: raise ValueError("课程ID不能为空") # 检查是否存在该课程 existing_course = self.get_by_id(course_id) if not existing_course: return False # 创建更新对象 update_course = Course( id=course_id, name=name if name is not None else existing_course.name, credit=credit if credit is not None else existing_course.credit ) # 验证并保存更新 return self.save(update_course) def search_by_name(self, keyword): """根据名称关键字搜索课程""" if not keyword: return self.get_all() connection = self.get_connection() if not connection: return [] try: cursor = connection.cursor(dictionary=True) query = f"SELECT * FROM {self.TABLE_NAME} WHERE name LIKE %s" cursor.execute(query, (f"%{keyword}%",)) rows = cursor.fetchall() return [Course(**row) for row in rows] except Error as e: print(f"搜索课程失败: {e}") return [] finally: if connection.is_connected(): cursor.close() connection.close() # 创建仓库实例供外部使用 course_repository = CourseRepository() ``` #### 2.4.4、业务逻辑层 在 `app/service/course_service.py` 中书写以下内容: ```python import logging from app.repository.course_repository import course_repository from app.model.course_model import Course # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class CourseService: """课程服务类 - 提供课程管理的业务逻辑""" def __init__(self): """初始化课程服务""" self.repository = course_repository def create_course(self, name, credit): """创建新课程""" try: logger.info(f"尝试创建课程: 名称='{name}', 学分={credit}") # 业务逻辑：检查课程名称是否已存在 existing_courses = self.repository.search_by_name(name) for course in existing_courses: if course.name == name: logger.warning(f"课程创建失败：名称 '{name}' 已存在") raise ValueError(f"课程名称 '{name}' 已存在") # 创建课程并保存 course = Course(name=name, credit=credit) success = self.repository.save(course) if success: logger.info(f"课程创建成功: ID={course.id}, 名称='{course.name}'") return course else: logger.error(f"课程保存失败: 名称='{name}'") return None except ValueError as ve: # 处理业务逻辑错误 logger.warning(f"业务逻辑错误: {str(ve)}") raise except Exception as e: # 处理其他未预期的错误 logger.error(f"创建课程时发生未预期错误: {str(e)}", exc_info=True) raise RuntimeError(f"创建课程失败: {str(e)}") def get_course_by_id(self, course_id): """根据ID获取课程""" try: logger.info(f"尝试获取课程: ID={course_id}") course = self.repository.get_by_id(course_id) if not course: logger.warning(f"课程不存在: ID={course_id}") raise ValueError(f"课程ID {course_id} 不存在") logger.info(f"成功获取课程: ID={course.id}, 名称='{course.name}'") return course except ValueError as ve: logger.warning(f"业务逻辑错误: {str(ve)}") raise except Exception as e: logger.error(f"获取课程时发生未预期错误: {str(e)}", exc_info=True) raise RuntimeError(f"获取课程失败: {str(e)}") def get_all_courses(self): """获取所有课程""" try: logger.info("尝试获取所有课程") courses = self.repository.get_all() logger.info(f"成功获取所有课程，共 {len(courses)} 条记录") return courses except Exception as e: logger.error(f"获取所有课程时发生错误: {str(e)}", exc_info=True) raise RuntimeError(f"获取所有课程失败: {str(e)}") def update_course(self, course_id, name=None, credit=None): """更新课程信息""" try: logger.info(f"尝试更新课程: ID={course_id}, 名称='{name}', 学分={credit}") # 验证课程是否存在 existing_course = self.repository.get_by_id(course_id) if not existing_course: logger.warning(f"课程更新失败：课程不存在，ID={course_id}") raise ValueError(f"课程ID {course_id} 不存在") # 如果更新名称，检查新名称是否与其他课程重复 if name and name != existing_course.name: courses = self.repository.search_by_name(name) for course in courses: if course.name == name and course.id != course_id: logger.warning(f"课程更新失败：名称 '{name}' 已存在") raise ValueError(f"课程名称 '{name}' 已存在") success = self.repository.update(course_id, name, credit) if success: logger.info(f"课程更新成功: ID={course_id}") else: logger.warning(f"课程更新失败: ID={course_id}") return success except ValueError as ve: logger.warning(f"业务逻辑错误: {str(ve)}") raise except Exception as e: logger.error(f"更新课程时发生未预期错误: {str(e)}", exc_info=True) raise RuntimeError(f"更新课程失败: {str(e)}") def delete_course(self, course_id): """删除课程""" try: logger.info(f"尝试删除课程: ID={course_id}") # 验证课程是否存在 course = self.repository.get_by_id(course_id) if not course: logger.warning(f"课程删除失败：课程不存在，ID={course_id}") raise ValueError(f"课程ID {course_id} 不存在") # 这里可以添加额外的业务逻辑，比如检查是否有学生选修了这门课程 # 例如：if self._has_enrolled_students(course_id): raise ValueError("该课程有学生选修，无法删除") success = self.repository.delete(course_id) if success: logger.info(f"课程删除成功: ID={course_id}") else: logger.warning(f"课程删除失败: ID={course_id}") return success except ValueError as ve: logger.warning(f"业务逻辑错误: {str(ve)}") raise except Exception as e: logger.error(f"删除课程时发生未预期错误: {str(e)}", exc_info=True) raise RuntimeError(f"删除课程失败: {str(e)}") def search_courses_by_name(self, keyword): """根据名称关键字搜索课程""" try: logger.info(f"尝试搜索课程: 关键字='{keyword}'") if not keyword or not keyword.strip(): logger.info("搜索关键字为空，返回所有课程") return self.repository.get_all() courses = self.repository.search_by_name(keyword.strip()) logger.info(f"搜索完成，找到 {len(courses)} 门匹配的课程") return courses except Exception as e: logger.error(f"搜索课程时发生错误: {str(e)}", exc_info=True) raise RuntimeError(f"搜索课程失败: {str(e)}") def get_courses_by_credit_range(self, min_credit, max_credit): """根据学分范围获取课程""" try: logger.info(f"尝试按学分范围获取课程: {min_credit}-{max_credit}") if min_credit > max_credit: logger.warning(f"学分范围无效: 最低学分({min_credit}) > 最高学分({max_credit})") raise ValueError("最低学分不能大于最高学分") all_courses = self.repository.get_all() filtered_courses = [course for course in all_courses if min_credit <= course.credit <= max_credit] logger.info(f"按学分范围筛选完成，找到 {len(filtered_courses)} 门课程") return filtered_courses except ValueError as ve: logger.warning(f"业务逻辑错误: {str(ve)}") raise except Exception as e: logger.error(f"按学分范围获取课程时发生错误: {str(e)}", exc_info=True) raise RuntimeError(f"获取课程失败: {str(e)}") # 创建服务实例供外部使用 course_service = CourseService() ``` #### 2.4.5、静态资源 在 项目根目录(即`pyproject.toml`所在的目录) 中创建 `static` 目录，该目录用来存放静态资源。 在 `static` 目录下创建 `css` 目录，并在其中创建 `styles.css` 文件，其中内容如下: ```css /* 全局样式重置 */ * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; line-height: 1.6; color: #333; background-color: #f4f4f4; padding: 20px; max-width: 1200px; margin: 0 auto; } /* 导航栏样式 */ nav { background-color: #35495e; color: white; padding: 15px 0; margin-bottom: 20px; border-radius: 8px; } nav ul { list-style: none; display: flex; justify-content: center; gap: 20px; flex-wrap: wrap; } nav ul li a { color: white; text-decoration: none; padding: 8px 16px; border-radius: 4px; transition: background-color 0.3s ease; } nav ul li a:hover { background-color: #42b983; } /* 容器样式 */ .container { background-color: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 10px rgba(0, 0, 0, 0.1); } /* 标题样式 */ h1, h2, h3 { color: #2c3e50; margin-bottom: 15px; } h1 { text-align: center; margin-bottom: 30px; } /* 表格样式 */ table { width: 100%; border-collapse: collapse; margin: 20px 0; } th, td { padding: 12px 15px; text-align: left; border-bottom: 1px solid #ddd; } th { background-color: #35495e; color: white; } tr:hover { background-color: #f5f5f5; } /* 按钮样式 */ .btn { display: inline-block; padding: 8px 16px; margin: 5px; border: none; border-radius: 4px; cursor: pointer; text-decoration: none; font-size: 14px; transition: background-color 0.3s ease; } .btn-primary { background-color: #42b983; color: white; } .btn-primary:hover { background-color: #3aa876; } .btn-warning { background-color: #f39c12; color: white; } .btn-warning:hover { background-color: #e67e22; } .btn-danger { background-color: #e74c3c; color: white; } .btn-danger:hover { background-color: #c0392b; } .btn-secondary { background-color: #95a5a6; color: white; } .btn-secondary:hover { background-color: #7f8c8d; } /* 表单样式 */ form { max-width: 600px; margin: 0 auto; } .form-group { margin-bottom: 15px; } label { display: block; margin-bottom: 5px; font-weight: bold; } input[type="text"], input[type="number"], textarea { width: 100%; padding: 10px; border: 1px solid #ddd; border-radius: 4px; font-size: 14px; } textarea { resize: vertical; min-height: 100px; } /* 消息提示样式 */ .message { padding: 15px; margin-bottom: 20px; border-radius: 4px; } .message.success { background-color: #d4edda; color: #155724; border: 1px solid #c3e6cb; } .message.error { background-color: #f8d7da; color: #721c24; border: 1px solid #f5c6cb; } /* 响应式设计 */ @media (max-width: 768px) { nav ul { flex-direction: column; align-items: center; } .container { padding: 15px; } table { font-size: 14px; } th, td { padding: 8px 10px; } } /* 首页样式 */ .dashboard { display: flex; flex-direction: column; align-items: center; justify-content: center; min-height: 400px; text-align: center; } .dashboard h1 { font-size: 2.5em; margin-bottom: 20px; } .dashboard p { font-size: 1.1em; margin-bottom: 30px; color: #666; } /* 操作栏样式 */ .actions { display: flex; gap: 10px; flex-wrap: wrap; } ``` #### 2.4.6、模板文件 项目中使用 `jinja2` 模板引擎技术。 所有模板文件集中存放在 `templates` 目录中。 项目根目录下的 `template_config.py` 是关于模板文件的配置文件，其中内容为: ```python from pathlib import Path from fastapi.templating import Jinja2Templates # 获取项目根目录 BASE_DIR = Path(__file__).resolve().parent.parent # 创建templates对象，指向项目根目录下的templates文件夹 templates = Jinja2Templates(directory=BASE_DIR / "templates") ``` ##### (1)、`base.html` 这里用 `base.html` 充当父模板，其中内容如下: ```jinja2 {% block title %}课程管理系统{% endblock %} {% block extra_css %}{% endblock %}
{% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% block content %}{% endblock %}
{% block extra_js %}{% endblock %} ``` ##### (2)、`index.html` 项目首页: ```jinja2 {% extends "base.html" %} {% block title %}首页 - 课程管理系统{% endblock %} {% block content %}

欢迎使用课程管理系统

这是一个简单的课程管理系统，可以帮助您管理课程信息。

查看所有课程 添加新课程
{% endblock %} ``` ##### (3)、`courses.html` 课程列表页面: ```jinja2 {% extends "base.html" %} {% block title %}课程列表 - 课程管理系统{% endblock %} {% block content %}

课程列表

{% if courses %} {% for course in courses %} {% endfor %}
ID 课程名称 学分 操作
{{ course.id }} {{ course.name }} {{ course.credit }}
查看 编辑
{% else %}

暂无课程数据。

{% endif %}
添加课程
{% endblock %} ``` ##### (4)、`create_course.html` 课程添加页面: ```jinja2 {% extends "base.html" %} {% block title %}添加课程 - 课程管理系统{% endblock %} {% block content %}

添加新课程

取消
{% endblock %} ``` ##### (5)、`edit_course.html` 课程编辑页面: ```jinja2 {% extends "base.html" %} {% block title %}编辑课程 - 课程管理系统{% endblock %} {% block content %}

编辑课程

{% if course %}
取消
{% else %}
未找到指定课程。
返回列表 {% endif %} {% endblock %} ``` ##### (6)、`course_detail.html` 课程详情页面: ```jinja2 {% extends "base.html" %} {% block title %}课程详情 - 课程管理系统{% endblock %} {% block content %}

课程详情

{% if course %}
课程ID：
{{ course.id }}
课程名称：
{{ course.name }}
学分：
{{ course.credit }}
编辑课程 返回列表
{% else %}
未找到指定课程。
返回列表 {% endif %} {% endblock %} {% block extra_css %} .course-detail { display: grid; grid-template-columns: 150px 1fr; gap: 15px; margin: 20px 0; padding: 20px; background-color: #f9f9f9; border-radius: 8px; } .course-detail dt { font-weight: bold; color: #2c3e50; } .course-detail dd { margin: 0; } {% endblock %} ``` #### 2.4.7、控制层 在 `app/controller/course_controller.py` 中书写以下内容: ```python from fastapi import APIRouter, HTTPException, Query, Request, Depends, Path as FastAPIPath, Form from fastapi.responses import JSONResponse, RedirectResponse from pathlib import Path from typing import List, Optional, Any import logging import traceback import uuid from datetime import datetime # 从main.py导入templates对象 from app.template_config import templates # 导入Pydantic模型 from pydantic import BaseModel, Field, field_validator, ValidationError # 导入服务层 try: from app.service.course_service import course_service from app.model.course_model import Course except ImportError as e: print(f"导入错误: {e}") # 确保即使导入失败也不会导致模块无法加载 course_service = None Course = None # 配置日志 - 更高级的配置 logger = logging.getLogger(__name__) # 如果logger没有处理器，则添加一个 if not logger.handlers: # 创建控制台处理器 console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # 创建格式化器 - 包含更多上下文信息 formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - ' 'RequestID: %(request_id)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) console_handler.setFormatter(formatter) # 添加处理器到logger logger.addHandler(console_handler) logger.setLevel(logging.INFO) # 创建APIRouter实例 api_router = APIRouter( prefix="/api/courses", tags=["courses"], responses={404: {"description": "Not found"}}, ) # 创建HTML路由实例 router = APIRouter() # 依赖项：生成请求ID def get_request_id(request: Request) -> str: """生成或获取请求ID""" request_id = request.headers.get("X-Request-ID") if not request_id: request_id = str(uuid.uuid4()) return request_id # 自定义日志记录装饰器 def log_operation(operation_name: str): """用于记录操作的装饰器""" def decorator(func): def wrapper(*args, **kwargs): # 获取请求和请求ID（从kwargs中） request = kwargs.get('request', None) request_id = kwargs.get('request_id', 'unknown') # 添加请求ID到日志记录的extra中 extra = {'request_id': request_id} # 记录操作开始 logger.info(f"开始 {operation_name} 操作", extra=extra) start_time = datetime.now() try: # 执行原始函数 result = func(*args, **kwargs) # 计算执行时间 execution_time = (datetime.now() - start_time).total_seconds() logger.info( f"{operation_name} 操作成功完成，耗时: {execution_time:.3f}s", extra=extra ) return result except Exception as e: # 记录操作失败 execution_time = (datetime.now() - start_time).total_seconds() logger.error( f"{operation_name} 操作失败，耗时: {execution_time:.3f}s, 错误: {str(e)}", extra=extra ) raise return wrapper return decorator # 自定义错误响应模型 class ErrorResponse(BaseModel): error_code: int message: str detail: Optional[str] = None # 检查服务是否可用的依赖项 def check_service_availability(): """检查course_service是否可用""" if course_service is None: raise HTTPException( status_code=503, detail="服务暂时不可用，请稍后再试" ) return course_service # 全局异常处理器 # 注意：异常处理器应该添加到FastAPI主应用实例中，而不是APIRouter # 此处移除了@router.exception_handler装饰器相关的代码 # 这些异常处理器应该在主应用中实现 # 请求模型 - 创建课程 class CourseCreate(BaseModel): name: str = Field(..., min_length=1, max_length=100, description="课程名称") credit: float = Field(..., ge=0.5, le=10.0, description="学分") @field_validator('credit') def validate_credit(cls, v): if v <= 0: raise ValueError('学分必须大于0') return v # 请求模型 - 更新课程 class CourseUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=100, description="课程名称") credit: Optional[float] = Field(None, ge=0.5, le=10.0, description="学分") @field_validator('credit') def validate_credit(cls, v): if v is not None and v <= 0: raise ValueError('学分必须大于0') return v # 响应模型 - 课程信息 class CourseResponse(BaseModel): id: int name: str credit: float class Config: from_attributes = True @classmethod def from_course(cls, course: Course) -> 'CourseResponse': """从Course模型转换为响应模型""" return cls( id=course.id, name=course.name, credit=course.credit ) # 创建课程 @api_router.post("/", response_model=CourseResponse, status_code=201) def create_course_api( request: Request, course_data: CourseCreate, request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """创建新课程""" extra = {'request_id': request_id} logger.info(f"收到创建课程请求: 课程名={course_data.name}", extra=extra) try: # 额外的业务逻辑验证 if len(course_data.name.strip()) == 0: raise HTTPException(status_code=400, detail="课程名称不能为空字符串") # 调用服务层创建课程 new_course = service.create_course( name=course_data.name, credit=course_data.credit ) logger.info(f"成功创建课程: ID={new_course.id}, 名称={new_course.name}", extra=extra) return CourseResponse.from_course(new_course) except HTTPException: raise except ValueError as e: logger.warning(f"创建课程失败: {str(e)}", extra=extra) raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"创建课程时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="创建课程时发生内部错误") # 获取所有课程 @api_router.get("/", response_model=List[CourseResponse]) def get_all_courses_api( request: Request, request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """获取所有课程列表""" extra = {'request_id': request_id} logger.info("收到获取所有课程请求", extra=extra) try: courses = service.get_all_courses() logger.info(f"获取课程列表成功，共 {len(courses)} 个课程", extra=extra) return [CourseResponse.from_course(course) for course in courses] except Exception as e: logger.error(f"获取课程列表时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="获取课程列表时发生内部错误") # 根据ID获取课程 @api_router.get("/{course_id}", response_model=CourseResponse) def get_course_by_id_api( request: Request, course_id: int = FastAPIPath(..., ge=1, description="课程ID"), request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """根据ID获取课程详情""" extra = {'request_id': request_id} logger.info(f"收到获取课程详情请求: 课程ID={course_id}", extra=extra) try: # 验证course_id参数 if course_id <= 0: raise HTTPException(status_code=400, detail="课程ID必须为正整数") course = service.get_course_by_id(course_id) logger.info(f"获取课程ID {course_id} 成功: 名称={course.name}", extra=extra) return CourseResponse.from_course(course) except HTTPException: raise except ValueError as e: logger.warning(f"课程ID {course_id} 不存在: {str(e)}", extra=extra) raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error(f"获取课程ID {course_id} 时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="获取课程详情时发生内部错误") # 更新课程 @api_router.put("/{course_id}", response_model=CourseResponse) def update_course_api( request: Request, course_id: int = FastAPIPath(..., ge=1, description="课程ID"), course_data: CourseUpdate = ..., request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """更新课程信息""" extra = {'request_id': request_id} logger.info(f"收到更新课程请求: 课程ID={course_id}, 更新数据={course_data.model_dump(exclude_unset=True)}", extra=extra) try: # 验证course_id参数 if course_id <= 0: raise HTTPException(status_code=400, detail="课程ID必须为正整数") # 将Pydantic模型转换为字典并过滤掉None值 update_data = course_data.model_dump(exclude_unset=True) # 如果没有提供更新数据 if not update_data: raise HTTPException(status_code=400, detail="至少需要提供一个更新字段") # 额外的业务逻辑验证 if 'name' in update_data and len(update_data['name'].strip()) == 0: raise HTTPException(status_code=400, detail="课程名称不能为空字符串") updated_course = service.update_course( course_id, **update_data ) logger.info(f"更新课程ID {course_id} 成功: 新名称={updated_course.name}", extra=extra) return CourseResponse.from_course(updated_course) except HTTPException: raise except ValueError as e: logger.warning(f"更新课程ID {course_id} 失败: {str(e)}", extra=extra) raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"更新课程ID {course_id} 时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="更新课程时发生内部错误") # 删除课程 @api_router.delete("/{course_id}", status_code=204) def delete_course_api( request: Request, course_id: int = FastAPIPath(..., ge=1, description="课程ID"), request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """删除课程""" extra = {'request_id': request_id} logger.info(f"收到删除课程请求: 课程ID={course_id}", extra=extra) try: # 验证course_id参数 if course_id <= 0: raise HTTPException(status_code=400, detail="课程ID必须为正整数") service.delete_course(course_id) logger.info(f"删除课程ID {course_id} 成功", extra=extra) return None except HTTPException: raise except ValueError as e: logger.warning(f"删除课程ID {course_id} 失败: {str(e)}", extra=extra) raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error(f"删除课程ID {course_id} 时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="删除课程时发生内部错误") # 根据名称搜索课程 @api_router.get("/search/name", response_model=List[CourseResponse]) def search_courses_by_name_api( request: Request, keyword: str = Query(..., min_length=1, max_length=100, description="搜索关键字"), request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """根据课程名称搜索课程""" extra = {'request_id': request_id} logger.info(f"收到搜索课程请求: 关键字='{keyword}'", extra=extra) try: # 额外的参数验证 if not keyword or keyword.isspace(): raise HTTPException(status_code=400, detail="搜索关键字不能为空或只包含空格") courses = service.search_courses_by_name(keyword) logger.info(f"搜索课程关键字 '{keyword}' 成功，共找到 {len(courses)} 个课程", extra=extra) return [CourseResponse.from_course(course) for course in courses] except HTTPException: raise except Exception as e: logger.error(f"搜索课程时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="搜索课程时发生内部错误") # 根据学分范围查询课程 @api_router.get("/search/credit", response_model=List[CourseResponse]) def get_courses_by_credit_range_api( request: Request, min_credit: float = Query(..., ge=0.5, le=10.0, description="最小学分"), max_credit: float = Query(..., ge=0.5, le=10.0, description="最大学分"), request_id: str = Depends(get_request_id), service: Any = Depends(check_service_availability) ): """根据学分范围查询课程""" extra = {'request_id': request_id} logger.info(f"收到按学分范围查询课程请求: 范围={min_credit}-{max_credit}", extra=extra) try: # 验证学分范围 if min_credit > max_credit: raise HTTPException(status_code=400, detail="最小学分不能大于最大学分") if min_credit < 0 or max_credit < 0: raise HTTPException(status_code=400, detail="学分必须为正数") courses = service.get_courses_by_credit_range(min_credit, max_credit) logger.info(f"查询学分范围 {min_credit}-{max_credit} 的课程成功，共找到 {len(courses)} 个课程", extra=extra) return [CourseResponse.from_course(course) for course in courses] except HTTPException: raise except Exception as e: logger.error(f"按学分范围查询课程时发生错误: {str(e)}", extra=extra) logger.error(traceback.format_exc()) raise HTTPException(status_code=500, detail="按学分范围查询课程时发生内部错误") # HTML路由函数 @router.get("/courses") def get_courses_html(request: Request): """显示所有课程的HTML页面""" try: courses = course_service.get_all_courses() return templates.TemplateResponse("courses.html", {"request": request, "courses": courses, "messages": []}) except Exception as e: return templates.TemplateResponse("courses.html", {"request": request, "courses": [], "error": str(e), "messages": []}) @router.get("/courses/create") def create_course_html(request: Request): """显示创建课程的表单页面""" return templates.TemplateResponse("create_course.html", {"request": request, "messages": []}) @router.get("/courses/{course_id}") def get_course_html(request: Request, course_id: int = FastAPIPath(..., ge=1)): """显示单个课程详情的HTML页面""" try: course = course_service.get_course_by_id(course_id) return templates.TemplateResponse("course_detail.html", {"request": request, "course": course, "messages": []}) except ValueError as e: return templates.TemplateResponse("course_detail.html", {"request": request, "course": None, "error": str(e), "messages": []}) except Exception as e: return templates.TemplateResponse("course_detail.html", {"request": request, "course": None, "error": "获取课程详情失败", "messages": []}) @router.post("/courses/create") def create_course_html(request: Request, name: str = Form(...), credit: float = Form(...)): """处理创建课程的表单提交""" try: # 直接传递参数给service层 course_service.create_course(name=name, credit=credit) return RedirectResponse(url="/courses", status_code=302) except ValueError as e: return templates.TemplateResponse("create_course.html", {"request": request, "error": str(e), "name": name, "credit": credit, "messages": []}) except Exception as e: return templates.TemplateResponse("create_course.html", {"request": request, "error": "创建课程失败", "name": name, "credit": credit, "messages": []}) @router.get("/courses/{course_id}/edit") def edit_course_html(request: Request, course_id: int = FastAPIPath(..., ge=1)): """显示编辑课程的表单页面""" try: course = course_service.get_course_by_id(course_id) return templates.TemplateResponse("edit_course.html", {"request": request, "course": course, "messages": []}) except ValueError as e: return templates.TemplateResponse("edit_course.html", {"request": request, "course": None, "error": str(e), "messages": []}) except Exception as e: return templates.TemplateResponse("edit_course.html", {"request": request, "course": None, "error": "获取课程信息失败", "messages": []}) @router.post("/courses/{course_id}/edit") def edit_course_html(request: Request, course_id: int = FastAPIPath(..., ge=1), name: str = Form(...), credit: float = Form(...)): """处理编辑课程的表单提交""" try: # 直接传递参数给service层 course_service.update_course(course_id, name=name, credit=credit) return RedirectResponse(url=f"/courses/{course_id}", status_code=302) except ValueError as e: # 获取当前课程信息以显示在表单中 try: current_course = course_service.get_course_by_id(course_id) except: current_course = None return templates.TemplateResponse("edit_course.html", {"request": request, "course": current_course, "error": str(e), "messages": []}) except Exception as e: try: current_course = course_service.get_course_by_id(course_id) except: current_course = None return templates.TemplateResponse("edit_course.html", {"request": request, "course": current_course, "error": "更新课程失败", "messages": []}) @router.post("/courses/{course_id}/delete") def delete_course_html(request: Request, course_id: int = FastAPIPath(..., ge=1)): """处理删除课程的请求""" try: course_service.delete_course(course_id) return RedirectResponse(url="/courses", status_code=303) except ValueError as e: # 删除失败后重定向回课程列表 courses = course_service.get_all_courses() if course_service else [] return templates.TemplateResponse("courses.html", {"request": request, "courses": courses, "error": str(e), "messages": []}) except Exception as e: # 记录错误但仍重定向回课程列表 courses = course_service.get_all_courses() if course_service else [] return templates.TemplateResponse("courses.html", {"request": request, "courses": courses, "error": "删除课程失败", "messages": []}) # 将API路由添加到主路由 router.include_router(api_router) ``` #### 2.4.8、项目入口 项目入口为 `app/main.py` : ```python import logging import uvicorn from fastapi import FastAPI, Request from fastapi.staticfiles import StaticFiles from pathlib import Path from app.controller.course_controller import router as course_router from app.template_config import templates # 获取项目根目录 BASE_DIR = Path(__file__).resolve().parent.parent # 初始化FastAPI应用 app = FastAPI(title="课程管理", version="1.0") # 配置静态文件目录（用于存放CSS、JavaScript等） app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static") logger = logging.getLogger(__name__) # 注册路由 app.include_router(course_router) @app.get("/") def root(request: Request): return templates.TemplateResponse("index.html", {"request": request, "messages": []}) @app.get("/health") def health_check(): return {"status": "healthy"} # 可以通过以下命令来启动项目: # uv run python -m app.main if __name__ == "__main__": uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True) ``` #### 2.4.9、启动项目 启动项目前请务必激活虚拟环境: ```cmd .venv\Scripts\activate ``` 启动项目要通过以下命令来完成: ```sh uv run python -m app.main ``` 项目启动成功后，在浏览器中输入以下地址来访问: ```http http://localhost:8000 ```