為什么 Cursor 生成的代碼總不符合你的習(xí)慣？試試這幾種方法 原創(chuàng)
目前團(tuán)隊研發(fā)已全面采用Cursor作為主力工具。雖然Claude Code和Argument Code在功能上更勝一籌，但考慮到成本因素和網(wǎng)絡(luò)問題，Cursor仍是我們的首選。然而我們發(fā)現(xiàn)，盡管使用相同的工具，團(tuán)隊成員的工作效率和體驗卻存在顯著差異——關(guān)鍵在于對Cursor的掌握程度。本文將分享日常工作中的最佳實踐，幫助大家充分發(fā)揮Cursor的潛力。
cursor開發(fā)全流程
要讓 Cursor 充分發(fā)揮其潛力，達(dá)到我們的預(yù)期效果，需要系統(tǒng)性地結(jié)合以下四個關(guān)鍵要素：
- 精準(zhǔn)的提示詞
- 合理且規(guī)范的開發(fā)流程
- 好用的cursorrules
- 好用的mcp工具
只有這四個方面協(xié)同優(yōu)化，才能讓 Cursor 真正成為高效的生產(chǎn)力工具，而不僅僅是一個普通的代碼編輯器。
prompt
不得不說，Prompt Enginer是 AI 時代入門的必備技能。一個好的提示詞能讓 Cursor 更精準(zhǔn)地理解需求，輸出更符合預(yù)期的代碼。我們通常將提示詞分為三個核心部分：
- 目標(biāo)描述（What）—— 明確你要實現(xiàn)什么
- 上下文信息（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.技術(shù)方案
當(dāng)我們有了需求文檔之后，我們就可讓cursor基于需求文檔、已有的代碼(如果有的話)和上一步項目梳理的代碼來進(jìn)一步生成技術(shù)架構(gòu)方案。
# 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.代碼落地
當(dāng)技術(shù)方案確定后，我們可以讓 Cursor 生成對應(yīng)的代碼。但如果代碼量較大，或者使用的 AI 模型（如免費版）能力有限，我們可以每次讓cursor實現(xiàn)一個功能模塊而不是一次實現(xiàn)所有代碼。
# 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實現(xiàn)自動化執(zhí)行：自動安裝依賴、運行測試并檢測錯誤。我們也可以手動執(zhí)行代碼，當(dāng)出現(xiàn)后端異常或前端Console/Network報錯時，把錯誤提交給cursor讓他幫忙做修復(fù)。理想的高效Agent 是可以一次生成可用的代碼，頂多和人交互兩三個輪次就能完成可用的代碼。
生成單測
當(dāng)代碼可以跑通之后，我們可以繼續(xù)給代碼添加單元測試，來保證代碼的可測試性。
# 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 項目，我們可以根據(jù)自己的編程語言選擇合適的規(guī)則集。
除了使用現(xiàn)成的規(guī)則外，還可以讓 Cursor 根據(jù)項目特點自動生成適配的規(guī)則：
Analyze the project, output a rule developed in python, and write it into the cursorrules file of the project在cursor里，我們可以為項目創(chuàng)建很多個cursorrules，每個cursorrules 對應(yīng)開發(fā)的不同階段。
項目文檔規(guī)范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 |
```代碼分析規(guī)則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技術(shù)方案Rule
# 技術(shù)方案設(shè)計文檔規(guī)范
## 關(guān)鍵規(guī)則
- 技術(shù)方案文檔必須遵循規(guī)定的章節(jié)結(jié)構(gòu)，包括名詞解釋、領(lǐng)域模型、應(yīng)用調(diào)用關(guān)系和詳細(xì)方案設(shè)計四個主要部分
- 名詞解釋部分必須建立業(yè)務(wù)和技術(shù)的統(tǒng)一語言，確保術(shù)語簡單易懂
- 領(lǐng)域模型需清晰表達(dá)業(yè)務(wù)實體及其關(guān)系，可使用UML圖或ER圖進(jìn)行可視化
- 應(yīng)用調(diào)用關(guān)系必須體現(xiàn)跨應(yīng)用的接口調(diào)用關(guān)系和消息隊列的生產(chǎn)消費關(guān)系
- 詳細(xì)方案設(shè)計應(yīng)按應(yīng)用和業(yè)務(wù)流程進(jìn)行分類，對每個接口的改動點、代碼分層和文檔變更進(jìn)行詳細(xì)說明
- 代碼改動點需重點突出實現(xiàn)思路，而不僅是羅列代碼變更
- 對外接口協(xié)議的所有變更（包括字段變更等）必須在接口文檔中明確體現(xiàn)
- 首先明確項目是否使用DDD框架或微服務(wù)架構(gòu)，并選擇相應(yīng)的技術(shù)方案設(shè)計模板
- 使用DDD架構(gòu)的項目，需明確說明各層級服務(wù)規(guī)劃及領(lǐng)域邊界設(shè)計
- 傳統(tǒng)分層架構(gòu)項目，應(yīng)明確說明標(biāo)準(zhǔn)MVC或MVP分層架構(gòu)設(shè)計
## 架構(gòu)識別與方案適配
### 如何判斷Python項目架構(gòu)類型
#### DDD領(lǐng)域驅(qū)動設(shè)計架構(gòu)
- 代碼中存在Domain、Application、Infrastructure等包結(jié)構(gòu)
- 存在Entity、ValueObject、DomainService、Repository等類
- 包結(jié)構(gòu)中有domain、application、infrastructure、interface等目錄
- 有明確的聚合根(AggregateRoot)和領(lǐng)域事件(DomainEvent)機(jī)制
#### 微服務(wù)架構(gòu)
- 使用FastAPI、Flask或Django作為Web框架
- 存在獨立的服務(wù)模塊和API網(wǎng)關(guān)
- 使用消息隊列(Redis、RabbitMQ、Kafka)進(jìn)行服務(wù)間通信
- 有服務(wù)注冊發(fā)現(xiàn)機(jī)制
#### 傳統(tǒng)分層架構(gòu)
- 標(biāo)準(zhǔn)的MVC或MVP模式
- Controller/View、Service、Repository/DAO層次結(jié)構(gòu)
- 使用Flask、Django等傳統(tǒng)Web框架
### 方案適配策略
- 使用DDD架構(gòu)的項目，技術(shù)方案需關(guān)注領(lǐng)域邊界設(shè)計和聚合設(shè)計
- 微服務(wù)架構(gòu)項目，技術(shù)方案關(guān)注服務(wù)拆分和接口設(shè)計
- 傳統(tǒng)分層架構(gòu)項目，技術(shù)方案關(guān)注分層職責(zé)和數(shù)據(jù)流轉(zhuǎn)
- 在方案開頭明確說明項目所使用的架構(gòu)模式
- 根據(jù)架構(gòu)特點選擇適當(dāng)?shù)脑O(shè)計模式和描述方式
## Python技術(shù)方案設(shè)計文檔模板
```markdown
# 技術(shù)方案設(shè)計文檔：[方案名稱]
## 文檔信息
- 作者：[作者姓名]
- 版本：[版本號，如v1.0]
- 日期：[創(chuàng)建/更新日期]
- 狀態(tài)：[草稿/已評審/已確認(rèn)]
- 架構(gòu)類型：[DDD架構(gòu)/微服務(wù)架構(gòu)/傳統(tǒng)分層架構(gòu)] - Python版本：[Python版本號]
- 框架：[Flask/Django/FastAPI等] - 版本：[框架版本號]
# 一、名詞解釋
[建立業(yè)務(wù)和技術(shù)的統(tǒng)一語言，盡量簡單易懂]
| 術(shù)語 | 解釋 |
|---|---|
| 術(shù)語1 | 含義說明 |
| 術(shù)語2 | 含義說明 |
# 二、領(lǐng)域模型
[描述業(yè)務(wù)領(lǐng)域中的核心實體及其關(guān)系，推薦使用UML圖表示]
## 核心實體
[列出核心業(yè)務(wù)實體及其屬性、行為]
## 實體關(guān)系
[描述實體間的關(guān)系，可使用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
```
# 三、應(yīng)用調(diào)用關(guān)系
[體現(xiàn)跨應(yīng)用的接口調(diào)用關(guān)系、消息隊列的生產(chǎn)消費關(guān)系]
## 系統(tǒng)架構(gòu)圖
[系統(tǒng)整體架構(gòu)圖，展示系統(tǒng)組件和交互關(guān)系]
```mermaid
flowchart TD
A[用戶服務(wù)] -->|HTTP API| B[訂單服務(wù)]
B -->|消息發(fā)送| C[Redis/RabbitMQ]
D[支付服務(wù)] -->|消息消費| C
D -->|數(shù)據(jù)存儲| E[(PostgreSQL)]
F[API網(wǎng)關(guān)] --> A
F --> B
F --> D
```
## 時序圖
[關(guān)鍵流程的時序圖，展示組件間的交互順序]
```mermaid
sequenceDiagram
participant Client as 客戶端
participant Gateway as API網(wǎng)關(guān)
participant UserService as 用戶服務(wù)
participant OrderService as 訂單服務(wù)
participant PaymentService as 支付服務(wù)
Client->>Gateway: 創(chuàng)建訂單請求
Gateway->>UserService: 驗證用戶信息
UserService-->>Gateway: 用戶驗證結(jié)果
Gateway->>OrderService: 創(chuàng)建訂單
OrderService->>PaymentService: 發(fā)起支付
PaymentService-->>OrderService: 支付結(jié)果
OrderService-->>Gateway: 訂單創(chuàng)建結(jié)果
Gateway-->>Client: 返回訂單信息
```
# 四、詳細(xì)方案設(shè)計
## 架構(gòu)選型
[說明本方案采用的架構(gòu)模式]
### 分層架構(gòu)說明
[描述本方案的分層架構(gòu)，說明各層職責(zé)]
#### DDD領(lǐng)域驅(qū)動設(shè)計架構(gòu)（DDD架構(gòu)項目）
```
# Interface層（接口層）
- Controllers：處理HTTP請求，參數(shù)校驗和響應(yīng)封裝
- DTOs：數(shù)據(jù)傳輸對象，用于接口間數(shù)據(jù)傳遞
- Serializers：數(shù)據(jù)序列化和反序列化
# Application層（應(yīng)用層）
- Services：應(yīng)用服務(wù)，協(xié)調(diào)領(lǐng)域?qū)ο笸瓿蓸I(yè)務(wù)用例
- Commands/Queries：命令和查詢對象，實現(xiàn)CQRS模式
- Handlers：命令和查詢處理器
# Domain層（領(lǐng)域?qū)樱?- Entities：實體對象，包含業(yè)務(wù)邏輯和業(yè)務(wù)規(guī)則
- ValueObjects：值對象，不可變的業(yè)務(wù)概念
- DomainServices：領(lǐng)域服務(wù)，處理跨聚合的業(yè)務(wù)邏輯
- Repositories：倉儲接口，定義數(shù)據(jù)訪問抽象
# Infrastructure層（基礎(chǔ)設(shè)施層）
- Repositories：倉儲實現(xiàn)，具體的數(shù)據(jù)訪問實現(xiàn)
- ExternalServices：外部服務(wù)集成
- Configuration：配置管理
```
#### 微服務(wù)架構(gòu)（微服務(wù)架構(gòu)項目）
```
# API Gateway層
- 路由轉(zhuǎn)發(fā)、負(fù)載均衡、認(rèn)證授權(quán)
- 限流熔斷、監(jiān)控日志
# Service層（各個微服務(wù)）
- Controllers：處理HTTP請求
- Services：業(yè)務(wù)邏輯處理
- Models：數(shù)據(jù)模型
- Repositories：數(shù)據(jù)訪問
# Message Queue層
- Producers：消息生產(chǎn)者
- Consumers：消息消費者
- Event Handlers：事件處理器
# Infrastructure層
- Database：數(shù)據(jù)存儲
- Cache：緩存服務(wù)
- Monitoring：監(jiān)控服務(wù)
```
#### 傳統(tǒng)分層架構(gòu)（傳統(tǒng)分層架構(gòu)項目）
```
# Presentation層（表現(xiàn)層）
- Controllers/Views：處理HTTP請求，參數(shù)校驗
- Templates：模板文件（如使用Django/Flask模板）
- Forms：表單處理和驗證
- Serializers：數(shù)據(jù)序列化
# Business層（業(yè)務(wù)層）
- Services：實現(xiàn)業(yè)務(wù)邏輯
- Managers：業(yè)務(wù)管理器，協(xié)調(diào)多個服務(wù)
- Utils：業(yè)務(wù)工具類
# Data Access層（數(shù)據(jù)訪問層）
- Models：數(shù)據(jù)模型（ORM模型）
- Repositories：數(shù)據(jù)訪問抽象
- DAOs：數(shù)據(jù)訪問對象
# Infrastructure層（基礎(chǔ)設(shè)施層）
- Database：數(shù)據(jù)庫配置和連接
- Cache：緩存配置
- External APIs：外部API集成
```
### 數(shù)據(jù)模型設(shè)計
[描述數(shù)據(jù)模型的設(shè)計，包括不同層次的數(shù)據(jù)模型]
```python
# DDD架構(gòu)數(shù)據(jù)模型
# DTO (Data Transfer Object) - 接口層數(shù)據(jù)傳輸對象
@dataclass
class UserCreateDTO:
username: str
email: str
password: str
# Entity - 領(lǐng)域?qū)嶓w
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:
# 業(yè)務(wù)規(guī)則：密碼強(qiáng)度驗證
if len(new_password) < 8:
raise DomainException("密碼長度不能少于8位")
self.password_hash = hash_password(new_password)
# Model - 數(shù)據(jù)庫模型（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
# 傳統(tǒng)分層架構(gòu)數(shù)據(jù)模型
# 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 - 數(shù)據(jù)序列化（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)
```
## 應(yīng)用1
### 業(yè)務(wù)流程1
#### xxx接口
**接口說明**：[詳細(xì)說明接口的用途和功能]
**接口路徑**：[HTTP方法] [路徑] 或 [RPC服務(wù)接口定義]
**請求參數(shù)**：
```json
{
"param1": "value1",
"param2": "value2"
}
```
**返回結(jié)果**：
```json
{
"code": 200,
"message": "success",
"data": {
"field1": "value1",
"field2": "value2"
}
}
```
#### 接口改動點
[說明接口的改動類型：新增、能力調(diào)整、擴(kuò)展等，并詳述改動內(nèi)容]
#### 代碼分層設(shè)計
[描述代碼的分層結(jié)構(gòu)，確保符合工程規(guī)范]
##### DDD架構(gòu)分層設(shè)計（DDD架構(gòu)項目）
```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):
"""創(chuàng)建用戶接口"""
# 職責(zé)：參數(shù)校驗、請求處理、結(jié)果封裝
data = request.get_json()
user_dto = UserCreateDTO(**data)
result = self.user_service.create_user(user_dto)
return jsonify({"code": 200, "data": result})
# Application層 - 應(yīng)用服務(wù)
# 位置：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:
"""創(chuàng)建用戶業(yè)務(wù)邏輯"""
# 職責(zé)：協(xié)調(diào)領(lǐng)域?qū)ο螅瑢崿F(xiàn)業(yè)務(wù)用例
# 檢查用戶名是否存在
if self.user_repository.find_by_username(user_dto.username):
raise BusinessException("用戶名已存在")
# 創(chuàng)建用戶實體
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層 - 領(lǐng)域?qū)嶓w
# 位置：src/domain/entities/user.py
class User(Entity):
def __init__(self, username: str, email: str, password_hash: str):
# 職責(zé)：封裝業(yè)務(wù)規(guī)則和業(yè)務(wù)邏輯
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':
"""創(chuàng)建用戶工廠方法"""
# 業(yè)務(wù)規(guī)則：用戶名長度驗證
if len(username) < 3:
raise DomainException("用戶名長度不能少于3位")
# 業(yè)務(wù)規(guī)則：郵箱格式驗證
if not cls._is_valid_email(email):
raise DomainException("郵箱格式不正確")
password_hash = hash_password(password)
return cls(username, email, password_hash)
# Infrastructure層 - 倉儲實現(xiàn)
# 位置：src/infrastructure/repositories/user_repository.py
class SqlAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
"""保存用戶到數(shù)據(jù)庫"""
# 職責(zé)：將領(lǐng)域?qū)ο筠D(zhuǎn)換為數(shù)據(jù)庫模型并持久化
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()
# 轉(zhuǎn)換回領(lǐng)域?qū)ο? return User(
username=user_model.username,
email=user_model.email,
password_hash=user_model.password_hash
)
```
##### 傳統(tǒng)分層架構(gòu)設(shè)計（傳統(tǒng)分層架構(gòu)項目）
```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():
"""創(chuàng)建用戶接口"""
# 職責(zé)：參數(shù)校驗、請求處理、結(jié)果封裝
data = request.get_json()
# 參數(shù)校驗
if not data.get('username') or not data.get('email'):
return jsonify({"code": 400, "message": "參數(shù)不完整"}), 400
# 調(diào)用服務(wù)層
user_service = UserService()
result = user_service.create_user(data)
return jsonify({"code": 200, "data": result})
# Service層 - 業(yè)務(wù)邏輯
# 位置：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:
"""創(chuàng)建用戶業(yè)務(wù)邏輯"""
# 職責(zé)：實現(xiàn)業(yè)務(wù)邏輯，協(xié)調(diào)數(shù)據(jù)訪問層
# 業(yè)務(wù)規(guī)則：檢查用戶名是否存在
if self.user_repository.find_by_username(user_data['username']):
raise BusinessException("用戶名已存在")
# 業(yè)務(wù)規(guī)則：密碼加密
user_data['password_hash'] = hash_password(user_data['password'])
del user_data['password']
# 創(chuàng)建用戶
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層 - 數(shù)據(jù)訪問
# 位置：app/repositories/user_repository.py
from app.models.user import User
from app.extensions import db
class UserRepository:
def save(self, user: User) -> User:
"""保存用戶到數(shù)據(jù)庫"""
# 職責(zé)：封裝數(shù)據(jù)訪問邏輯
db.session.add(user)
db.session.commit()
return user
def find_by_username(self, username: str) -> User:
"""根據(jù)用戶名查找用戶"""
return User.query.filter_by(username=username).first()
# Model層 - 數(shù)據(jù)模型
# 位置：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):
"""轉(zhuǎn)換為字典"""
return {
'id': self.id,
'username': self.username,
'email': self.email,
'created_at': self.created_at.isoformat()
}
```
##### 微服務(wù)架構(gòu)設(shè)計（微服務(wù)架構(gòu)項目）
```python
# API Gateway - 網(wǎng)關(guān)路由
# 位置：gateway/main.py
from fastapi import FastAPI, HTTPException
import httpx
app = FastAPI()
@app.post("/api/users")
async def create_user_gateway(user_data: dict):
"""用戶創(chuàng)建網(wǎng)關(guān)接口"""
# 職責(zé)：路由轉(zhuǎn)發(fā)、認(rèn)證授權(quán)、限流
# 認(rèn)證檢查
if not validate_token(request.headers.get('Authorization')):
raise HTTPException(status_code=401, detail="未授權(quán)")
# 轉(zhuǎn)發(fā)到用戶服務(wù)
async with httpx.AsyncClient() as client:
response = await client.post(
"http://user-service:8001/users",
jsnotallow=user_data
)
return response.json()
# User Service - 用戶微服務(wù)
# 位置：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)
):
"""創(chuàng)建用戶"""
# 職責(zé)：處理用戶相關(guān)的業(yè)務(wù)邏輯
# 檢查用戶是否存在
existing_user = await user_repo.find_by_username(user_data.username)
if existing_user:
raise HTTPException(status_code=400, detail="用戶名已存在")
# 創(chuàng)建用戶
user = User(
username=user_data.username,
email=user_data.email,
password_hash=hash_password(user_data.password)
)
saved_user = await user_repo.save(user)
# 發(fā)送用戶創(chuàng)建事件
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):
"""處理用戶創(chuàng)建事件"""
# 職責(zé)：處理異步事件，發(fā)送歡迎郵件
email_service = EmailService()
await email_service.send_welcome_email(
user_id=event_data['user_id'],
username=event_data['username']
)
```
#### 設(shè)計模式應(yīng)用（Python特有）
##### 裝飾器模式應(yīng)用
```python
# 權(quán)限控制裝飾器
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": "權(quán)限不足"}), 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()
# 數(shù)據(jù)庫
database = providers.Singleton(
Database,
connection_string=config.database.connection_string
)
# 倉儲
user_repository = providers.Factory(
SqlAlchemyUserRepository,
sessinotallow=database.provided.session
)
# 服務(wù)
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
```
#### 代碼改動點
[詳述需要改動的代碼，重點說明實現(xiàn)思路]
1. **Controller/View層改動**：
```python
# 新增用戶管理控制器
# 位置：src/interface/controllers/user_controller.py
# 改動內(nèi)容：新增用戶CRUD接口
# 實現(xiàn)思路：使用Flask Blueprint組織路由，通過依賴注入獲取服務(wù)實例
```
2. **Service層改動**：
```python
# 新增用戶服務(wù)
# 位置：src/application/services/user_service.py
# 改動內(nèi)容：實現(xiàn)用戶注冊、登錄、信息更新等業(yè)務(wù)邏輯
# 實現(xiàn)思路：采用領(lǐng)域驅(qū)動設(shè)計，通過倉儲模式訪問數(shù)據(jù)
```
3. **Repository層改動**：
```python
# 新增用戶倉儲實現(xiàn)
# 位置：src/infrastructure/repositories/user_repository.py
# 改動內(nèi)容：實現(xiàn)用戶數(shù)據(jù)的CRUD操作
# 實現(xiàn)思路：使用SQLAlchemy ORM，實現(xiàn)倉儲接口
```
4. **DDD特定改動（DDD架構(gòu)項目）**：
```python
# 新增用戶聚合根
# 位置：src/domain/aggregates/user_aggregate.py
# 改動內(nèi)容：定義用戶聚合的業(yè)務(wù)規(guī)則和不變量
# 實現(xiàn)思路：通過工廠方法創(chuàng)建實體，封裝業(yè)務(wù)規(guī)則
```
5. **微服務(wù)特定改動（微服務(wù)架構(gòu)項目）**：
```python
# 新增用戶微服務(wù)
# 位置：services/user_service/
# 改動內(nèi)容：獨立的用戶服務(wù)，包含完整的用戶管理功能
# 實現(xiàn)思路：使用FastAPI構(gòu)建REST API，通過消息隊列與其他服務(wù)通信
```
## 數(shù)據(jù)庫變更
### 表結(jié)構(gòu)設(shè)計
[描述需要新增或修改的數(shù)據(jù)庫表結(jié)構(gòu)]
#### 表名：users
| 字段名 | 數(shù)據(jù)類型 | 是否為空 | 主鍵 | 默認(rèn)值 | 注釋 |
|----|---|---|---|---|---|
| id | BIGINT | 否 | 是 | AUTO_INCREMENT | 主鍵ID |
| username | VARCHAR(50) | 否 | 否 | - | 用戶名 |
| email | VARCHAR(100) | 否 | 否 | - | 郵箱地址 |
| password_hash | VARCHAR(255) | 否 | 否 | - | 密碼哈希 |
| status | TINYINT | 否 | 否 | 1 | 用戶狀態(tài)(1:正常,0:禁用) |
| created_at | DATETIME | 否 | 否 | CURRENT_TIMESTAMP | 創(chuàng)建時間 |
| 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)
```
### 數(shù)據(jù)庫遷移腳本（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')
)
# 創(chuàng)建索引
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')
```
### 索引設(shè)計
| 索引名 | 字段 | 索引類型 | 說明 |
|----|---|---|---|
| PRIMARY | id | 主鍵 | 主鍵索引 |
| uk_username | username | 唯一 | 用戶名唯一索引 |
| uk_email | email | 唯一 | 郵箱唯一索引 |
| idx_created_at | created_at | 普通 | 創(chuàng)建時間索引，用于排序查詢 |
| idx_status | status | 普通 | 狀態(tài)索引，用于篩選有效用戶 |
## 接口文檔變更
### API接口規(guī)范
[描述需要新增或修改的接口文檔]
#### 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):
"""用戶創(chuàng)建請求"""
username: str = Field(..., min_length=3, max_length=50, descriptinotallow="用戶名")
email: EmailStr = Field(..., descriptinotallow="郵箱地址")
password: str = Field(..., min_length=8, descriptinotallow="密碼")
class UserResponse(BaseModel):
"""用戶響應(yīng)"""
id: int = Field(..., descriptinotallow="用戶ID")
username: str = Field(..., descriptinotallow="用戶名")
email: str = Field(..., descriptinotallow="郵箱地址")
status: bool = Field(..., descriptinotallow="用戶狀態(tài)")
created_at: datetime = Field(..., descriptinotallow="創(chuàng)建時間")
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)
):
"""
創(chuàng)建用戶
- **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='用戶管理相關(guān)操作')
# 定義請求模型
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)
})
# 定義響應(yīng)模型
user_response_model = api.model('UserResponse', {
'id': fields.Integer(descriptinotallow='用戶ID'),
'username': fields.String(descriptinotallow='用戶名'),
'email': fields.String(descriptinotallow='郵箱地址'),
'status': fields.Boolean(descriptinotallow='用戶狀態(tài)'),
'created_at': fields.DateTime(descriptinotallow='創(chuàng)建時間')
})
@api.route('/')
class UserListResource(Resource):
@api.expect(user_create_model)
@api.marshal_with(user_response_model, code=201)
@api.doc('create_user', descriptinotallow='創(chuàng)建新用戶')
def post(self):
"""創(chuàng)建用戶"""
data = request.get_json()
# 處理邏輯
pass
```
## 配置變更
### 應(yīng)用配置
```python
# 位置：config/settings.py
import os
from typing import Optional
class Settings:
"""應(yīng)用配置"""
# 應(yīng)用基礎(chǔ)配置
APP_NAME: str = "User Management System"
APP_VERSION: str = "1.0.0"
DEBUG: bool = os.getenv("DEBUG", "False").lower() == "true"
# 數(shù)據(jù)庫配置
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
# 安裝系統(tǒng)依賴
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
# 復(fù)制應(yīng)用代碼
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
```
## 非功能性需求
### 性能需求
- **響應(yīng)時間**: 用戶注冊接口響應(yīng)時間 < 500ms
- **并發(fā)量**: 支持1000個并發(fā)用戶注冊
- **吞吐量**: 每秒處理100個用戶注冊請求
### 可用性需求
- **系統(tǒng)可用率**:99.9%
- **故障恢復(fù)時間**: < 5分鐘
- **數(shù)據(jù)備份**: 每日自動備份，保留30天
### 擴(kuò)展性需求
- **水平擴(kuò)展**: 支持通過增加服務(wù)實例來擴(kuò)展處理能力
- **數(shù)據(jù)庫分片**: 支持用戶數(shù)據(jù)按用戶ID分片存儲
- **緩存策略**: 使用Redis緩存熱點用戶數(shù)據(jù)
## 兼容性與平滑遷移方案
### 版本兼容性
- **API版本控制**: 使用URL路徑版本控制（/api/v1/, /api/v2/）
- **數(shù)據(jù)庫遷移**: 使用Alembic進(jìn)行數(shù)據(jù)庫版本管理
- **向后兼容**: 新版本API保持對舊版本的兼容性
### 平滑遷移方案
```python
# 數(shù)據(jù)遷移腳本示例
# 位置：scripts/migrate_user_data.py
importasyncio
fromsrc.infrastructure.databaseimportget_db_session
fromsrc.infrastructure.models.user_modelimportUserModel
asyncdefmigrate_user_data():
"""遷移用戶數(shù)據(jù)"""
asyncwithget_db_session() assession:
# 批量處理用戶數(shù)據(jù)
batch_size = 1000
offset = 0
whileTrue:
users = awaitsession.execute(
select(UserModel).offset(offset).limit(batch_size)
)
user_list = users.scalars().all()
ifnotuser_list:
break
# 處理用戶數(shù)據(jù)遷移邏輯
foruserinuser_list:
# 數(shù)據(jù)轉(zhuǎn)換和更新
pass
awaitsession.commit()
offset += batch_size
print(f"已處理 {offset} 個用戶")
if__name__ == "__main__":
asyncio.run(migrate_user_data())
```
## 風(fēng)險與應(yīng)對措施
| 風(fēng)險 | 可能性 | 影響 | 應(yīng)對措施 |
|---|----|---|---|
| 數(shù)據(jù)庫連接池耗盡 | 中 | 高 | 配置連接池監(jiān)控，實現(xiàn)連接池動態(tài)調(diào)整 |
| 密碼哈希算法安全性 | 低 | 高 | 使用bcrypt或Argon2等安全哈希算法 |
| 郵箱驗證服務(wù)不可用 | 中 | 中 | 實現(xiàn)郵件服務(wù)降級，支持多個郵件服務(wù)商 |
| 用戶數(shù)據(jù)泄露 | 低 | 高 | 實現(xiàn)數(shù)據(jù)加密存儲，定期安全審計 |
| 高并發(fā)下的性能問題 | 高 | 中 | 實現(xiàn)限流、緩存和異步處理 |
## Python特有的最佳實踐
### 代碼規(guī)范
```python
# 使用類型注解
fromtypingimportOptional, List, Dict, Any
defcreate_user(user_data:Dict[str, Any]) -> Optional[User]:
"""創(chuàng)建用戶，返回用戶對象或None"""
pass
# 使用dataclass簡化數(shù)據(jù)類
from dataclasses import dataclass
@dataclass
class UserDTO:
username: str
email: str
password: str
def __post_init__(self):
"""數(shù)據(jù)驗證"""
if len(self.username) < 3:
raise ValueError("用戶名長度不能少于3位")
```
### 異步編程
```python
# 異步服務(wù)實現(xiàn)
import asyncio
from typing import List
class AsyncUserService:
async def create_users_batch(self, users_data: List[Dict]) -> List[User]:
"""批量創(chuàng)建用戶"""
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:
"""異步創(chuàng)建單個用戶"""
# 異步數(shù)據(jù)庫操作
user = await self.user_repository.save_async(user_data)
# 異步發(fā)送郵件
await self.email_service.send_welcome_email_async(user.email)
return user
```
### 上下文管理器
```python
# 數(shù)據(jù)庫事務(wù)管理
from contextlib import asynccontextmanager
@asynccontextmanager
async def database_transaction():
"""數(shù)據(jù)庫事務(wù)上下文管理器"""
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架構(gòu)項目技術(shù)方案示例
#### 用戶管理系統(tǒng) - 創(chuàng)建用戶功能
**架構(gòu)類型**: DDD領(lǐng)域驅(qū)動設(shè)計架構(gòu) - Python 3.11 + FastAPI 0.104.1
##### 代碼分層設(shè)計
```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)
):
"""創(chuàng)建用戶接口"""
# 職責(zé)：參數(shù)校驗、請求處理、結(jié)果封裝
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層 - 應(yīng)用服務(wù)
# 位置: 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:
"""創(chuàng)建用戶應(yīng)用服務(wù)"""
# 職責(zé)：協(xié)調(diào)領(lǐng)域?qū)ο螅瑢崿F(xiàn)業(yè)務(wù)用例
# 檢查用戶是否已存在
existing_user = await self.user_repo.find_by_email(request.email)
if existing_user:
raise UserAlreadyExistsException("郵箱已被注冊")
# 創(chuàng)建用戶領(lǐng)域?qū)ο? 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層 - 領(lǐng)域?qū)嶓w
# 位置: 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':
"""創(chuàng)建用戶工廠方法"""
# 職責(zé)：封裝用戶創(chuàng)建的業(yè)務(wù)規(guī)則
# 驗證用戶名
username_vo = Username(username)
if not username_vo.is_valid():
raise InvalidUsernameException("用戶名格式不正確")
# 驗證郵箱
email_vo = Email(email)
if not email_vo.is_valid():
raise InvalidEmailException("郵箱格式不正確")
# 驗證密碼強(qiáng)度
if not password_service.is_strong_password(password):
raise WeakPasswordException("密碼強(qiáng)度不夠")
# 生成密碼哈希
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("新密碼強(qiáng)度不夠")
self.password_hash = password_service.hash_password(new_password)
# Infrastructure層 - 倉儲實現(xiàn)
# 位置: 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:
"""保存用戶到數(shù)據(jù)庫"""
# 職責(zé)：將領(lǐng)域?qū)ο筠D(zhuǎn)換為數(shù)據(jù)庫模型并持久化
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)
# 轉(zhuǎn)換回領(lǐng)域?qū)ο? 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
)
```
根據(jù)上述DDD架構(gòu)設(shè)計，創(chuàng)建用戶功能的調(diào)用鏈為：
1. 客戶端調(diào)用 `POST /api/v1/users/` 接口
2. UserController 接收請求并調(diào)用 UserService.create_user()
3. UserService 協(xié)調(diào)領(lǐng)域?qū)ο螅{(diào)用 User.create() 工廠方法
4. User實體執(zhí)行業(yè)務(wù)規(guī)則驗證，創(chuàng)建用戶對象
5. UserService 調(diào)用 UserRepository.save() 持久化用戶
6. SqlAlchemyUserRepository 將領(lǐng)域?qū)ο筠D(zhuǎn)換為數(shù)據(jù)庫模型并保存
7. 返回創(chuàng)建的用戶對象，層層返回到客戶端
### 傳統(tǒng)分層架構(gòu)項目技術(shù)方案示例
#### 博客系統(tǒng) - 文章管理功能
**架構(gòu)類型**: 傳統(tǒng)分層架構(gòu) - Python 3.11 + Flask 2.3.3
##### 代碼分層設(shè)計
```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():
"""創(chuàng)建文章接口"""
# 職責(zé)：參數(shù)校驗、請求處理、結(jié)果封裝
data = request.get_json()
# 參數(shù)驗證
schema = ArticleCreateSchema()
try:
validated_data = schema.load(data)
except ValidationError as e:
return jsonify({"errors": e.messages}), 400
# 調(diào)用服務(wù)層
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層 - 業(yè)務(wù)邏輯
# 位置: 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:
"""創(chuàng)建文章業(yè)務(wù)邏輯"""
# 職責(zé)：實現(xiàn)業(yè)務(wù)邏輯，協(xié)調(diào)數(shù)據(jù)訪問層
# 業(yè)務(wù)規(guī)則：生成文章slug
slug = generate_slug(article_data['title'])
# 業(yè)務(wù)規(guī)則：檢查slug是否重復(fù)
if self.article_repo.find_by_slug(slug):
slug = f"{slug}-{int(time.time())}"
# 業(yè)務(wù)規(guī)則：設(shè)置發(fā)布狀態(tài)
article_data['slug'] = slug
article_data['status'] = 'draft' # 默認(rèn)為草稿
article_data['created_at'] = datetime.now()
# 創(chuàng)建文章
article = Article(**article_data)
return self.article_repo.save(article)
def publish_article(self, article_id: int) -> Article:
"""發(fā)布文章"""
article = self.article_repo.find_by_id(article_id)
if not article:
raise ArticleNotFoundException("文章不存在")
# 業(yè)務(wù)規(guī)則：只有草稿狀態(tài)的文章可以發(fā)布
if article.status != 'draft':
raise InvalidOperationException("只有草稿狀態(tài)的文章可以發(fā)布")
article.status = 'published'
article.published_at = datetime.now()
return self.article_repo.save(article)
# Repository層 - 數(shù)據(jù)訪問
# 位置: app/repositories/article_repository.py
from app.models.article import Article
from app.extensions import db
class ArticleRepository:
def save(self, article: Article) -> Article:
"""保存文章到數(shù)據(jù)庫"""
# 職責(zé)：封裝數(shù)據(jù)訪問邏輯
db.session.add(article)
db.session.commit()
return article
def find_by_id(self, article_id: int) -> Article:
"""根據(jù)ID查找文章"""
return Article.query.get(article_id)
def find_by_slug(self, slug: str) -> Article:
"""根據(jù)slug查找文章"""
return Article.query.filter_by(slug=slug).first()
def find_published_articles(self, page: int = 1, per_page: int = 10):
"""查找已發(fā)布的文章"""
return Article.query.filter_by(status='published')\
.order_by(Article.published_at.desc())\
.paginate(page=page, per_page=per_page)
# Model層 - 數(shù)據(jù)模型
# 位置: 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):
"""轉(zhuǎn)換為字典"""
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
}
```
根據(jù)上述傳統(tǒng)分層架構(gòu)設(shè)計，創(chuàng)建文章功能的調(diào)用鏈為：
1. 客戶端調(diào)用 `POST /api/articles/` 接口
2. ArticleController 接收請求，驗證參數(shù)
3. Controller 調(diào)用 ArticleService.create_article()
4. ArticleService 處理業(yè)務(wù)邏輯（生成slug、設(shè)置狀態(tài)等）
5. Service 調(diào)用 ArticleRepository.save() 保存文章
6. Repository 使用 SQLAlchemy ORM 將文章保存到數(shù)據(jù)庫
7. 返回創(chuàng)建的文章對象，層層返回到客戶端
### 微服務(wù)架構(gòu)項目技術(shù)方案示例
#### 電商系統(tǒng) - 訂單處理功能
**架構(gòu)類型**: 微服務(wù)架構(gòu) - Python 3.11 + FastAPI 0.104.1 + Docker
##### 微服務(wù)架構(gòu)設(shè)計
```python
# API Gateway - 網(wǎng)關(guān)服務(wù)
# 位置: 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="電商系統(tǒng)API網(wǎng)關(guān)")
@app.post("/api/orders")
async def create_order_gateway(
order_data: dict,
user_id: str = Depends(verify_token)
):
"""訂單創(chuàng)建網(wǎng)關(guān)接口"""
# 職責(zé)：路由轉(zhuǎn)發(fā)、認(rèn)證授權(quán)、限流
# 添加用戶ID到請求數(shù)據(jù)
order_data['user_id'] = user_id
# 轉(zhuǎn)發(fā)到訂單服務(wù)
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 - 訂單微服務(wù)
# 位置: 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="訂單服務(wù)")
@app.post("/orders", status_code=201)
async def create_order(
order_request: OrderCreateRequest,
order_service: OrderService = Depends(get_order_service)
):
"""創(chuàng)建訂單"""
# 職責(zé)：處理訂單相關(guān)的業(yè)務(wù)邏輯
try:
# 驗證商品庫存
await order_service.validate_inventory(order_request.items)
# 創(chuàng)建訂單
order = await order_service.create_order(order_request)
# 發(fā)布訂單創(chuàng)建事件
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="訂單創(chuàng)建失敗")
# Inventory Service - 庫存微服務(wù)
# 位置: 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="庫存服務(wù)")
@app.post("/inventory/reserve")
async def reserve_inventory(
reserve_request: InventoryReserveRequest,
inventory_service: InventoryService = Depends(get_inventory_service)
):
"""預(yù)留庫存"""
# 職責(zé)：處理庫存預(yù)留邏輯
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 - 支付微服務(wù)
# 位置: 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):
"""處理訂單創(chuàng)建事件"""
# 職責(zé)：處理異步事件，創(chuàng)建支付記錄
payment_service = PaymentService()
# 創(chuàng)建支付記錄
payment = await payment_service.create_payment(
order_id=event_data['order_id'],
amount=event_data['total_amount'],
user_id=event_data['user_id']
)
# 發(fā)布支付創(chuàng)建事件
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):
"""發(fā)布事件"""
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:
```
根據(jù)上述微服務(wù)架構(gòu)設(shè)計，創(chuàng)建訂單功能的調(diào)用鏈為：
1. 客戶端調(diào)用 API Gateway 的 `POST /api/orders` 接口
2. Gateway 驗證用戶身份，轉(zhuǎn)發(fā)請求到 Order Service
3. Order Service 調(diào)用 Inventory Service 驗證庫存
4. Order Service 創(chuàng)建訂單，發(fā)布 "order.created" 事件
5. Payment Service 消費事件，創(chuàng)建支付記錄
6. 各服務(wù)將數(shù)據(jù)保存到各自的數(shù)據(jù)庫
7. 返回訂單創(chuàng)建結(jié)果，層層返回到客戶端
## 方案設(shè)計工作流程
1. **架構(gòu)識別階段**
- 確定項目使用的Python框架（Django/Flask/FastAPI）
- 識別架構(gòu)模式（DDD/微服務(wù)/傳統(tǒng)分層）
- 確定數(shù)據(jù)庫類型（PostgreSQL/MySQL/MongoDB）
- 識別消息隊列和緩存方案
2. **需求分析階段**
- 確定功能邊界和核心業(yè)務(wù)流程
- 識別核心業(yè)務(wù)實體和領(lǐng)域模型
- 確定API接口設(shè)計和數(shù)據(jù)結(jié)構(gòu)
- 識別異步處理需求
3. **方案設(shè)計階段**
- 根據(jù)架構(gòu)特點進(jìn)行分層設(shè)計
- 確定接口實現(xiàn)和組件交互
- 設(shè)計數(shù)據(jù)庫結(jié)構(gòu)和ORM模型
- 對于微服務(wù)項目，設(shè)計服務(wù)拆分和通信機(jī)制
4. **方案評審階段**
- 驗證方案與Python最佳實踐的一致性
- 驗證功能覆蓋度和完整性
- 評估性能問題和擴(kuò)展性
- 確保方案文檔結(jié)構(gòu)清晰、內(nèi)容完整中間件規(guī)范Rule
# 中間件使用規(guī)范
---
description: `此規(guī)則適用于go項目的單元測試開發(fā)規(guī)范，Go單元測試提供全面指南，規(guī)范測試結(jié)構(gòu)、mock技術(shù)和斷言方法，確保測試代碼質(zhì)量與可維護(hù)性。`
globs:
alwaysApply: false
---
# 中間件客戶端調(diào)用規(guī)范
## 關(guān)鍵規(guī)則
- **所有HTTP和HSF等中間件客戶端必須放在framework/client目錄下**
- **必須遵循統(tǒng)一的命名規(guī)范：服務(wù)功能+Service命名方式**
- **必須使用Context作為第一個參數(shù)，支持分布式追蹤和日志記錄**
- **中間件客戶端必須從配置中心讀取配置，不允許硬編碼**
- **中間件調(diào)用必須記錄完整的請求和響應(yīng)日志**
- **必須實現(xiàn)統(tǒng)一的錯誤處理和返回機(jī)制**
- **應(yīng)對關(guān)鍵調(diào)用實現(xiàn)緩存機(jī)制，減少直接調(diào)用次數(shù)**
- **請求和響應(yīng)結(jié)構(gòu)體必須實現(xiàn)JavaClassName方法（HSF特有）**
- **HSF服務(wù)必須在init方法中注冊模型**
- **客戶端調(diào)用需要進(jìn)行合理的超時控制**
## HTTP客戶端標(biāo)準(zhǔn)實現(xiàn)
### 客戶端定義規(guī)范
```go
// 客戶端標(biāo)準(zhǔn)定義
type XXXHttpClient struct {
// 可選的客戶端配置
}
// 必須定義統(tǒng)一的初始化方法
func NewXXXHttpClient(ctx context.Context) *XXXHttpClient {
return &XXXHttpClient{}
}
```
### 請求參數(shù)定義
```go
// 請求參數(shù)必須使用結(jié)構(gòu)體定義
type RequestParams struct {
// 請求字段
Field1 string `json:"field1"`
Field2 int `json:"field2"`
// ...
}
```
### 標(biāo)準(zhǔn)HTTP調(diào)用實現(xiàn)
```go
// 標(biāo)準(zhǔn)HTTP調(diào)用方法
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. 構(gòu)建URL和請求參數(shù)
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. 構(gòu)建請求選項
opt := simplehttp.BuildOptions(&simplehttp.Options{
Method: "GET", // 或 POST 等
Params: httpParams,
Timeout: conf.Timeout,
HostWithVip: conf.VipHost,
RecordWithParams: httpcommon.RecordWithParams,
})
// 4. 發(fā)送請求并記錄日志
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. 解析響應(yīng)
var response ResponseType
err = json.Unmarshal(respData, &response)
if err != nil {
return nil, common.Errorf(common.ErrorInternal, "unmarshal failed.err:%s", err.Error())
}
// 7. 返回結(jié)果
return response, nil
}
```
### 帶緩存的HTTP調(diào)用
```go
func GetDataWithCache(ctx context.Context, key string, params RequestParams) (ResponseType, error) {
var resp ResponseType
cacheKey := fmt.Sprintf("cache_key_prefix_%s", key)
// 使用緩存機(jī)制
err := mcache.GetLocalCacheFiveSecond(ctx).Once(ctx, cacheKey, &resp, 5*time.Second, func() (interface{}, error) {
// 調(diào)用實際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客戶端標(biāo)準(zhǔn)實現(xiàn)
### 服務(wù)定義規(guī)范
```go
// 全局服務(wù)實例
var XXXService = new(XXXServiceImpl)
// 注冊HSF模型
func init() {
hsfCommon.RegisterModel(&ModelType1{})
hsfCommon.RegisterModel(&ModelType2{})
// 其他模型注冊...
}
// 服務(wù)結(jié)構(gòu)體定義
type XXXServiceImpl struct {
// 方法定義，必須遵循標(biāo)準(zhǔn)方法簽名
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
// 請求模型必須實現(xiàn)JavaClassName方法
type RequestType struct {
Field1 string `json:"field1" hessian:"field1"`
Field2 int64 `json:"field2" hessian:"field2"`
// ...
}
func (RequestType) JavaClassName() string {
return"com.package.RequestType"
}
// 響應(yīng)模型必須實現(xiàn)JavaClassName方法
type ResponseType struct {
Code int32 `json:"code"`
Data interface{} `json:"data"`
Success bool `json:"success"`
// ...
}
func (ResponseType) JavaClassName() string {
return"com.package.ResponseType"
}
```
### 標(biāo)準(zhǔn)HSF調(diào)用實現(xiàn)
```go
// 標(biāo)準(zhǔn)HSF調(diào)用方法
func CallHSFService(ctx context.Context, request *RequestType) (*DataType, *common.Error) {
// 1. 調(diào)用HSF服務(wù)
hsfResp, e := XXXService.MethodName(ctx, []interface{}{request})
// 2. 記錄請求和響應(yīng)日志
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. 結(jié)果處理
if hsfResp != nil {
result := ParseResponseData(hsfResp.Data)
return result, nil
}
return nil, nil
}
// 解析響應(yīng)數(shù)據(jù)的標(biāo)準(zhǔn)方法
func ParseResponseData(data interface{}) *DataType {
if data == nil {
return nil
}
if items, ok := data.(SpecificResponseType); ok {
// 處理數(shù)據(jù)轉(zhuǎn)換
result := &DataType{
// 數(shù)據(jù)轉(zhuǎn)換邏輯
}
return result
}
return nil
}
```
### 帶緩存的HSF調(diào)用
```go
func GetHSFDataWithCache(ctx context.Context, param1, param2 string) (*DataType, error) {
var resp *DataType
cacheKey := fmt.Sprintf("hsf_cache_key_%s_%s", param1, param2)
// 使用緩存機(jī)制
err := mcache.GetLocalCacheFiveSecond(ctx).Once(ctx, cacheKey, &resp, 5*time.Second, func() (interface{}, error) {
// 構(gòu)建HSF請求
request := &RequestType{
Field1: param1,
Field2: param2,
}
// 調(diào)用HSF服務(wù)
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
}
```
## 錯誤處理規(guī)范
- **所有中間件調(diào)用都必須返回標(biāo)準(zhǔn)化的錯誤**
- **錯誤必須包含錯誤碼和錯誤信息**
- **網(wǎng)絡(luò)錯誤必須歸類為InternalError**
- **參數(shù)錯誤必須歸類為InvalidError**
- **業(yè)務(wù)邏輯錯誤必須根據(jù)具體場景進(jìn)行分類**
```go
// 錯誤處理示例
if err != nil {
// 網(wǎng)絡(luò)錯誤
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())
}
// 業(yè)務(wù)錯誤
if resp.Code != 200 {
return nil, common.Errorf(common.ErrorBusiness, "business error: %s", resp.Message)
}
```
## 日志記錄規(guī)范
- **所有中間件調(diào)用必須記錄請求和響應(yīng)日志**
- **日志必須包含追蹤ID、請求參數(shù)和響應(yīng)結(jié)果**
- **敏感信息（如密碼、token）必須在日志中脫敏**
- **必須使用統(tǒng)一的日志模塊和日志格式**
```go
// 標(biāo)準(zhǔn)日志示例
common.LogInfof(ctx, "module_name", "method_name, param1=%v, param2=%v, resp=%s, err=%v",
param1, param2, respJson, err)
```
## 示例
<example>
// HTTP客戶端調(diào)用示例
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"
)
// 請求參數(shù)定義
type SearchParams struct {
Query string `json:"query"`
Latitude float64 `json:"latitude"`
Longitude float64 `json:"longitude"`
}
// 響應(yīng)結(jié)構(gòu)定義
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"`
}
// 發(fā)送搜索請求
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")
}
// 構(gòu)建請求參數(shù)
httpParams := map[string]string{
"query": params.Query,
"latitude": fmt.Sprintf("%f", params.Latitude),
"longitude": fmt.Sprintf("%f", params.Longitude),
"trace_id": eagleeye.GetTraceId(ctx),
}
// 構(gòu)建請求選項
opt := simplehttp.BuildOptions(&simplehttp.Options{
Method: "GET",
Params: httpParams,
Timeout: conf.Timeout,
HostWithVip: conf.VipHost,
RecordWithParams: httpcommon.RecordWithParams,
})
// 發(fā)送請求
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())
}
// 解析響應(yīng)
var response SearchResponse
if err := json.Unmarshal(respData, &response); err != nil {
return nil, common.Errorf(common.ErrorInternal, "unmarshal search response failed: %s", err.Error())
}
// 業(yè)務(wù)錯誤處理
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客戶端調(diào)用示例
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"
)
// 全局服務(wù)實例
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"
}
// 響應(yīng)模型
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"
}
// 產(chǎn)品詳情
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"
}
// 服務(wù)結(jié)構(gòu)體
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")
}
// 查詢產(chǎn)品信息
func GetProductDetail(ctx context.Context, productId string) (*ProductDetail, *common.Error) {
// 構(gòu)建請求
request := &ProductQueryRequest{
ProductId: productId,
Fields: []string{"id", "name", "price", "description"},
}
// 調(diào)用HSF服務(wù)
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())
}
// 結(jié)果處理
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")
}
// 帶緩存的產(chǎn)品查詢
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調(diào)用不規(guī)范
var badHsfService = struct {
Method func(args []interface{}) (interface{}, error)
}{}
// 錯誤1: 不遵循標(biāo)準(zhǔn)接口定義
// 錯誤2: 沒有使用Context參數(shù)
// 錯誤3: 沒有注冊模型
// 錯誤4: 錯誤處理不規(guī)范
func BadHsfCall(id string) interface{} {
result, _ := badHsfService.Method([]interface{}{id})
// 忽略錯誤處理
return result
}
</example>MCP
MCP作為cursor的加強(qiáng)工具，可以讓工作中更加流暢，可以在MCP廣場獲取你想要的 MCP廣場，比如我們可以在日常開發(fā)中飛書用的比較多，就可以使用飛書的mcp工具，實現(xiàn)文檔處理、對話管理、日程安排等多種自動化場景。這個過程是增量的，當(dāng)日常工作中有MCP的需求時，就可以搜索對應(yīng)的MCP服務(wù)并應(yīng)用到cursor中。
總結(jié)
根據(jù)團(tuán)隊成員的反饋，Cursor在提升編程效率方面基本達(dá)到了預(yù)期。現(xiàn)在，甚至項目經(jīng)理（PM）也開始使用Cursor編寫腳本，而不再像以前那樣依賴研發(fā)人員。當(dāng)然，Cursor并非萬能，面對一些復(fù)雜的大型需求時，其表現(xiàn)有點差強(qiáng)人意，在這一層面不知道最近大火的Claude Code和Argument Code的表現(xiàn)如何？
本文轉(zhuǎn)載自???AI 博物院??? 作者：longyunfeigu
?著作權(quán)歸作者所有，如需轉(zhuǎn)載，請注明出處，否則將追究法律責(zé)任
贊
收藏
回復(fù)
分享
微博
QQ
微信
舉報
回復(fù)
相關(guān)推薦
