為什么 Cursor 生成的代碼總不符合你的習慣？試試這幾種方法 原創
目前團隊研發已全面采用Cursor作為主力工具。雖然Claude Code和Argument Code在功能上更勝一籌，但考慮到成本因素和網絡問題，Cursor仍是我們的首選。然而我們發現，盡管使用相同的工具，團隊成員的工作效率和體驗卻存在顯著差異——關鍵在于對Cursor的掌握程度。本文將分享日常工作中的最佳實踐，幫助大家充分發揮Cursor的潛力。
cursor開發全流程
要讓 Cursor 充分發揮其潛力，達到我們的預期效果，需要系統性地結合以下四個關鍵要素：
- 精準的提示詞
- 合理且規范的開發流程
- 好用的cursorrules
- 好用的mcp工具
只有這四個方面協同優化，才能讓 Cursor 真正成為高效的生產力工具，而不僅僅是一個普通的代碼編輯器。
prompt
不得不說，Prompt Enginer是 AI 時代入門的必備技能。一個好的提示詞能讓 Cursor 更精準地理解需求，輸出更符合預期的代碼。我們通常將提示詞分為三個核心部分：
- 目標描述（What）—— 明確你要實現什么
- 上下文信息（Why & Background）—— 提供必要的背景，幫助 AI 理解場景
- 具體要求（How）—— 定義輸出格式、約束條件等
下面給出一些不同階段的提示詞示例：
項目梳理
# Objective
Please conduct an in-depth analysis of the current codebase and generate a project overview document.
# Requirements
The project review document you generate must be produced strictly in accordance with the "Project Document Organization Standards" in the project rules.
# output
Please output the project organization document and place it in the appropriate location of the project.技術方案
當我們有了需求文檔之后，我們就可讓cursor基于需求文檔、已有的代碼(如果有的話)和上一步項目梳理的代碼來進一步生成技術架構方案。
# Goal
Please generate the technical solution based on the requirements document. Note that you only need to output the detailed technical solution document. At this stage, there is no need to modify the code. At this point, the requirements document has been placed in our project in the form of a document.
# Background Knowledge
To help you generate the technical solution better, I have provided you with:
(1) Project code
(2) Requirements Document: "XX.md" s
(3) Project understanding document: "XX.md"
# Core Tasks
## 1. Document Analysis and Understanding stage
Complete the following analysis before finishing the scheme design:
- Understand the requirements in detail:
Please confirm that you have deeply understood all the requirement descriptions and functional changes mentioned in "Requirement.md".
If there are any points you don't understand or find any contradictions, please mark them immediately and submit a note.
- Understanding of code architecture:
Gain a thorough understanding of the project's documentation and the hierarchical structure of the existing code base, and determine the insertion positions for new features.
List the reusable utility classes, exception handling mechanisms and common interfaces (such as the 'utils.py' and 'ErrorCode' enumeration classes).
## 2. Scheme design stage
Please design a detailed solution based on the requirements and place the generated technical solution in the project docs directory. No code generation is required at this stage.
# Requirement
The technical solution you generate must be strictly produced in accordance with the "Technical Solution Design Document Specification" in the project rules and conform to the technical solution design document template.
# Output
Please output the technical solution and place the generated technical solution in the appropriate position of the project. There is no need to generate code.代碼落地
當技術方案確定后，我們可以讓 Cursor 生成對應的代碼。但如果代碼量較大，或者使用的 AI 模型（如免費版）能力有限，我們可以每次讓cursor實現一個功能模塊而不是一次實現所有代碼。
# Goal
Please generate the code according to the designed plan.
# Background Knowledge
To help you generate the code better, I have provided you with:
(1) Project code
(2) Requirements Document: "XX.md"
(3) Technical Solution: "XX.md"
(4) Project Understanding Document: "XX.md"
# Core Tasks
## 1. Document Analysis and Understanding stage
Complete the following analysis before starting to write the code:
- Demand matching degree check:
Gain a thorough understanding of the requirements document and the solution design document, and confirm the complete consistency between "Solution Design.MD" and "Requirement.MD" in terms of functional points, input and output, and abnormal scenarios.
If any contradiction is found, please mark it immediately and submit a note.
- Understanding of code architecture:
Gain a thorough understanding of the project's documentation and the hierarchical structure of the existing code base, and determine the insertion positions for new features.
List the reusable utility classes, exception handling mechanisms and common interfaces (such as the 'utils.py' and 'ErrorCode' enumeration classes).
2. Code generation stage
If you have clearly defined the requirements and technical solutions, please complete the code writing work.
# Requirement
1. You must follow the following core principles:
The code you generate must refer to the code style of the current project.
2. If the project already has available methods, it is necessary to consider reuse, extension of the existing methods, or method overloading to ensure the minimum granularity of changes and reduce repetitive code.
# Output
Please generate the code and place it in the appropriate position in the code base.在代碼編寫完成后，可通過AI Agent實現自動化執行：自動安裝依賴、運行測試并檢測錯誤。我們也可以手動執行代碼，當出現后端異常或前端Console/Network報錯時，把錯誤提交給cursor讓他幫忙做修復。理想的高效Agent 是可以一次生成可用的代碼，頂多和人交互兩三個輪次就能完成可用的代碼。
生成單測
當代碼可以跑通之后，我們可以繼續給代碼添加單元測試，來保證代碼的可測試性。
# Task
Please generate a test for the xx file.
# Requirement
The unit test code you generate must refer to the existing single-test method style of the current project.
# Example
Copy a written single test from your current project as an example to prompt the large model.CursorRules
目前 GitHub 上有一些開源項目分享了實用的 Cursor Rules 資源，例如 awesome-cursorrules 項目，我們可以根據自己的編程語言選擇合適的規則集。
除了使用現成的規則外，還可以讓 Cursor 根據項目特點自動生成適配的規則：
Analyze the project, output a rule developed in python, and write it into the cursorrules file of the project在cursor里，我們可以為項目創建很多個cursorrules，每個cursorrules 對應開發的不同階段。
項目文檔規范Rule
# Project Documentation Standards
**The target audience of the documentation is explicitly software developers**, with the aim of helping the development team quickly understand the system architecture, business logic, and technical implementation details, facilitating code maintenance, feature expansion, and knowledge transfer.
## Key Rules
- Project documentation must include four core sections: Project Overview, Core Domain Model, Project Structure, and External Dependencies.
- Interface documentation must be written and maintained in accordance with the @Interface Documentation Standard.
- Business process documentation must be written and maintained in accordance with the @Business Process Documentation Standard.
- Documentation should remain objective, based on existing code rather than an idealized state.
- Terminology used in the documentation must align with the terminology in the code.
- Documentation should use Markdown format, supporting diagrams, tables, and code blocks.
- Code examples must be extracted from actual code, not fabricated.
- Diagrams should use Mermaid or PlantUML syntax to ensure maintainability.
- Documentation should reference specific code file paths to help readers locate related implementations.
- First, determine whether the project uses the GBF framework, and then select the appropriate documentation structure and content based on the actual architecture.
- All documentation must be uniformly placed in the `docs` directory and use the specified Chinese names.
- **The documentation generation process must ensure complete coverage of all content, with no omissions allowed.**
## Documentation Optimization and Structure Guidelines
- **Main Index Documentation**: Create a main index document for each core section, containing links to sub-documents and brief descriptions.
- **In-Document Navigation**: Documents exceeding 500 lines must include a table of contents at the beginning.
- **Hierarchical Structure**: Organize using the "Pyramid Structure" (top level: core concepts; middle level: main functional modules; bottom level: specific implementation details).
- **Document Splitting**: Split interfaces by business domain if they exceed 20; split core entities by business domain if they exceed 10.
## Documentation Structure and Content Requirements
### 1. Project Overview - `docs/Project Overview.md`
Must include: project background, project objectives, functional overview, technology stack, and architecture type (explicitly stating whether the GBF framework is used).
### 2. Core Domain Model - `docs/Domain Model Description.md`
Must include:
- Domain model overview: definitions and boundaries of core business concepts.
- Core entity-relationship diagram: represented using E-R diagrams or class diagrams.
- Model interactions in key business scenarios.
- Data flow relationships.
**Mandatory Domain Model Scanning Rules**:
- **Comprehensive Scan**: Includes `*Entity.java`, `*DO.java`, `*PO.java`, `*Model.java`, classes annotated with `@Entity`, `@Table`, `@Document`, core models in the service layer, and DTO/VO classes.
- **Directory Structure Identification**: Java class files located in `model`, `domain`, `entity` directories and their subdirectories, as well as class files under domain-specific package paths (e.g., `*.domain.*`, `*.model.*`, `*.entity.*`).
- **Complete Extraction**: Entity attributes and business meanings, entity relationships, aggregate structures, lifecycle, and state transitions.
- **Identification Rules**: Attribute constraints, entity relationship constraints, state transition rules.
**Domain Model Analysis Strategy**:
- Scan all entity classes and value objects, supporting multiple ORM frameworks.
- Extract associations (via field types, generic parameters, and ORM annotations).
- Identify aggregate roots and boundaries (via package structure and class relationships).
- Analyze inheritance structures (including abstract classes, interfaces, and implementation classes).
- Extract business methods and state transition logic.
- Generate complete attribute tables and business rule descriptions.
**GBF Framework Project Supplement**: Extension point definitions and implementations, industry/scenario customization points, routing conditions, and dynamic selection mechanisms.
### 3. Interface Documentation - `docs/Interface Documentation.md`
Interface documentation should be created and maintained in accordance with the dedicated @Interface Documentation Standard to ensure complete recording and updating of API interfaces.
### 4. Business Processes - `docs/Business Process Description.md`
Business process documentation should be created and maintained in accordance with the dedicated @Business Process Documentation Standard to ensure complete recording and updating of business processes.
### 5. Project Structure - `docs/Project Structure Description.md`
Must include: project module division, code organization structure, key package descriptions, and layered architecture explanation.
**GBF Framework Project Supplement** - `docs/GBF Framework Application Description.md`:
GBF layered structure, extension point file locations, industry customization directories, scenario customization directories.
### 6. External Dependencies and Downstream Services - `docs/External Dependencies Description.md`
Must include:
- Downstream service overview: list and purposes of all dependent external services.
- Call relationship diagram: system interactions with external services.
## Documentation Generation Workflow
1. **Architecture Identification**: Determine project architecture type, identify key components, and layered structure.
2. **Code Analysis**: Identify core business packages and classes, analyze domain models, extract interface definitions, and understand call chains.
3. **Content Organization**: Organize information according to documentation structure, extract code examples, and create diagrams.
4. **Review and Refinement**: Verify consistency between documentation and code, supplement key information, and refine diagrams and examples.
- **Interface Coverage Verification**: Ensure all interfaces in the overview document are fully described in detailed documentation.
- **Documentation Completeness Check**: Ensure no necessary interfaces or service descriptions are omitted.
5. **Regular Updates**: Integrate with code review processes, update documentation for major changes, and conduct comprehensive reviews quarterly.
## Examples
### Domain Model Example
```markdown
## Core Entity-Relationship Diagram
```mermaid
classDiagram
class Item {
+Long id
+String name
+BigDecimal price
+String status
+validatePrice()
+changeStatus(String)
}
class TyingRule {
+Long id
+Long mainItemId
+List<Long> subItemIds
+Date startTime
+Date endTime
+enable()
+disable()
}
Item "1" -- "n" TyingRule: Defined as main product
TyingRule "1" -- "n" Item: Associated bundled products
```
## Entity Attribute Details
### Item Product Entity
| Attribute | Type | Description |
|----|---|---|
| id | Long | Unique product identifier |
| name | String | Product name, length limit: 2-50 characters |
| price | BigDecimal | Product price, precise to 2 decimal places, minimum: 0.01 |
| status | String | Product status, enum values: ON_SHELF, OFF_SHELF, DELETED |
#### Business Rules
- Product price must be greater than 0.
- Product status can only transition in a specific flow (ON_SHELF -> OFF_SHELF -> DELETED).
```
### Business Process Example
```markdown
## Bundling Rule Creation Process
### Core Flowchart
```mermaid
flowchart TD
A[Create Request] --> B{Validate Parameters}
B -->|Invalid| C[Return Error]
B -->|Valid| D[Query Main Product]
D --> E{Product Exists?}
E -->|No| F[Return Error]
E -->|Yes| G[Query Bundled Products]
G --> H{Products Exist?}
H -->|No| I[Return Error]
H -->|Yes| J[Save Rule]
J --> K[Return Success]
```
### Call Chain
**Entry Point**: `ItemTyingController.createTyingRule()`
**Call Flow**:
1. Request parameter validation - `validateTyingRequest(request)`
2. Query main product information - `itemService.getItemById()`
3. Validate main product status - `validateItemStatus(item)`
4. Query and validate bundled product list - `validateSubItems()`
5. Build and save bundling rule - `tyingRuleRepository.save()`
6. Publish rule creation event - `eventPublisher.publishEvent()`
### Key Decision Points
| Decision Point | Condition | Handling Path |
|-----|---|----|
| Parameter Validation | Main product ID is empty | Return parameter error |
| Main Product Validation | Main product does not exist | Return product not found error |
| Bundled Product Validation | Invalid products exist | Return product invalid error |
```代碼分析規則Rule
# Code Analysis Rules
## Objective
Conduct in-depth analysis of complete business processes based on code entry points to generate detailed business process documentation, facilitating team understanding and code maintenance.
## Key Rules
- **Must generate analysis documents and save them in the project's `docs` directory**
- **Must use `sequential-thinking` to assist analysis**
- **Must delve into the internal logic of methods, which may require code retrieval**
- **Recommended to use `sequential-thinking` to assist in code retrieval**
### 1. Focus on Core Business Logic
- Ignore secondary logic such as logging and basic parameter validation
- Ignore technical details in exception handling, focusing only on business exception handling logic
- Ignore utility method calls unrelated to business (e.g., string manipulation, collection operations)
- Focus on key logic such as business state transitions, process branches, and core computations
### 2. Delve into Method Call Chains
- Trace the internal implementation of each key method, not just the method call level
- Analyze the internal business logic of each important method in the call chain
- For external dependencies (e.g., HSF, RPC calls), explain their functionality and business significance
- Conduct in-depth analysis of the conditions and handling logic for each key business branch
### 3. Leverage Existing Documentation
- Prioritize using descriptions from existing documentation to avoid redundant analysis
- If a method is already documented in detail, directly reference that content
- "Stand on the shoulders of giants" by supplementing and refining based on existing documentation
- Annotate discrepancies between existing documentation and code implementation
### 4. Documentation Output Standards
- Save analysis results in the `/docs` directory using Markdown format
- Document naming format: `BusinessName-ProcessAnalysis.md` (e.g., `OrderCreation-ProcessAnalysis.md`)
- Documents must include a method call tree to clearly display call hierarchy relationships
- Use step-by-step business process descriptions to outline the complete handling process
## Documentation Structure Template
```markdown
# Business Name Process Analysis
## Functional Overview
[Briefly describe the purpose and role of this business function]
## Entry Method
`com.example.Class.method`
## Method Call Tree
Entry Method
├─ Level 1 Method 1
│ ├─ Level 2 Method 1.1
│ ├─ Level 2 Method 1.2
├─ Level 1 Method 2
├─ Level 2 Method 2.1
└─ Level 2 Method 2.2
└─ Level 3 Method
## Detailed Business Process
1. [Step 1: Business logic description]
2. [Step 2: Business logic description]
- [Sub-step 2.1: Detailed logic]
- [Sub-step 2.2: Detailed logic]
3. [Step 3: Business logic description]
## Key Business Rules
- [Rule 1: Describe the business rule and its conditions]
- [Rule 2: Describe the business rule and its conditions]
## Data Flow
- Input: [Describe method input and its business meaning]
- Processing: [Describe key data processing and transformations]
- Output: [Describe method output and its business meaning]
## Extension Points/Branch Logic
- [Branch 1: Trigger conditions and handling logic]
- [Branch 2: Trigger conditions and handling logic]
## External Dependencies
- Note dependencies on external systems
## Notes
- [List points requiring special attention in the implementation]
## System Interaction Diagram
- If the business process involves multiple system modules, use PlantUML to draw a sequence diagram
## Code Analysis Techniques
### Step 1: Identify Business Entry Points
- Determine the starting point for code analysis (typically public methods in Controller, Facade, or Service layers)
- Understand the calling context and business background of the method
### Step 2: Build Method Call Trees
- Starting from the entry method, trace all important method calls
- Use indentation to represent call hierarchy, clearly displaying call relationships
- Ignore non-core method calls (e.g., logging, parameter validation)
### Step 3: Analyze Business Processes
- Analyze business processing steps in the order of code execution
- Focus on business state transitions and branch logic
- Extract key business rules and data processing logic
### Step 4: Organize Business Rules
- Summarize implicit business rules in conditional judgments
- Analyze handling differences under various scenarios
- Extract core decision points in business logic
### Step 5: Describe Data Flow
- Analyze the source, processing, and destination of key data
- Explain data model transformations and business meanings
- Track state changes of core business entities
## Example Analysis
Refer to the [OrderQuery.md](/docs/OrderQuery.md) document for a complete analysis example:
This example demonstrates a full analysis of the order query business, including:
- Method call tree showing the complete call chain
- Detailed business process broken down step-by-step
- Clear listing of key business rules
- Explicit description of external dependencies like HSF interfaces
- Detailed explanations of special handling logic (e.g., promotional refund button exposure)
## Characteristics of Good Analysis
1. **Completeness**: Covers all core business logic and branches
2. **Hierarchical Structure**: Clearly displays the hierarchy of processing flows
3. **Business-Oriented**: Described from a business perspective, not technical implementation details
4. **Accuracy**: Precisely reflects the actual processing logic in the code
5. **Clarity**: Expressed in a way that even business personnel can understand
6. **Practicality**: Helps readers quickly grasp business processes and rules技術方案Rule
# 技術方案設計文檔規范
## 關鍵規則
- 技術方案文檔必須遵循規定的章節結構，包括名詞解釋、領域模型、應用調用關系和詳細方案設計四個主要部分
- 名詞解釋部分必須建立業務和技術的統一語言，確保術語簡單易懂
- 領域模型需清晰表達業務實體及其關系，可使用UML圖或ER圖進行可視化
- 應用調用關系必須體現跨應用的接口調用關系和消息隊列的生產消費關系
- 詳細方案設計應按應用和業務流程進行分類，對每個接口的改動點、代碼分層和文檔變更進行詳細說明
- 代碼改動點需重點突出實現思路，而不僅是羅列代碼變更
- 對外接口協議的所有變更（包括字段變更等）必須在接口文檔中明確體現
- 首先明確項目是否使用DDD框架或微服務架構，并選擇相應的技術方案設計模板
- 使用DDD架構的項目，需明確說明各層級服務規劃及領域邊界設計
- 傳統分層架構項目，應明確說明標準MVC或MVP分層架構設計
## 架構識別與方案適配
### 如何判斷Python項目架構類型
#### DDD領域驅動設計架構
- 代碼中存在Domain、Application、Infrastructure等包結構
- 存在Entity、ValueObject、DomainService、Repository等類
- 包結構中有domain、application、infrastructure、interface等目錄
- 有明確的聚合根(AggregateRoot)和領域事件(DomainEvent)機制
#### 微服務架構
- 使用FastAPI、Flask或Django作為Web框架
- 存在獨立的服務模塊和API網關
- 使用消息隊列(Redis、RabbitMQ、Kafka)進行服務間通信
- 有服務注冊發現機制
#### 傳統分層架構
- 標準的MVC或MVP模式
- Controller/View、Service、Repository/DAO層次結構
- 使用Flask、Django等傳統Web框架
### 方案適配策略
- 使用DDD架構的項目，技術方案需關注領域邊界設計和聚合設計
- 微服務架構項目，技術方案關注服務拆分和接口設計
- 傳統分層架構項目，技術方案關注分層職責和數據流轉
- 在方案開頭明確說明項目所使用的架構模式
- 根據架構特點選擇適當的設計模式和描述方式
## Python技術方案設計文檔模板
```markdown
# 技術方案設計文檔：[方案名稱]
## 文檔信息
- 作者：[作者姓名]
- 版本：[版本號，如v1.0]
- 日期：[創建/更新日期]
- 狀態：[草稿/已評審/已確認]
- 架構類型：[DDD架構/微服務架構/傳統分層架構] - Python版本：[Python版本號]
- 框架：[Flask/Django/FastAPI等] - 版本：[框架版本號]
# 一、名詞解釋
[建立業務和技術的統一語言，盡量簡單易懂]
| 術語 | 解釋 |
|---|---|
| 術語1 | 含義說明 |
| 術語2 | 含義說明 |
# 二、領域模型
[描述業務領域中的核心實體及其關系，推薦使用UML圖表示]
## 核心實體
[列出核心業務實體及其屬性、行為]
## 實體關系
[描述實體間的關系，可使用ER圖]
```mermaid
classDiagram
class User {
+int id
+str username
+str email
+datetime created_at
+create_user()
+update_profile()
}
class Order {
+int id
+int user_id
+decimal total_amount
+str status
+datetime created_at
+create_order()
+update_status()
}
User ||--o{ Order : places
```
# 三、應用調用關系
[體現跨應用的接口調用關系、消息隊列的生產消費關系]
## 系統架構圖
[系統整體架構圖，展示系統組件和交互關系]
```mermaid
flowchart TD
A[用戶服務] -->|HTTP API| B[訂單服務]
B -->|消息發送| C[Redis/RabbitMQ]
D[支付服務] -->|消息消費| C
D -->|數據存儲| E[(PostgreSQL)]
F[API網關] --> A
F --> B
F --> D
```
## 時序圖
[關鍵流程的時序圖，展示組件間的交互順序]
```mermaid
sequenceDiagram
participant Client as 客戶端
participant Gateway as API網關
participant UserService as 用戶服務
participant OrderService as 訂單服務
participant PaymentService as 支付服務
Client->>Gateway: 創建訂單請求
Gateway->>UserService: 驗證用戶信息
UserService-->>Gateway: 用戶驗證結果
Gateway->>OrderService: 創建訂單
OrderService->>PaymentService: 發起支付
PaymentService-->>OrderService: 支付結果
OrderService-->>Gateway: 訂單創建結果
Gateway-->>Client: 返回訂單信息
```
# 四、詳細方案設計
## 架構選型
[說明本方案采用的架構模式]
### 分層架構說明
[描述本方案的分層架構，說明各層職責]
#### DDD領域驅動設計架構（DDD架構項目）
```
# Interface層（接口層）
- Controllers：處理HTTP請求，參數校驗和響應封裝
- DTOs：數據傳輸對象，用于接口間數據傳遞
- Serializers：數據序列化和反序列化
# Application層（應用層）
- Services：應用服務，協調領域對象完成業務用例
- Commands/Queries：命令和查詢對象，實現CQRS模式
- Handlers：命令和查詢處理器
# Domain層（領域層）
- Entities：實體對象，包含業務邏輯和業務規則
- ValueObjects：值對象，不可變的業務概念
- DomainServices：領域服務，處理跨聚合的業務邏輯
- Repositories：倉儲接口，定義數據訪問抽象
# Infrastructure層（基礎設施層）
- Repositories：倉儲實現，具體的數據訪問實現
- ExternalServices：外部服務集成
- Configuration：配置管理
```
#### 微服務架構（微服務架構項目）
```
# API Gateway層
- 路由轉發、負載均衡、認證授權
- 限流熔斷、監控日志
# Service層（各個微服務）
- Controllers：處理HTTP請求
- Services：業務邏輯處理
- Models：數據模型
- Repositories：數據訪問
# Message Queue層
- Producers：消息生產者
- Consumers：消息消費者
- Event Handlers：事件處理器
# Infrastructure層
- Database：數據存儲
- Cache：緩存服務
- Monitoring：監控服務
```
#### 傳統分層架構（傳統分層架構項目）
```
# Presentation層（表現層）
- Controllers/Views：處理HTTP請求，參數校驗
- Templates：模板文件（如使用Django/Flask模板）
- Forms：表單處理和驗證
- Serializers：數據序列化
# Business層（業務層）
- Services：實現業務邏輯
- Managers：業務管理器，協調多個服務
- Utils：業務工具類
# Data Access層（數據訪問層）
- Models：數據模型（ORM模型）
- Repositories：數據訪問抽象
- DAOs：數據訪問對象
# Infrastructure層（基礎設施層）
- Database：數據庫配置和連接
- Cache：緩存配置
- External APIs：外部API集成
```
### 數據模型設計
[描述數據模型的設計，包括不同層次的數據模型]
```python
# DDD架構數據模型
# DTO (Data Transfer Object) - 接口層數據傳輸對象
@dataclass
class UserCreateDTO:
username: str
email: str
password: str
# Entity - 領域實體
class User(Entity):
def __init__(self, username: str, email: str, password_hash: str):
self.username = username
self.email = email
self.password_hash = password_hash
self.created_at = datetime.now()
def change_password(self, new_password: str) -> None:
# 業務規則：密碼強度驗證
if len(new_password) < 8:
raise DomainException("密碼長度不能少于8位")
self.password_hash = hash_password(new_password)
# Model - 數據庫模型（Infrastructure層）
class UserModel(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)
password_hash = Column(String(255), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
```
```python
# 傳統分層架構數據模型
# Form - 表單驗證（Flask-WTF/Django Forms）
class UserRegistrationForm(FlaskForm):
username = StringField('用戶名', validators=[DataRequired(), Length(min=3, max=20)])
email = StringField('郵箱', validators=[DataRequired(), Email()])
password = PasswordField('密碼', validators=[DataRequired(), Length(min=8)])
# Serializer - 數據序列化（Django REST Framework/Marshmallow）
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ['id', 'username', 'email', 'created_at']
# Model - ORM模型
class User(db.Model): # SQLAlchemy/Django ORM
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
```
## 應用1
### 業務流程1
#### xxx接口
**接口說明**：[詳細說明接口的用途和功能]
**接口路徑**：[HTTP方法] [路徑] 或 [RPC服務接口定義]
**請求參數**：
```json
{
"param1": "value1",
"param2": "value2"
}
```
**返回結果**：
```json
{
"code": 200,
"message": "success",
"data": {
"field1": "value1",
"field2": "value2"
}
}
```
#### 接口改動點
[說明接口的改動類型：新增、能力調整、擴展等，并詳述改動內容]
#### 代碼分層設計
[描述代碼的分層結構，確保符合工程規范]
##### DDD架構分層設計（DDD架構項目）
```python
# Interface層 - 接口控制器
# 位置：src/interface/controllers/user_controller.py
from flask import Flask, request, jsonify
from src.application.services.user_service import UserService
from src.interface.dtos.user_dto import UserCreateDTO
class UserController:
def __init__(self, user_service: UserService):
self.user_service = user_service
def create_user(self):
"""創建用戶接口"""
# 職責：參數校驗、請求處理、結果封裝
data = request.get_json()
user_dto = UserCreateDTO(**data)
result = self.user_service.create_user(user_dto)
return jsonify({"code": 200, "data": result})
# Application層 - 應用服務
# 位置：src/application/services/user_service.py
class UserService:
def __init__(self, user_repository: UserRepository):
self.user_repository = user_repository
def create_user(self, user_dto: UserCreateDTO) -> dict:
"""創建用戶業務邏輯"""
# 職責：協調領域對象，實現業務用例
# 檢查用戶名是否存在
if self.user_repository.find_by_username(user_dto.username):
raise BusinessException("用戶名已存在")
# 創建用戶實體
user = User.create(
username=user_dto.username,
email=user_dto.email,
password=user_dto.password
)
# 保存用戶
saved_user = self.user_repository.save(user)
return {"id": saved_user.id, "username": saved_user.username}
# Domain層 - 領域實體
# 位置：src/domain/entities/user.py
class User(Entity):
def __init__(self, username: str, email: str, password_hash: str):
# 職責：封裝業務規則和業務邏輯
self.username = username
self.email = email
self.password_hash = password_hash
self.created_at = datetime.now()
@classmethod
def create(cls, username: str, email: str, password: str) -> 'User':
"""創建用戶工廠方法"""
# 業務規則：用戶名長度驗證
if len(username) < 3:
raise DomainException("用戶名長度不能少于3位")
# 業務規則：郵箱格式驗證
if not cls._is_valid_email(email):
raise DomainException("郵箱格式不正確")
password_hash = hash_password(password)
return cls(username, email, password_hash)
# Infrastructure層 - 倉儲實現
# 位置：src/infrastructure/repositories/user_repository.py
class SqlAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
"""保存用戶到數據庫"""
# 職責：將領域對象轉換為數據庫模型并持久化
user_model = UserModel(
username=user.username,
email=user.email,
password_hash=user.password_hash,
created_at=user.created_at
)
self.session.add(user_model)
self.session.commit()
# 轉換回領域對象
return User(
username=user_model.username,
email=user_model.email,
password_hash=user_model.password_hash
)
```
##### 傳統分層架構設計（傳統分層架構項目）
```python
# Controller層 - Flask示例
# 位置：app/controllers/user_controller.py
from flask import Blueprint, request, jsonify
from app.services.user_service import UserService
user_bp = Blueprint('user', __name__)
@user_bp.route('/users', methods=['POST'])
def create_user():
"""創建用戶接口"""
# 職責：參數校驗、請求處理、結果封裝
data = request.get_json()
# 參數校驗
if not data.get('username') or not data.get('email'):
return jsonify({"code": 400, "message": "參數不完整"}), 400
# 調用服務層
user_service = UserService()
result = user_service.create_user(data)
return jsonify({"code": 200, "data": result})
# Service層 - 業務邏輯
# 位置：app/services/user_service.py
from app.models.user import User
from app.repositories.user_repository import UserRepository
class UserService:
def __init__(self):
self.user_repository = UserRepository()
def create_user(self, user_data: dict) -> dict:
"""創建用戶業務邏輯"""
# 職責：實現業務邏輯，協調數據訪問層
# 業務規則：檢查用戶名是否存在
if self.user_repository.find_by_username(user_data['username']):
raise BusinessException("用戶名已存在")
# 業務規則：密碼加密
user_data['password_hash'] = hash_password(user_data['password'])
del user_data['password']
# 創建用戶
user = User(**user_data)
saved_user = self.user_repository.save(user)
return {
"id": saved_user.id,
"username": saved_user.username,
"email": saved_user.email
}
# Repository層 - 數據訪問
# 位置：app/repositories/user_repository.py
from app.models.user import User
from app.extensions import db
class UserRepository:
def save(self, user: User) -> User:
"""保存用戶到數據庫"""
# 職責：封裝數據訪問邏輯
db.session.add(user)
db.session.commit()
return user
def find_by_username(self, username: str) -> User:
"""根據用戶名查找用戶"""
return User.query.filter_by(username=username).first()
# Model層 - 數據模型
# 位置：app/models/user.py
from app.extensions import db
from datetime import datetime
class User(db.Model):
"""用戶模型"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
def to_dict(self):
"""轉換為字典"""
return {
'id': self.id,
'username': self.username,
'email': self.email,
'created_at': self.created_at.isoformat()
}
```
##### 微服務架構設計（微服務架構項目）
```python
# API Gateway - 網關路由
# 位置：gateway/main.py
from fastapi import FastAPI, HTTPException
import httpx
app = FastAPI()
@app.post("/api/users")
async def create_user_gateway(user_data: dict):
"""用戶創建網關接口"""
# 職責：路由轉發、認證授權、限流
# 認證檢查
if not validate_token(request.headers.get('Authorization')):
raise HTTPException(status_code=401, detail="未授權")
# 轉發到用戶服務
async with httpx.AsyncClient() as client:
response = await client.post(
"http://user-service:8001/users",
jsnotallow=user_data
)
return response.json()
# User Service - 用戶微服務
# 位置：services/user_service/main.py
from fastapi import FastAPI, Depends
from services.user_service.models import User
from services.user_service.repositories import UserRepository
app = FastAPI()
@app.post("/users")
async def create_user(
user_data: UserCreateRequest,
user_repo: UserRepository = Depends(get_user_repository)
):
"""創建用戶"""
# 職責：處理用戶相關的業務邏輯
# 檢查用戶是否存在
existing_user = await user_repo.find_by_username(user_data.username)
if existing_user:
raise HTTPException(status_code=400, detail="用戶名已存在")
# 創建用戶
user = User(
username=user_data.username,
email=user_data.email,
password_hash=hash_password(user_data.password)
)
saved_user = await user_repo.save(user)
# 發送用戶創建事件
await publish_event("user.created", {
"user_id": saved_user.id,
"username": saved_user.username
})
return {"id": saved_user.id, "username": saved_user.username}
# Message Queue - 事件處理
# 位置：services/notification_service/consumers.py
import asyncio
from services.notification_service.email_service import EmailService
async def handle_user_created_event(event_data: dict):
"""處理用戶創建事件"""
# 職責：處理異步事件，發送歡迎郵件
email_service = EmailService()
await email_service.send_welcome_email(
user_id=event_data['user_id'],
username=event_data['username']
)
```
#### 設計模式應用（Python特有）
##### 裝飾器模式應用
```python
# 權限控制裝飾器
from functools import wraps
from flask import request, jsonify
def require_auth(permission: str):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
token = request.headers.get('Authorization')
if not validate_token_permission(token, permission):
return jsonify({"code": 401, "message": "權限不足"}), 401
return f(*args, **kwargs)
return decorated_function
return decorator
# 使用示例
@user_bp.route('/users/<int:user_id>', methods=['DELETE'])
@require_auth('user:delete')
def delete_user(user_id):
# 刪除用戶邏輯
pass
```
##### 依賴注入模式
```python
# 依賴注入容器
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject
class Container(containers.DeclarativeContainer):
# 配置
config = providers.Configuration()
# 數據庫
database = providers.Singleton(
Database,
connection_string=config.database.connection_string
)
# 倉儲
user_repository = providers.Factory(
SqlAlchemyUserRepository,
sessinotallow=database.provided.session
)
# 服務
user_service = providers.Factory(
UserService,
user_repository=user_repository
)
# 控制器中使用依賴注入
class UserController:
@inject
def __init__(self, user_service: UserService = Provide[Container.user_service]):
self.user_service = user_service
```
#### 代碼改動點
[詳述需要改動的代碼，重點說明實現思路]
1. **Controller/View層改動**：
```python
# 新增用戶管理控制器
# 位置：src/interface/controllers/user_controller.py
# 改動內容：新增用戶CRUD接口
# 實現思路：使用Flask Blueprint組織路由，通過依賴注入獲取服務實例
```
2. **Service層改動**：
```python
# 新增用戶服務
# 位置：src/application/services/user_service.py
# 改動內容：實現用戶注冊、登錄、信息更新等業務邏輯
# 實現思路：采用領域驅動設計，通過倉儲模式訪問數據
```
3. **Repository層改動**：
```python
# 新增用戶倉儲實現
# 位置：src/infrastructure/repositories/user_repository.py
# 改動內容：實現用戶數據的CRUD操作
# 實現思路：使用SQLAlchemy ORM，實現倉儲接口
```
4. **DDD特定改動（DDD架構項目）**：
```python
# 新增用戶聚合根
# 位置：src/domain/aggregates/user_aggregate.py
# 改動內容：定義用戶聚合的業務規則和不變量
# 實現思路：通過工廠方法創建實體，封裝業務規則
```
5. **微服務特定改動（微服務架構項目）**：
```python
# 新增用戶微服務
# 位置：services/user_service/
# 改動內容：獨立的用戶服務，包含完整的用戶管理功能
# 實現思路：使用FastAPI構建REST API，通過消息隊列與其他服務通信
```
## 數據庫變更
### 表結構設計
[描述需要新增或修改的數據庫表結構]
#### 表名：users
| 字段名 | 數據類型 | 是否為空 | 主鍵 | 默認值 | 注釋 |
|----|---|---|---|---|---|
| id | BIGINT | 否 | 是 | AUTO_INCREMENT | 主鍵ID |
| username | VARCHAR(50) | 否 | 否 | - | 用戶名 |
| email | VARCHAR(100) | 否 | 否 | - | 郵箱地址 |
| password_hash | VARCHAR(255) | 否 | 否 | - | 密碼哈希 |
| status | TINYINT | 否 | 否 | 1 | 用戶狀態(1:正常,0:禁用) |
| created_at | DATETIME | 否 | 否 | CURRENT_TIMESTAMP | 創建時間 |
| updated_at | DATETIME | 否 | 否 | CURRENT_TIMESTAMP ON UPDATE | 更新時間 |
### SQLAlchemy模型定義
```python
# 位置：src/infrastructure/models/user_model.py
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(50), unique=True, nullable=False, index=True)
email = Column(String(100), unique=True, nullable=False, index=True)
password_hash = Column(String(255), nullable=False)
status = Column(Boolean, default=True, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, notallow=datetime.utcnow)
```
### 數據庫遷移腳本（Alembic）
```python
# 位置：migrations/versions/001_create_users_table.py
"""create users table
Revision ID: 001
Revises:
Create Date: 2024-01-15 10:30:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers
revision = '001'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
op.create_table(
'users',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('username', sa.String(length=50), nullable=False),
sa.Column('email', sa.String(length=100), nullable=False),
sa.Column('password_hash', sa.String(length=255), nullable=False),
sa.Column('status', sa.Boolean(), nullable=False, default=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('username'),
sa.UniqueConstraint('email')
)
# 創建索引
op.create_index('idx_users_username', 'users', ['username'])
op.create_index('idx_users_email', 'users', ['email'])
op.create_index('idx_users_created_at', 'users', ['created_at'])
def downgrade():
op.drop_table('users')
```
### 索引設計
| 索引名 | 字段 | 索引類型 | 說明 |
|----|---|---|---|
| PRIMARY | id | 主鍵 | 主鍵索引 |
| uk_username | username | 唯一 | 用戶名唯一索引 |
| uk_email | email | 唯一 | 郵箱唯一索引 |
| idx_created_at | created_at | 普通 | 創建時間索引，用于排序查詢 |
| idx_status | status | 普通 | 狀態索引，用于篩選有效用戶 |
## 接口文檔變更
### API接口規范
[描述需要新增或修改的接口文檔]
#### FastAPI自動文檔生成
```python
# 位置：src/interface/schemas/user_schema.py
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime
class UserCreateRequest(BaseModel):
"""用戶創建請求"""
username: str = Field(..., min_length=3, max_length=50, descriptinotallow="用戶名")
email: EmailStr = Field(..., descriptinotallow="郵箱地址")
password: str = Field(..., min_length=8, descriptinotallow="密碼")
class UserResponse(BaseModel):
"""用戶響應"""
id: int = Field(..., descriptinotallow="用戶ID")
username: str = Field(..., descriptinotallow="用戶名")
email: str = Field(..., descriptinotallow="郵箱地址")
status: bool = Field(..., descriptinotallow="用戶狀態")
created_at: datetime = Field(..., descriptinotallow="創建時間")
class Config:
from_attributes = True
# 位置：src/interface/controllers/user_controller.py
from fastapi import APIRouter, Depends, HTTPException, status
from src.interface.schemas.user_schema import UserCreateRequest, UserResponse
router = APIRouter(prefix="/api/v1/users", tags=["用戶管理"])
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
user_data: UserCreateRequest,
user_service: UserService = Depends(get_user_service)
):
"""
創建用戶
- **username**: 用戶名，3-50個字符
- **email**: 郵箱地址，必須是有效的郵箱格式
- **password**: 密碼，至少8個字符
"""
try:
user = await user_service.create_user(user_data)
return UserResponse.from_orm(user)
except UserExistsException:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="用戶名或郵箱已存在"
)
```
### Flask-RESTX文檔生成
```python
# 位置：src/interface/controllers/user_controller.py
from flask_restx import Namespace, Resource, fields
from flask import request
api = Namespace('users', descriptinotallow='用戶管理相關操作')
# 定義請求模型
user_create_model = api.model('UserCreate', {
'username': fields.String(required=True, descriptinotallow='用戶名', min_length=3, max_length=50),
'email': fields.String(required=True, descriptinotallow='郵箱地址'),
'password': fields.String(required=True, descriptinotallow='密碼', min_length=8)
})
# 定義響應模型
user_response_model = api.model('UserResponse', {
'id': fields.Integer(descriptinotallow='用戶ID'),
'username': fields.String(descriptinotallow='用戶名'),
'email': fields.String(descriptinotallow='郵箱地址'),
'status': fields.Boolean(descriptinotallow='用戶狀態'),
'created_at': fields.DateTime(descriptinotallow='創建時間')
})
@api.route('/')
class UserListResource(Resource):
@api.expect(user_create_model)
@api.marshal_with(user_response_model, code=201)
@api.doc('create_user', descriptinotallow='創建新用戶')
def post(self):
"""創建用戶"""
data = request.get_json()
# 處理邏輯
pass
```
## 配置變更
### 應用配置
```python
# 位置：config/settings.py
import os
from typing import Optional
class Settings:
"""應用配置"""
# 應用基礎配置
APP_NAME: str = "User Management System"
APP_VERSION: str = "1.0.0"
DEBUG: bool = os.getenv("DEBUG", "False").lower() == "true"
# 數據庫配置
DATABASE_URL: str = os.getenv(
"DATABASE_URL",
"postgresql://user:password@localhost:5432/userdb"
)
# Redis配置
REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")
# JWT配置
SECRET_KEY: str = os.getenv("SECRET_KEY", "your-secret-key")
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# 郵件配置
SMTP_HOST: str = os.getenv("SMTP_HOST", "smtp.gmail.com")
SMTP_PORT: int = int(os.getenv("SMTP_PORT", "587"))
SMTP_USERNAME: str = os.getenv("SMTP_USERNAME", "")
SMTP_PASSWORD: str = os.getenv("SMTP_PASSWORD", "")
# 日志配置
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
LOG_FILE_PATH: str = os.getenv("LOG_FILE_PATH", "logs/app.log")
settings = Settings()
```
### Docker配置
```dockerfile
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安裝系統依賴
RUN apt-get update && apt-get install -y \
gcc \
postgresql-client \
&& rm -rf /var/lib/apt/lists/*
# 安裝Python依賴
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 復制應用代碼
COPY . .
# 暴露端口
EXPOSE 8000
# 啟動命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```
```yaml
# docker-compose.yml
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://postgres:password@db:5432/userdb
- REDIS_URL=redis://redis:6379/0
depends_on:
- db
- redis
volumes:
- ./logs:/app/logs
db:
image: postgres:15
environment:
- POSTGRES_DB=userdb
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
postgres_data:
```
## 測試策略
### 單元測試
```python
# 位置：tests/unit/test_user_service.py
import pytest
from unittest.mock import Mock, AsyncMock
from src.application.services.user_service import UserService
from src.domain.entities.user import User
from src.domain.exceptions import UserExistsException
class TestUserService:
@pytest.fixture
def user_repository_mock(self):
return Mock()
@pytest.fixture
def user_service(self, user_repository_mock):
return UserService(user_repository_mock)
def test_create_user_success(self, user_service, user_repository_mock):
# Arrange
user_data = {
"username": "testuser",
"email": "test@example.com",
"password": "password123"
}
user_repository_mock.find_by_username.return_value = None
user_repository_mock.save.return_value = User(**user_data)
# Act
result = user_service.create_user(user_data)
# Assert
assert result.username == "testuser"
user_repository_mock.save.assert_called_once()
def test_create_user_duplicate_username(self, user_service, user_repository_mock):
# Arrange
user_data = {
"username": "existinguser",
"email": "test@example.com",
"password": "password123"
}
user_repository_mock.find_by_username.return_value = User(**user_data)
# Act & Assert
with pytest.raises(UserExistsException):
user_service.create_user(user_data)
```
### 集成測試
```python
# 位置：tests/integration/test_user_api.py
import pytest
from fastapi.testclient import TestClient
from src.main import app
from src.infrastructure.database import get_db
from tests.fixtures.database import test_db
client = TestClient(app)
class TestUserAPI:
def test_create_user_integration(self, test_db):
# Arrange
user_data = {
"username": "integrationtest",
"email": "integration@example.com",
"password": "password123"
}
# Act
response = client.post("/api/v1/users/", jsnotallow=user_data)
# Assert
assert response.status_code == 201
data = response.json()
assert data["username"] == "integrationtest"
assert data["email"] == "integration@example.com"
assert "id" in data
```
## 非功能性需求
### 性能需求
- **響應時間**: 用戶注冊接口響應時間 < 500ms
- **并發量**: 支持1000個并發用戶注冊
- **吞吐量**: 每秒處理100個用戶注冊請求
### 可用性需求
- **系統可用率**:99.9%
- **故障恢復時間**: < 5分鐘
- **數據備份**: 每日自動備份，保留30天
### 擴展性需求
- **水平擴展**: 支持通過增加服務實例來擴展處理能力
- **數據庫分片**: 支持用戶數據按用戶ID分片存儲
- **緩存策略**: 使用Redis緩存熱點用戶數據
## 兼容性與平滑遷移方案
### 版本兼容性
- **API版本控制**: 使用URL路徑版本控制（/api/v1/, /api/v2/）
- **數據庫遷移**: 使用Alembic進行數據庫版本管理
- **向后兼容**: 新版本API保持對舊版本的兼容性
### 平滑遷移方案
```python
# 數據遷移腳本示例
# 位置：scripts/migrate_user_data.py
importasyncio
fromsrc.infrastructure.databaseimportget_db_session
fromsrc.infrastructure.models.user_modelimportUserModel
asyncdefmigrate_user_data():
"""遷移用戶數據"""
asyncwithget_db_session() assession:
# 批量處理用戶數據
batch_size = 1000
offset = 0
whileTrue:
users = awaitsession.execute(
select(UserModel).offset(offset).limit(batch_size)
)
user_list = users.scalars().all()
ifnotuser_list:
break
# 處理用戶數據遷移邏輯
foruserinuser_list:
# 數據轉換和更新
pass
awaitsession.commit()
offset += batch_size
print(f"已處理 {offset} 個用戶")
if__name__ == "__main__":
asyncio.run(migrate_user_data())
```
## 風險與應對措施
| 風險 | 可能性 | 影響 | 應對措施 |
|---|----|---|---|
| 數據庫連接池耗盡 | 中 | 高 | 配置連接池監控，實現連接池動態調整 |
| 密碼哈希算法安全性 | 低 | 高 | 使用bcrypt或Argon2等安全哈希算法 |
| 郵箱驗證服務不可用 | 中 | 中 | 實現郵件服務降級，支持多個郵件服務商 |
| 用戶數據泄露 | 低 | 高 | 實現數據加密存儲，定期安全審計 |
| 高并發下的性能問題 | 高 | 中 | 實現限流、緩存和異步處理 |
## Python特有的最佳實踐
### 代碼規范
```python
# 使用類型注解
fromtypingimportOptional, List, Dict, Any
defcreate_user(user_data:Dict[str, Any]) -> Optional[User]:
"""創建用戶，返回用戶對象或None"""
pass
# 使用dataclass簡化數據類
from dataclasses import dataclass
@dataclass
class UserDTO:
username: str
email: str
password: str
def __post_init__(self):
"""數據驗證"""
if len(self.username) < 3:
raise ValueError("用戶名長度不能少于3位")
```
### 異步編程
```python
# 異步服務實現
import asyncio
from typing import List
class AsyncUserService:
async def create_users_batch(self, users_data: List[Dict]) -> List[User]:
"""批量創建用戶"""
tasks = [self.create_user(user_data) for user_data in users_data]
return await asyncio.gather(*tasks)
async def create_user(self, user_data: Dict) -> User:
"""異步創建單個用戶"""
# 異步數據庫操作
user = await self.user_repository.save_async(user_data)
# 異步發送郵件
await self.email_service.send_welcome_email_async(user.email)
return user
```
### 上下文管理器
```python
# 數據庫事務管理
from contextlib import asynccontextmanager
@asynccontextmanager
async def database_transaction():
"""數據庫事務上下文管理器"""
session = get_db_session()
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
# 使用示例
async def create_user_with_profile(user_data: Dict, profile_data: Dict):
async with database_transaction() as session:
user = await create_user(user_data, session)
profile = await create_profile(profile_data, user.id, session)
return user, profile
```
## 示例
### DDD架構項目技術方案示例
#### 用戶管理系統 - 創建用戶功能
**架構類型**: DDD領域驅動設計架構 - Python 3.11 + FastAPI 0.104.1
##### 代碼分層設計
```python
# Interface層 - 接口控制器
# 位置: src/interface/api/user_controller.py
from fastapi import APIRouter, Depends, HTTPException, status
from src.interface.schemas.user_schema import UserCreateRequest, UserResponse
from src.application.services.user_service import UserService
router = APIRouter(prefix="/api/v1/users", tags=["用戶管理"])
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
user_request: UserCreateRequest,
user_service: UserService = Depends(get_user_service)
):
"""創建用戶接口"""
# 職責：參數校驗、請求處理、結果封裝
try:
user = await user_service.create_user(user_request)
return UserResponse.from_domain(user)
except UserAlreadyExistsException as e:
raise HTTPException(status_code=400, detail=str(e))
# Application層 - 應用服務
# 位置: src/application/services/user_service.py
from src.domain.entities.user import User
from src.domain.repositories.user_repository import UserRepository
from src.domain.services.password_service import PasswordService
class UserService:
def __init__(self, user_repo: UserRepository, password_service: PasswordService):
self.user_repo = user_repo
self.password_service = password_service
async def create_user(self, request: UserCreateRequest) -> User:
"""創建用戶應用服務"""
# 職責：協調領域對象，實現業務用例
# 檢查用戶是否已存在
existing_user = await self.user_repo.find_by_email(request.email)
if existing_user:
raise UserAlreadyExistsException("郵箱已被注冊")
# 創建用戶領域對象
user = User.create(
username=request.username,
email=request.email,
password=request.password,
password_service=self.password_service
)
# 保存用戶
return await self.user_repo.save(user)
# Domain層 - 領域實體
# 位置: src/domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime
from src.domain.value_objects.email import Email
from src.domain.value_objects.username import Username
from src.domain.services.password_service import PasswordService
@dataclass
class User:
"""用戶聚合根"""
id: Optional[int] = None
username: Username = None
email: Email = None
password_hash: str = None
created_at: datetime = None
@classmethod
def create(cls, username: str, email: str, password: str,
password_service: PasswordService) -> 'User':
"""創建用戶工廠方法"""
# 職責：封裝用戶創建的業務規則
# 驗證用戶名
username_vo = Username(username)
if not username_vo.is_valid():
raise InvalidUsernameException("用戶名格式不正確")
# 驗證郵箱
email_vo = Email(email)
if not email_vo.is_valid():
raise InvalidEmailException("郵箱格式不正確")
# 驗證密碼強度
if not password_service.is_strong_password(password):
raise WeakPasswordException("密碼強度不夠")
# 生成密碼哈希
password_hash = password_service.hash_password(password)
return cls(
username=username_vo,
email=email_vo,
password_hash=password_hash,
created_at=datetime.now()
)
def change_password(self, new_password: str, password_service: PasswordService):
"""修改密碼"""
if not password_service.is_strong_password(new_password):
raise WeakPasswordException("新密碼強度不夠")
self.password_hash = password_service.hash_password(new_password)
# Infrastructure層 - 倉儲實現
# 位置: src/infrastructure/repositories/sqlalchemy_user_repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from src.domain.repositories.user_repository import UserRepository
from src.infrastructure.models.user_model import UserModel
class SqlAlchemyUserRepository(UserRepository):
def __init__(self, session: AsyncSession):
self.session = session
async def save(self, user: User) -> User:
"""保存用戶到數據庫"""
# 職責：將領域對象轉換為數據庫模型并持久化
user_model = UserModel(
username=str(user.username),
email=str(user.email),
password_hash=user.password_hash,
created_at=user.created_at
)
self.session.add(user_model)
await self.session.commit()
await self.session.refresh(user_model)
# 轉換回領域對象
return User(
id=user_model.id,
username=Username(user_model.username),
email=Email(user_model.email),
password_hash=user_model.password_hash,
created_at=user_model.created_at
)
```
根據上述DDD架構設計，創建用戶功能的調用鏈為：
1. 客戶端調用 `POST /api/v1/users/` 接口
2. UserController 接收請求并調用 UserService.create_user()
3. UserService 協調領域對象，調用 User.create() 工廠方法
4. User實體執行業務規則驗證，創建用戶對象
5. UserService 調用 UserRepository.save() 持久化用戶
6. SqlAlchemyUserRepository 將領域對象轉換為數據庫模型并保存
7. 返回創建的用戶對象，層層返回到客戶端
### 傳統分層架構項目技術方案示例
#### 博客系統 - 文章管理功能
**架構類型**: 傳統分層架構 - Python 3.11 + Flask 2.3.3
##### 代碼分層設計
```python
# Controller層 - Flask控制器
# 位置: app/controllers/article_controller.py
from flask import Blueprint, request, jsonify
from app.services.article_service import ArticleService
from app.schemas.article_schema import ArticleCreateSchema
article_bp = Blueprint('articles', __name__, url_prefix='/api/articles')
@article_bp.route('/', methods=['POST'])
def create_article():
"""創建文章接口"""
# 職責：參數校驗、請求處理、結果封裝
data = request.get_json()
# 參數驗證
schema = ArticleCreateSchema()
try:
validated_data = schema.load(data)
except ValidationError as e:
return jsonify({"errors": e.messages}), 400
# 調用服務層
article_service = ArticleService()
try:
article = article_service.create_article(validated_data)
return jsonify(article.to_dict()), 201
except BusinessException as e:
return jsonify({"error": str(e)}), 400
# Service層 - 業務邏輯
# 位置: app/services/article_service.py
from app.models.article import Article
from app.repositories.article_repository import ArticleRepository
from app.utils.slug_generator import generate_slug
class ArticleService:
def __init__(self):
self.article_repo = ArticleRepository()
def create_article(self, article_data: dict) -> Article:
"""創建文章業務邏輯"""
# 職責：實現業務邏輯，協調數據訪問層
# 業務規則：生成文章slug
slug = generate_slug(article_data['title'])
# 業務規則：檢查slug是否重復
if self.article_repo.find_by_slug(slug):
slug = f"{slug}-{int(time.time())}"
# 業務規則：設置發布狀態
article_data['slug'] = slug
article_data['status'] = 'draft' # 默認為草稿
article_data['created_at'] = datetime.now()
# 創建文章
article = Article(**article_data)
return self.article_repo.save(article)
def publish_article(self, article_id: int) -> Article:
"""發布文章"""
article = self.article_repo.find_by_id(article_id)
if not article:
raise ArticleNotFoundException("文章不存在")
# 業務規則：只有草稿狀態的文章可以發布
if article.status != 'draft':
raise InvalidOperationException("只有草稿狀態的文章可以發布")
article.status = 'published'
article.published_at = datetime.now()
return self.article_repo.save(article)
# Repository層 - 數據訪問
# 位置: app/repositories/article_repository.py
from app.models.article import Article
from app.extensions import db
class ArticleRepository:
def save(self, article: Article) -> Article:
"""保存文章到數據庫"""
# 職責：封裝數據訪問邏輯
db.session.add(article)
db.session.commit()
return article
def find_by_id(self, article_id: int) -> Article:
"""根據ID查找文章"""
return Article.query.get(article_id)
def find_by_slug(self, slug: str) -> Article:
"""根據slug查找文章"""
return Article.query.filter_by(slug=slug).first()
def find_published_articles(self, page: int = 1, per_page: int = 10):
"""查找已發布的文章"""
return Article.query.filter_by(status='published')\
.order_by(Article.published_at.desc())\
.paginate(page=page, per_page=per_page)
# Model層 - 數據模型
# 位置: app/models/article.py
from app.extensions import db
from datetime import datetime
class Article(db.Model):
"""文章模型"""
__tablename__ = 'articles'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
slug = db.Column(db.String(200), unique=True, nullable=False)
content = db.Column(db.Text, nullable=False)
status = db.Column(db.String(20), default='draft') # draft, published, archived
author_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
published_at = db.Column(db.DateTime)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, notallow=datetime.utcnow)
def to_dict(self):
"""轉換為字典"""
return {
'id': self.id,
'title': self.title,
'slug': self.slug,
'content': self.content,
'status': self.status,
'author_id': self.author_id,
'created_at': self.created_at.isoformat() if self.created_at else None,
'published_at': self.published_at.isoformat() if self.published_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
```
根據上述傳統分層架構設計，創建文章功能的調用鏈為：
1. 客戶端調用 `POST /api/articles/` 接口
2. ArticleController 接收請求，驗證參數
3. Controller 調用 ArticleService.create_article()
4. ArticleService 處理業務邏輯（生成slug、設置狀態等）
5. Service 調用 ArticleRepository.save() 保存文章
6. Repository 使用 SQLAlchemy ORM 將文章保存到數據庫
7. 返回創建的文章對象，層層返回到客戶端
### 微服務架構項目技術方案示例
#### 電商系統 - 訂單處理功能
**架構類型**: 微服務架構 - Python 3.11 + FastAPI 0.104.1 + Docker
##### 微服務架構設計
```python
# API Gateway - 網關服務
# 位置: gateway/main.py
from fastapi import FastAPI, HTTPException, Depends
from gateway.auth import verify_token
from gateway.routing import route_to_service
import httpx
app = FastAPI(title="電商系統API網關")
@app.post("/api/orders")
async def create_order_gateway(
order_data: dict,
user_id: str = Depends(verify_token)
):
"""訂單創建網關接口"""
# 職責：路由轉發、認證授權、限流
# 添加用戶ID到請求數據
order_data['user_id'] = user_id
# 轉發到訂單服務
async with httpx.AsyncClient() as client:
response = await client.post(
"http://order-service:8001/orders",
jsnotallow=order_data,
timeout=30.0
)
if response.status_code != 201:
raise HTTPException(
status_code=response.status_code,
detail=response.json()
)
return response.json()
# Order Service - 訂單微服務
# 位置: services/order_service/main.py
from fastapi import FastAPI, HTTPException, Depends
from services.order_service.models import Order, OrderItem
from services.order_service.services import OrderService
from services.order_service.schemas import OrderCreateRequest
from services.order_service.events import publish_event
app = FastAPI(title="訂單服務")
@app.post("/orders", status_code=201)
async def create_order(
order_request: OrderCreateRequest,
order_service: OrderService = Depends(get_order_service)
):
"""創建訂單"""
# 職責：處理訂單相關的業務邏輯
try:
# 驗證商品庫存
await order_service.validate_inventory(order_request.items)
# 創建訂單
order = await order_service.create_order(order_request)
# 發布訂單創建事件
await publish_event("order.created", {
"order_id": order.id,
"user_id": order.user_id,
"total_amount": float(order.total_amount),
"items": [item.to_dict() for item in order.items]
})
return order.to_dict()
except InsufficientInventoryException as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail="訂單創建失敗")
# Inventory Service - 庫存微服務
# 位置: services/inventory_service/main.py
from fastapi import FastAPI, HTTPException
from services.inventory_service.models import Inventory
from services.inventory_service.services import InventoryService
app = FastAPI(title="庫存服務")
@app.post("/inventory/reserve")
async def reserve_inventory(
reserve_request: InventoryReserveRequest,
inventory_service: InventoryService = Depends(get_inventory_service)
):
"""預留庫存"""
# 職責：處理庫存預留邏輯
try:
reservation = await inventory_service.reserve_items(
reserve_request.items,
reserve_request.order_id
)
return {"reservation_id": reservation.id}
except InsufficientStockException as e:
raise HTTPException(status_code=400, detail=str(e))
# Payment Service - 支付微服務
# 位置: services/payment_service/consumers.py
import asyncio
from services.payment_service.services import PaymentService
from services.payment_service.events import consume_events
async def handle_order_created_event(event_data: dict):
"""處理訂單創建事件"""
# 職責：處理異步事件，創建支付記錄
payment_service = PaymentService()
# 創建支付記錄
payment = await payment_service.create_payment(
order_id=event_data['order_id'],
amount=event_data['total_amount'],
user_id=event_data['user_id']
)
# 發布支付創建事件
await publish_event("payment.created", {
"payment_id": payment.id,
"order_id": payment.order_id,
"amount": float(payment.amount)
})
# Message Queue - 事件處理
# 位置: shared/events/event_bus.py
import asyncio
import json
from typing import Dict, Callable
import aio_pika
class EventBus:
def __init__(self, rabbitmq_url: str):
self.connection = None
self.channel = None
self.handlers: Dict[str, Callable] = {}
self.rabbitmq_url = rabbitmq_url
async def connect(self):
"""連接到RabbitMQ"""
self.connection = await aio_pika.connect_robust(self.rabbitmq_url)
self.channel = await self.connection.channel()
async def publish_event(self, event_type: str, event_data: dict):
"""發布事件"""
message = aio_pika.Message(
json.dumps({
"type": event_type,
"data": event_data,
"timestamp": time.time()
}).encode()
)
await self.channel.default_exchange.publish(
message,
routing_key=event_type
)
async def subscribe(self, event_type: str, handler: Callable):
"""訂閱事件"""
queue = await self.channel.declare_queue(
f"{event_type}_queue",
durable=True
)
async def message_handler(message: aio_pika.IncomingMessage):
try:
event_data = json.loads(message.body.decode())
await handler(event_data['data'])
await message.ack()
except Exception as e:
print(f"Error handling event {event_type}: {e}")
await message.nack(requeue=True)
await queue.consume(message_handler)
# Docker Compose配置
# 位置: docker-compose.yml
version: '3.8'
services:
gateway:
build: ./gateway
ports:
- "8000:8000"
environment:
- ORDER_SERVICE_URL=http://order-service:8001
- INVENTORY_SERVICE_URL=http://inventory-service:8002
depends_on:
- order-service
- inventory-service
order-service:
build: ./services/order_service
ports:
- "8001:8001"
environment:
- DATABASE_URL=postgresql://postgres:password@postgres:5432/orders
- RABBITMQ_URL=amqp://guest:guest@rabbitmq:5672/
depends_on:
- postgres
- rabbitmq
inventory-service:
build: ./services/inventory_service
ports:
- "8002:8002"
environment:
- DATABASE_URL=postgresql://postgres:password@postgres:5432/inventory
- REDIS_URL=redis://redis:6379/0
depends_on:
- postgres
- redis
payment-service:
build: ./services/payment_service
environment:
- DATABASE_URL=postgresql://postgres:password@postgres:5432/payments
- RABBITMQ_URL=amqp://guest:guest@rabbitmq:5672/
depends_on:
- postgres
- rabbitmq
postgres:
image: postgres:15
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
rabbitmq:
image: rabbitmq:3-management
ports:
- "15672:15672"
volumes:
postgres_data:
```
根據上述微服務架構設計，創建訂單功能的調用鏈為：
1. 客戶端調用 API Gateway 的 `POST /api/orders` 接口
2. Gateway 驗證用戶身份，轉發請求到 Order Service
3. Order Service 調用 Inventory Service 驗證庫存
4. Order Service 創建訂單，發布 "order.created" 事件
5. Payment Service 消費事件，創建支付記錄
6. 各服務將數據保存到各自的數據庫
7. 返回訂單創建結果，層層返回到客戶端
## 方案設計工作流程
1. **架構識別階段**
- 確定項目使用的Python框架（Django/Flask/FastAPI）
- 識別架構模式（DDD/微服務/傳統分層）
- 確定數據庫類型（PostgreSQL/MySQL/MongoDB）
- 識別消息隊列和緩存方案
2. **需求分析階段**
- 確定功能邊界和核心業務流程
- 識別核心業務實體和領域模型
- 確定API接口設計和數據結構
- 識別異步處理需求
3. **方案設計階段**
- 根據架構特點進行分層設計
- 確定接口實現和組件交互
- 設計數據庫結構和ORM模型
- 對于微服務項目，設計服務拆分和通信機制
4. **方案評審階段**
- 驗證方案與Python最佳實踐的一致性
- 驗證功能覆蓋度和完整性
- 評估性能問題和擴展性
- 確保方案文檔結構清晰、內容完整中間件規范Rule
# 中間件使用規范
---
description: `此規則適用于go項目的單元測試開發規范，Go單元測試提供全面指南，規范測試結構、mock技術和斷言方法，確保測試代碼質量與可維護性。`
globs:
alwaysApply: false
---
# 中間件客戶端調用規范
## 關鍵規則
- **所有HTTP和HSF等中間件客戶端必須放在framework/client目錄下**
- **必須遵循統一的命名規范：服務功能+Service命名方式**
- **必須使用Context作為第一個參數，支持分布式追蹤和日志記錄**
- **中間件客戶端必須從配置中心讀取配置，不允許硬編碼**
- **中間件調用必須記錄完整的請求和響應日志**
- **必須實現統一的錯誤處理和返回機制**
- **應對關鍵調用實現緩存機制，減少直接調用次數**
- **請求和響應結構體必須實現JavaClassName方法（HSF特有）**
- **HSF服務必須在init方法中注冊模型**
- **客戶端調用需要進行合理的超時控制**
## HTTP客戶端標準實現
### 客戶端定義規范
```go
// 客戶端標準定義
type XXXHttpClient struct {
// 可選的客戶端配置
}
// 必須定義統一的初始化方法
func NewXXXHttpClient(ctx context.Context) *XXXHttpClient {
return &XXXHttpClient{}
}
```
### 請求參數定義
```go
// 請求參數必須使用結構體定義
type RequestParams struct {
// 請求字段
Field1 string `json:"field1"`
Field2 int `json:"field2"`
// ...
}
```
### 標準HTTP調用實現
```go
// 標準HTTP調用方法
func (client *XXXHttpClient) SendRequest(ctx context.Context, params RequestParams) (ResponseType, error) {
// 1. 從配置中心獲取URL配置
var conf simplehttp.URLSetting
urlConf := mconfig.UrlConfig()
if err := urlConf.UnmarshalKey("ConfigKey", &conf); err != nil || conf.URL == "" {
return nil, common.Errorf(common.ErrorInternal, "url conf miss")
}
// 2. 構建URL和請求參數
url := conf.URL + "/api/endpoint"
httpParams := map[string]string{
"param1": params.Field1,
"param2": types.IWrapper(params.Field2).String(),
// 必須加入追蹤ID
"trace_id": eagleeye.GetTraceId(ctx),
}
// 3. 構建請求選項
opt := simplehttp.BuildOptions(&simplehttp.Options{
Method: "GET", // 或 POST 等
Params: httpParams,
Timeout: conf.Timeout,
HostWithVip: conf.VipHost,
RecordWithParams: httpcommon.RecordWithParams,
})
// 4. 發送請求并記錄日志
respData, err := simplehttp.RequestWithContext(ctx, url, opt)
common.LogInfof(ctx, "log_type", "request info, params=%v, err=%v", params, err)
// 5. 錯誤處理
if err != nil {
return nil, common.Errorf(common.ErrorInternal, "request failed.err:%s", err.Error())
}
// 6. 解析響應
var response ResponseType
err = json.Unmarshal(respData, &response)
if err != nil {
return nil, common.Errorf(common.ErrorInternal, "unmarshal failed.err:%s", err.Error())
}
// 7. 返回結果
return response, nil
}
```
### 帶緩存的HTTP調用
```go
func GetDataWithCache(ctx context.Context, key string, params RequestParams) (ResponseType, error) {
var resp ResponseType
cacheKey := fmt.Sprintf("cache_key_prefix_%s", key)
// 使用緩存機制
err := mcache.GetLocalCacheFiveSecond(ctx).Once(ctx, cacheKey, &resp, 5*time.Second, func() (interface{}, error) {
// 調用實際API
data, e := SendRequest(ctx, params)
// 記錄日志
common.LogDebugf(ctx, "module_name", "GetData, key:%s, data:%+v, err:%v", key, data, e)
// 錯誤處理
if e != nil {
return nil, errors.New(e.Error())
}
return data, nil
})
if err != nil {
return nil, err
}
return resp, nil
}
```
## HSF客戶端標準實現
### 服務定義規范
```go
// 全局服務實例
var XXXService = new(XXXServiceImpl)
// 注冊HSF模型
func init() {
hsfCommon.RegisterModel(&ModelType1{})
hsfCommon.RegisterModel(&ModelType2{})
// 其他模型注冊...
}
// 服務結構體定義
type XXXServiceImpl struct {
// 方法定義，必須遵循標準方法簽名
MethodName func(ctx context.Context, args []interface{}) (*ResponseType, error)
}
// 接口名配置
func (s *XXXServiceImpl) InterfaceName() string {
return mconfig.UrlConfig().GetString("ServiceName.interfaceName")
}
// 版本配置
func (s *XXXServiceImpl) Version() string {
return mconfig.UrlConfig().GetString("ServiceName.version")
}
// 組名配置
func (s *XXXServiceImpl) Group() string {
return mconfig.UrlConfig().GetString("ServiceName.group")
}
// 超時配置
func (s *XXXServiceImpl) TimeoutMs() int {
return mconfig.UrlConfig().GetInt("ServiceName.timeout")
}
```
### 請求模型定義
```go
// 請求模型必須實現JavaClassName方法
type RequestType struct {
Field1 string `json:"field1" hessian:"field1"`
Field2 int64 `json:"field2" hessian:"field2"`
// ...
}
func (RequestType) JavaClassName() string {
return"com.package.RequestType"
}
// 響應模型必須實現JavaClassName方法
type ResponseType struct {
Code int32 `json:"code"`
Data interface{} `json:"data"`
Success bool `json:"success"`
// ...
}
func (ResponseType) JavaClassName() string {
return"com.package.ResponseType"
}
```
### 標準HSF調用實現
```go
// 標準HSF調用方法
func CallHSFService(ctx context.Context, request *RequestType) (*DataType, *common.Error) {
// 1. 調用HSF服務
hsfResp, e := XXXService.MethodName(ctx, []interface{}{request})
// 2. 記錄請求和響應日志
reqJson, _ := json.Marshal(request)
respJson, _ := json.Marshal(hsfResp)
common.LogInfof(ctx, "hsf_call", "hsf resp:%s, err:%v, req:%s",
string(respJson), e, string(reqJson))
// 3. 錯誤處理
if e != nil {
return nil, common.Errorf(common.ErrorInternal, "HSF call failed.err:%s", e.Error())
}
// 4. 結果處理
if hsfResp != nil {
result := ParseResponseData(hsfResp.Data)
return result, nil
}
return nil, nil
}
// 解析響應數據的標準方法
func ParseResponseData(data interface{}) *DataType {
if data == nil {
return nil
}
if items, ok := data.(SpecificResponseType); ok {
// 處理數據轉換
result := &DataType{
// 數據轉換邏輯
}
return result
}
return nil
}
```
### 帶緩存的HSF調用
```go
func GetHSFDataWithCache(ctx context.Context, param1, param2 string) (*DataType, error) {
var resp *DataType
cacheKey := fmt.Sprintf("hsf_cache_key_%s_%s", param1, param2)
// 使用緩存機制
err := mcache.GetLocalCacheFiveSecond(ctx).Once(ctx, cacheKey, &resp, 5*time.Second, func() (interface{}, error) {
// 構建HSF請求
request := &RequestType{
Field1: param1,
Field2: param2,
}
// 調用HSF服務
data, e := CallHSFService(ctx, request)
// 記錄日志
common.LogDebugf(ctx, "hsf_module", "GetHSFData, key:%s, data:%+v, err:%v", cacheKey, data, e)
// 錯誤處理
if e != nil {
return nil, errors.New(e.Error())
}
return data, nil
})
if err != nil {
return nil, err
}
return resp, nil
}
```
## 錯誤處理規范
- **所有中間件調用都必須返回標準化的錯誤**
- **錯誤必須包含錯誤碼和錯誤信息**
- **網絡錯誤必須歸類為InternalError**
- **參數錯誤必須歸類為InvalidError**
- **業務邏輯錯誤必須根據具體場景進行分類**
```go
// 錯誤處理示例
if err != nil {
// 網絡錯誤
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
return nil, common.Errorf(common.ErrorTimeout, "request timeout: %s", err.Error())
}
// 一般錯誤
return nil, common.Errorf(common.ErrorInternal, "request failed: %s", err.Error())
}
// 業務錯誤
if resp.Code != 200 {
return nil, common.Errorf(common.ErrorBusiness, "business error: %s", resp.Message)
}
```
## 日志記錄規范
- **所有中間件調用必須記錄請求和響應日志**
- **日志必須包含追蹤ID、請求參數和響應結果**
- **敏感信息（如密碼、token）必須在日志中脫敏**
- **必須使用統一的日志模塊和日志格式**
```go
// 標準日志示例
common.LogInfof(ctx, "module_name", "method_name, param1=%v, param2=%v, resp=%s, err=%v",
param1, param2, respJson, err)
```
## 示例
<example>
// HTTP客戶端調用示例
package client
import (
"amap-aos-activity/basic/common"
"amap-aos-activity/framework/mconfig"
"context"
"encoding/json"
"fmt"
"gitlab.alibaba-inc.com/amap-go/eagleeye-go"
"gitlab.alibaba-inc.com/amap-go/http-client/simplehttp"
httpcommon "gitlab.alibaba-inc.com/amap-go/http-client/common"
"time"
)
// 請求參數定義
type SearchParams struct {
Query string `json:"query"`
Latitude float64 `json:"latitude"`
Longitude float64 `json:"longitude"`
}
// 響應結構定義
type SearchResponse struct {
Code int `json:"code"`
Message string `json:"message"`
Data []SearchItem `json:"data"`
}
type SearchItem struct {
ID string `json:"id"`
Name string `json:"name"`
Distance float64 `json:"distance"`
}
// 發送搜索請求
func SendSearchRequest(ctx context.Context, params SearchParams) (*SearchResponse, *common.Error) {
// 從配置中獲取URL
var conf simplehttp.URLSetting
urlConf := mconfig.UrlConfig()
if err := urlConf.UnmarshalKey("SearchService", &conf); err != nil || conf.URL == "" {
return nil, common.Errorf(common.ErrorInternal, "search service url conf miss")
}
// 構建請求參數
httpParams := map[string]string{
"query": params.Query,
"latitude": fmt.Sprintf("%f", params.Latitude),
"longitude": fmt.Sprintf("%f", params.Longitude),
"trace_id": eagleeye.GetTraceId(ctx),
}
// 構建請求選項
opt := simplehttp.BuildOptions(&simplehttp.Options{
Method: "GET",
Params: httpParams,
Timeout: conf.Timeout,
HostWithVip: conf.VipHost,
RecordWithParams: httpcommon.RecordWithParams,
})
// 發送請求
url := conf.URL + "/search/api"
respData, err := simplehttp.RequestWithContext(ctx, url, opt)
common.LogInfof(ctx, "search_service", "search request, params=%v, err=%v", params, err)
// 錯誤處理
if err != nil {
return nil, common.Errorf(common.ErrorInternal, "search request failed: %s", err.Error())
}
// 解析響應
var response SearchResponse
if err := json.Unmarshal(respData, &response); err != nil {
return nil, common.Errorf(common.ErrorInternal, "unmarshal search response failed: %s", err.Error())
}
// 業務錯誤處理
if response.Code != 200 {
return nil, common.Errorf(common.ErrorBusiness, "search business error: %s", response.Message)
}
return &response, nil
}
// 帶緩存的搜索請求
func SearchWithCache(ctx context.Context, params SearchParams) (*SearchResponse, error) {
var resp *SearchResponse
cacheKey := fmt.Sprintf("search_%s_%f_%f", params.Query, params.Latitude, params.Longitude)
err := mcache.GetLocalCacheFiveSecond(ctx).Once(ctx, cacheKey, &resp, 30*time.Second, func() (interface{}, error) {
result, e := SendSearchRequest(ctx, params)
if e != nil {
return nil, e
}
return result, nil
})
if err != nil {
return nil, err
}
return resp, nil
}
</example>
<example>
// HSF客戶端調用示例
package client
import (
"amap-aos-activity/basic/common"
"amap-aos-activity/framework/mconfig"
"context"
"encoding/json"
"errors"
"fmt"
hsfCommon "gitlab.alibaba-inc.com/amap-go/hsf-go/common"
"time"
)
// 全局服務實例
var ProductService = new(ProductServiceImpl)
// 注冊HSF模型
func init() {
hsfCommon.RegisterModel(&ProductQueryRequest{})
hsfCommon.RegisterModel(&ProductQueryResponse{})
hsfCommon.RegisterModel(&ProductDetail{})
}
// 請求模型
type ProductQueryRequest struct {
ProductId string `json:"productId" hessian:"productId"`
Fields []string `json:"fields" hessian:"fields"`
}
func (ProductQueryRequest) JavaClassName() string {
return"com.example.product.request.ProductQueryRequest"
}
// 響應模型
type ProductQueryResponse struct {
Code int32 `json:"code"`
Data *ProductDetail `json:"data"`
Success bool `json:"success"`
Message string `json:"message"`
}
func (ProductQueryResponse) JavaClassName() string {
return"com.example.product.response.ProductQueryResponse"
}
// 產品詳情
type ProductDetail struct {
Id string `json:"id" hessian:"id"`
Name string `json:"name" hessian:"name"`
Price int64 `json:"price" hessian:"price"`
Description string `json:"description" hessian:"description"`
}
func (ProductDetail) JavaClassName() string {
return"com.example.product.model.ProductDetail"
}
// 服務結構體
type ProductServiceImpl struct {
QueryProduct func(ctx context.Context, args []interface{}) (*ProductQueryResponse, error)
}
// 接口配置
func (s *ProductServiceImpl) InterfaceName() string {
return mconfig.UrlConfig().GetString("ProductService.interfaceName")
}
func (s *ProductServiceImpl) Version() string {
return mconfig.UrlConfig().GetString("ProductService.version")
}
func (s *ProductServiceImpl) Group() string {
return mconfig.UrlConfig().GetString("ProductService.group")
}
func (s *ProductServiceImpl) TimeoutMs() int {
return mconfig.UrlConfig().GetInt("ProductService.timeout")
}
// 查詢產品信息
func GetProductDetail(ctx context.Context, productId string) (*ProductDetail, *common.Error) {
// 構建請求
request := &ProductQueryRequest{
ProductId: productId,
Fields: []string{"id", "name", "price", "description"},
}
// 調用HSF服務
resp, err := ProductService.QueryProduct(ctx, []interface{}{request})
// 記錄日志
reqJson, _ := json.Marshal(request)
respJson, _ := json.Marshal(resp)
common.LogInfof(ctx, "product_service", "query product, req=%s, resp=%s, err=%v",
string(reqJson), string(respJson), err)
// 錯誤處理
if err != nil {
return nil, common.Errorf(common.ErrorInternal, "query product failed: %s", err.Error())
}
// 結果處理
if resp != nil {
if !resp.Success || resp.Code != 200 {
return nil, common.Errorf(common.ErrorBusiness, "business error: %s", resp.Message)
}
return resp.Data, nil
}
return nil, common.Errorf(common.ErrorInternal, "empty response")
}
// 帶緩存的產品查詢
func GetProductWithCache(ctx context.Context, productId string) (*ProductDetail, error) {
var product *ProductDetail
cacheKey := fmt.Sprintf("product_detail_%s", productId)
err := mcache.GetLocalCacheFiveSecond(ctx).Once(ctx, cacheKey, &product, 5*time.Minute, func() (interface{}, error) {
detail, e := GetProductDetail(ctx, productId)
if e != nil {
return nil, errors.New(e.Error())
}
return detail, nil
})
if err != nil {
return nil, err
}
return product, nil
}
</example>
<example type="invalid">
// 錯誤示例：硬編碼URL和缺少日志記錄
package client
import (
"context"
"encoding/json"
"net/http"
"io/ioutil"
)
// 錯誤1: 硬編碼URL
// 錯誤2: 沒有使用配置中心
// 錯誤3: 沒有傳遞和使用context
func BadSearchRequest(query string) ([]byte, error) {
// 硬編碼URL
url := "http://search.example.com/api?query=" + query
// 沒有超時控制
resp, err := http.Get(url)
if err != nil {
return nil, err
}
defer resp.Body.Close()
// 沒有日志記錄
return ioutil.ReadAll(resp.Body)
}
// 錯誤示例：HSF調用不規范
var badHsfService = struct {
Method func(args []interface{}) (interface{}, error)
}{}
// 錯誤1: 不遵循標準接口定義
// 錯誤2: 沒有使用Context參數
// 錯誤3: 沒有注冊模型
// 錯誤4: 錯誤處理不規范
func BadHsfCall(id string) interface{} {
result, _ := badHsfService.Method([]interface{}{id})
// 忽略錯誤處理
return result
}
</example>MCP
MCP作為cursor的加強工具，可以讓工作中更加流暢，可以在MCP廣場獲取你想要的 MCP廣場，比如我們可以在日常開發中飛書用的比較多，就可以使用飛書的mcp工具，實現文檔處理、對話管理、日程安排等多種自動化場景。這個過程是增量的，當日常工作中有MCP的需求時，就可以搜索對應的MCP服務并應用到cursor中。
總結
根據團隊成員的反饋，Cursor在提升編程效率方面基本達到了預期。現在，甚至項目經理（PM）也開始使用Cursor編寫腳本，而不再像以前那樣依賴研發人員。當然，Cursor并非萬能，面對一些復雜的大型需求時，其表現有點差強人意，在這一層面不知道最近大火的Claude Code和Argument Code的表現如何？
本文轉載自???AI 博物院??? 作者：longyunfeigu
?著作權歸作者所有，如需轉載，請注明出處，否則將追究法律責任
贊
收藏
回復
分享
微博
QQ
微信
舉報
回復
相關推薦
