commit e8b2a9a99cd40e55bf142aa624434e1dde468240 Author: liyuqi Date: Wed Jun 4 10:39:32 2025 +0800 尝试提交 diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..13566b8 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,8 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Editor-based HTTP Client requests +/httpRequests/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml diff --git a/.idea/inspectionProfiles/Project_Default.xml b/.idea/inspectionProfiles/Project_Default.xml new file mode 100644 index 0000000..6bc93ab --- /dev/null +++ b/.idea/inspectionProfiles/Project_Default.xml @@ -0,0 +1,12 @@ + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..105ce2d --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..0478804 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,4 @@ + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..25e529e --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/wms-py.iml b/.idea/wms-py.iml new file mode 100644 index 0000000..74d515a --- /dev/null +++ b/.idea/wms-py.iml @@ -0,0 +1,10 @@ + + + + + + + + + + \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..c9c005f --- /dev/null +++ b/README.md @@ -0,0 +1,258 @@ +# WMS FastAPI 项目 + +这是一个基于 FastAPI 的仓库管理系统(WMS)，采用标准的项目结构，支持多环境配置。 + +## 项目结构 + +``` +wms-py/ +├── app/ # 应用核心代码 +│ ├── __init__.py +│ ├── main.py # 主应用文件 +│ ├── config/ # 配置模块 +│ │ ├── __init__.py +│ │ └── settings.py # 配置管理 +│ ├── controllers/ # 控制器(路由) +│ │ ├── __init__.py +│ │ ├── base_controller.py # 基础接口 +│ │ ├── database_controller.py # 数据库测试接口 +│ │ └── location_controller.py # 库位管理接口 🆕 +│ ├── services/ # 业务服务层 🆕 +│ │ ├── __init__.py +│ │ └── location_service.py # 库位业务逻辑 +│ ├── models/ # 数据模型 🆕 +│ │ ├── __init__.py +│ │ └── location.py # 库位数据模型 +│ ├── schemas/ # Pydantic模式 🆕 +│ │ ├── __init__.py +│ │ ├── common.py # 通用响应模式 +│ │ └── location.py # 库位请求/响应模式 +│ └── utils/ # 工具类 +│ ├── __init__.py +│ ├── database.py # 数据库工具类 +│ └── string_utils.py # 字符串工具 🆕 +├── config/ # 环境配置文件 +│ ├── development.yaml # 开发环境配置 +│ └── production.yaml # 生产环境配置 +├── requirements.txt # 依赖包列表 +├── run.py # 启动脚本 +├── create_tables.py # 数据库表创建脚本 🆕 +└── README.md # 项目说明 +``` + +## 功能特性 + +- ✅ 多环境配置支持 (开发/生产) +- ✅ YAML 配置文件 +- ✅ 数据库连接管理 +- ✅ 标准项目结构 +- ✅ API 文档自动生成 +- ✅ 数据库测试接口 +- ✅ 健康检查接口 +- ✅ **库位管理模块** 🆕 + - 批量生成库位 + - 查询已使用库位数量 + - 库位信息查询 + - 库位统计功能 + +## 快速开始 + +### 1. 安装依赖 + +```bash +pip install -r requirements.txt +``` + +### 2. 配置数据库 + +根据需要修改配置文件： + +**开发环境** (`config/development.yaml`): +```yaml +database: + host: "localhost" + port: 3306 + username: "root" + password: "root" + database: "wms_ntyc" +``` + +### 3. 创建数据库表 + +```bash +python create_tables.py +``` + +### 4. 运行应用 + +```bash +# 开发环境 +python run.py + +# 生产环境 +set ENVIRONMENT=production && python run.py +``` + +## API接口 + +### 基础接口 +- `GET /` - 系统信息 +- `GET /health` - 健康检查 +- `GET /config` - 配置信息 + +### 数据库测试接口 +- `GET /database/test` - 测试数据库连接 +- `GET /database/tables` - 获取所有表名 +- `GET /database/query?sql=SELECT * FROM table_name` - 执行SQL查询 +- `GET /database/table/{table_name}` - 查看指定表数据 + +### 库位管理接口 🆕 + +**库位生成**: +- `POST /wms/location/genLocations` - 批量生成库位 + +请求示例： +```json +{ + "l_row": 5, + "l_col": 8, + "l_layer": 6, + "l_depth": 2, + "sub_area": "A", + "area_id": 1 +} +``` + +**库位查询**: +- `GET /wms/location/getUsedLocations?equipment_id=1&location_type=1` - 获取已使用库位数量 +- `GET /wms/location/location/{location_id}` - 根据ID查询库位信息 +- `GET /wms/location/locations/area/{area_id}` - 根据区域查询库位列表 +- `GET /wms/location/locations/stats` - 获取库位统计信息 + +### API 文档 +- `GET /docs` - Swagger UI 文档 +- `GET /redoc` - ReDoc 文档 + +## 库位管理功能详解 + +### 1. 库位生成逻辑 + +库位ID格式：`{子区域}{行号02d}-{列号02d}-{层号02d}-{深度02d}` + +例如： +- 子区域: A +- 行: 1, 列: 2, 层: 3, 深度: 1 +- 生成ID: `A01-02-03-01` + +### 2. 数据库表结构 + +```sql +CREATE TABLE t_app_location ( + location_id VARCHAR(50) PRIMARY KEY COMMENT '库位ID', + location_name VARCHAR(100) DEFAULT '' COMMENT '库位名称', + location_type INT DEFAULT 1 COMMENT '库位类型', + is_occupy INT DEFAULT 0 COMMENT '是否占用 0-未占用 1-已占用', + is_enable INT DEFAULT 0 COMMENT '是否启用', + equipment_id INT DEFAULT 0 COMMENT '设备ID', + aisle_num_left INT COMMENT '左侧巷道号', + aisle_num_right INT COMMENT '右侧巷道号', + l_row INT COMMENT '行', + l_col INT COMMENT '列', + l_layer INT COMMENT '层', + l_depth INT COMMENT '深度', + remark TEXT DEFAULT '' COMMENT '备注', + area_id INT DEFAULT 1 COMMENT '区域ID', + sub_area VARCHAR(50) DEFAULT '' COMMENT '子区域' +); +``` + +### 3. 使用示例 + +**生成库位**： +```bash +curl -X POST "http://localhost:12315/wms/location/genLocations" \ + -H "Content-Type: application/json" \ + -d '{ + "l_row": 2, + "l_col": 3, + "l_layer": 4, + "l_depth": 1, + "sub_area": "A", + "area_id": 1 + }' +``` + +**查询统计**： +```bash +curl "http://localhost:12315/wms/location/locations/stats" +``` + +## 项目架构说明 + +### 分层架构 + +1. **Controller层** (`app/controllers/`) + - 处理HTTP请求 + - 参数验证 + - 路由定义 + +2. **Service层** (`app/services/`) + - 业务逻辑处理 + - 事务管理 + - 数据操作 + +3. **Model层** (`app/models/`) + - 数据库表映射 + - ORM模型定义 + +4. **Schema层** (`app/schemas/`) + - 请求/响应数据验证 + - API文档生成 + +### 依赖注入 + +项目使用FastAPI的依赖注入系统： +- 数据库会话注入 +- 服务类注入 +- 配置注入 + +### 对应关系 + +| Java概念 | Python/FastAPI概念 | 文件位置 | +|---------|-------------------|----------| +| @RestController | APIRouter | `app/controllers/` | +| @Service | Service类 | `app/services/` | +| @Entity | SQLAlchemy Model | `app/models/` | +| DTO | Pydantic Schema | `app/schemas/` | +| @Autowired | Depends() | 依赖注入 | + +## IDE 配置 + +### PyCharm 配置 + +创建运行配置： +- **Name**: `WMS Development` +- **Script path**: `run.py` +- **Environment variables**: `ENVIRONMENT=development` + +## 注意事项 + +1. 确保数据库服务已启动 +2. 首次运行前执行 `python create_tables.py` 创建表 +3. 修改配置文件后需要重启应用 +4. 库位生成前请确认参数，避免重复数据 + +## 故障排除 + +### 常见问题 + +1. **表不存在** + - 运行 `python create_tables.py` 创建表 + +2. **库位ID重复** + - 检查参数是否与已有数据冲突 + - 清理测试数据后重新生成 + +3. **数据库连接失败** + - 检查配置文件中的数据库信息 + - 确认数据库服务状态 \ No newline at end of file diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..205a0f3 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,3 @@ +from .config import settings + +# WMS FastAPI 应用包 \ No newline at end of file diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..1f51e10 --- /dev/null +++ b/app/config.py @@ -0,0 +1,85 @@ +import os +from pathlib import Path +import yaml +from pydantic import BaseModel, computed_field, Field +from dotenv import load_dotenv + +# 加载环境变量 +load_dotenv() + + +class DatabaseConfig(BaseModel): + driver: str = "mysql+pymysql" + host: str + port: int = 3306 + name: str + username: str + password: str + params: str = "characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowMultiQueries=true&rewriteBatchedStatements=true" + + @computed_field + @property + def url(self) -> str: + """生成数据库连接 URL""" + return f"{self.driver}://{self.username}:{self.password}@{self.host}:{self.port}/{self.name}?{self.params}" + + +class ServerConfig(BaseModel): + port: int + context_path: str = "/" + + +class AppConfig(BaseModel): + name: str = "wms_main" + max_file_size: int = 100 * 1024 * 1024 # 100MB + max_request_size: int = 1000 * 1024 * 1024 # 1000MB + + +class Settings(BaseModel): + """全局配置模型""" + app: AppConfig + server: ServerConfig + database: DatabaseConfig + env: str = Field(default="development") # 当前环境名称 + + @classmethod + def from_env(cls): + """从环境变量加载配置""" + env_name = os.getenv("ENV", "development") + + # 获取项目根目录 + base_dir = Path(__file__).resolve().parent.parent + + # 加载基础配置 + base_path = base_dir / "config" / "base.yaml" + base_data = {} + if base_path.exists(): + with open(base_path, "r") as f: + base_data = yaml.safe_load(f) or {} + + # 加载环境特定配置 + env_path = base_dir / "config" / f"{env_name}.yaml" + env_data = {} + if env_path.exists(): + with open(env_path, "r") as f: + env_data = yaml.safe_load(f) or {} + + # 深度合并配置 + def deep_merge(base: dict, update: dict) -> dict: + """递归合并两个字典""" + for key, value in update.items(): + if isinstance(value, dict) and key in base and isinstance(base[key], dict): + base[key] = deep_merge(base[key], value) + else: + base[key] = value + return base + + # 合并配置 + merged_config = deep_merge(base_data, env_data) + merged_config["env"] = env_name # 添加环境名称 + + return cls(**merged_config) + + +# 创建全局配置对象 +settings = Settings.from_env() diff --git a/app/config/__init__.py b/app/config/__init__.py new file mode 100644 index 0000000..d157510 --- /dev/null +++ b/app/config/__init__.py @@ -0,0 +1 @@ +# 配置包 \ No newline at end of file diff --git a/app/config/settings.py b/app/config/settings.py new file mode 100644 index 0000000..c8fb5fc --- /dev/null +++ b/app/config/settings.py @@ -0,0 +1,71 @@ +import os +import yaml +from typing import Dict, Any +from pydantic import BaseModel + + +class DatabaseConfig(BaseModel): + """数据库配置""" + host: str + port: int = 3306 + username: str + password: str + database: str + charset: str = "utf8mb4" + + +class AppConfig(BaseModel): + """应用配置""" + name: str + debug: bool = False + host: str = "localhost" + port: int = 12315 + database: DatabaseConfig + + +class Settings: + """配置管理器""" + + def __init__(self): + self._config: AppConfig = None + self._env = os.getenv("ENVIRONMENT", "development") + self.load_config() + + def load_config(self): + """加载配置文件""" + config_file = f"config/{self._env}.yaml" + + if not os.path.exists(config_file): + raise FileNotFoundError(f"配置文件不存在: {config_file}") + + with open(config_file, 'r', encoding='utf-8') as f: + config_data = yaml.safe_load(f) + + # 解析数据库配置 + db_config = DatabaseConfig(**config_data['database']) + + # 解析应用配置 + app_data = config_data['app'] + app_data['database'] = db_config + + self._config = AppConfig(**app_data) + + @property + def config(self) -> AppConfig: + """获取配置""" + return self._config + + @property + def environment(self) -> str: + """获取当前环境""" + return self._env + + @property + def database_url(self) -> str: + """获取数据库连接URL""" + db = self._config.database + return f"mysql+pymysql://{db.username}:{db.password}@{db.host}:{db.port}/{db.database}?charset={db.charset}" + + +# 全局配置实例 +settings = Settings() \ No newline at end of file diff --git a/app/controllers/__init__.py b/app/controllers/__init__.py new file mode 100644 index 0000000..6127e45 --- /dev/null +++ b/app/controllers/__init__.py @@ -0,0 +1 @@ +# 控制器包 \ No newline at end of file diff --git a/app/controllers/base_controller.py b/app/controllers/base_controller.py new file mode 100644 index 0000000..7b0b978 --- /dev/null +++ b/app/controllers/base_controller.py @@ -0,0 +1,37 @@ +from fastapi import APIRouter +from app.config.settings import settings + +router = APIRouter(tags=["基础接口"]) + + +@router.get("/") +async def root(): + """根路径""" + return { + "message": "WMS FastAPI 系统", + "app_name": settings.config.name, + "environment": settings.environment, + "debug": settings.config.debug, + "version": "1.0.0" + } + + +@router.get("/config") +async def get_config(): + """获取配置信息""" + return { + "app": { + "name": settings.config.name, + "debug": settings.config.debug, + "host": settings.config.host, + "port": settings.config.port + }, + "environment": settings.environment, + "database": { + "host": settings.config.database.host, + "port": settings.config.database.port, + "database": settings.config.database.database, + "username": settings.config.database.username, + "charset": settings.config.database.charset + } + } diff --git a/app/controllers/database_controller.py b/app/controllers/database_controller.py new file mode 100644 index 0000000..beeb077 --- /dev/null +++ b/app/controllers/database_controller.py @@ -0,0 +1,44 @@ +from fastapi import APIRouter, Depends, Query +from sqlalchemy.orm import Session +from app.utils.database import get_db, DatabaseUtils +from typing import Optional + +router = APIRouter(prefix="/database", tags=["数据库测试"]) + + +@router.get("/test") +async def test_database_connection(): + """测试数据库连接""" + return DatabaseUtils.test_connection() + + +@router.get("/tables") +async def list_all_tables(): + """获取所有表名""" + return DatabaseUtils.get_all_tables() + + +@router.get("/query") +async def execute_query( + sql: str = Query(..., description="要执行的SQL查询语句"), + limit: int = Query(100, description="返回记录数限制", ge=1, le=1000) +): + """执行SQL查询""" + return DatabaseUtils.execute_query(sql, limit) + + +@router.get("/table/{table_name}") +async def list_table_data( + table_name: str, + limit: int = Query(100, description="返回记录数限制", ge=1, le=1000) +): + """查看指定表的数据""" + sql = f"SELECT * FROM {table_name}" + return DatabaseUtils.execute_query(sql, limit) + + +@router.get("/table/{table_name}/structure") +async def get_table_structure(table_name: str): + """获取表结构""" + sql = f"DESCRIBE {table_name}" + return DatabaseUtils.execute_query(sql, 1000) \ No newline at end of file diff --git a/app/controllers/location_controller.py b/app/controllers/location_controller.py new file mode 100644 index 0000000..11de217 --- /dev/null +++ b/app/controllers/location_controller.py @@ -0,0 +1,31 @@ +from fastapi import APIRouter, Depends, Query, HTTPException +from sqlalchemy.orm import Session +from typing import List +from app.utils.database import get_db +from app.services.location_service import LocationService +from app.schemas.location import LocationQueryRequest +from app.schemas.common import BaseWmsApiResponse, WmsApiResponse +from app.models.location import TAppLocation + +router = APIRouter(prefix="/wms/location", tags=["库位管理"]) + + +def get_location_service(db: Session = Depends(get_db)) -> LocationService: + return LocationService(db) + + +@router.post("/genLocations", response_model=BaseWmsApiResponse) +async def gen_locations( + location_query: LocationQueryRequest, + location_service: LocationService = Depends(get_location_service) +) -> BaseWmsApiResponse: + return location_service.gen_locations(location_query) + + +@router.get("/getUsedLocations", response_model=WmsApiResponse) +async def get_used_locations( + equipment_id: int = Query(..., description="设备ID"), + location_type: int = Query(..., description="库位类型"), + location_service: LocationService = Depends(get_location_service) +) -> WmsApiResponse: + return location_service.get_used_locations(equipment_id, location_type) \ No newline at end of file diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..523e591 --- /dev/null +++ b/app/main.py @@ -0,0 +1,50 @@ +from fastapi import FastAPI +from app.config.settings import settings +from app.utils.database import init_database, close_database +from app.controllers import base_controller, database_controller, location_controller + +# 创建FastAPI应用 +app = FastAPI( + title=settings.config.name, + description="仓库管理系统 API", + version="1.0.0", + debug=settings.config.debug +) + + +@app.on_event("startup") +async def startup_event(): + """应用启动事件""" + print("=" * 50) + print(f"🚀 启动 {settings.config.name}") + print(f"📦 环境: {settings.environment}") + print(f"🔧 调试模式: {settings.config.debug}") + print(f"🌐 地址: http://{settings.config.host}:{settings.config.port}") + print("=" * 50) + + # 初始化数据库 + init_database() + + +@app.on_event("shutdown") +async def shutdown_event(): + """应用关闭事件""" + print("🔄 正在关闭应用...") + close_database() + print("✅ 应用已关闭") + + +# 注册路由 +app.include_router(base_controller.router) +app.include_router(database_controller.router) +app.include_router(location_controller.router) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run( + "app.main:app", + host=settings.config.host, + port=settings.config.port, + reload=settings.config.debug + ) diff --git a/app/models/__init__.py b/app/models/__init__.py new file mode 100644 index 0000000..f3ec3a8 --- /dev/null +++ b/app/models/__init__.py @@ -0,0 +1 @@ +# 数据模型包 \ No newline at end of file diff --git a/app/models/location.py b/app/models/location.py new file mode 100644 index 0000000..e8eda82 --- /dev/null +++ b/app/models/location.py @@ -0,0 +1,45 @@ +from sqlalchemy import Column, String, Integer, Text +from app.utils.database import Base + + +class TAppLocation(Base): + __tablename__ = "t_app_location" + + location_id = Column(String(50), primary_key=True, comment="库位ID") + location_name = Column(String(100), default="", comment="库位名称") + location_type = Column(Integer, default=1, comment="库位类型") + is_occupy = Column(Integer, default=0, comment="是否占用 0-未占用 1-已占用") + is_enable = Column(Integer, default=0, comment="是否启用") + equipment_id = Column(Integer, default=0, comment="设备ID") + aisle_num_left = Column(Integer, comment="左侧巷道号") + aisle_num_right = Column(Integer, comment="右侧巷道号") + l_row = Column(Integer, comment="行") + l_col = Column(Integer, comment="列") + l_layer = Column(Integer, comment="层") + l_depth = Column(Integer, comment="深度") + remark = Column(Text, default="", comment="备注") + area_id = Column(Integer, default=1, comment="区域ID") + sub_area = Column(String(50), default="", comment="子区域") + + def __init__(self, location_id=None, location_name="", location_type=1, + is_occupy=0, is_enable=0, equipment_id=0, aisle_num_left=None, + aisle_num_right=None, l_row=None, l_col=None, l_layer=None, + l_depth=None, remark="", area_id=1, sub_area=""): + self.location_id = location_id + self.location_name = location_name + self.location_type = location_type + self.is_occupy = is_occupy + self.is_enable = is_enable + self.equipment_id = equipment_id + self.aisle_num_left = aisle_num_left + self.aisle_num_right = aisle_num_right + self.l_row = l_row + self.l_col = l_col + self.l_layer = l_layer + self.l_depth = l_depth + self.remark = remark + self.area_id = area_id + self.sub_area = sub_area + + def __repr__(self): + return f"" \ No newline at end of file diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py new file mode 100644 index 0000000..859408a --- /dev/null +++ b/app/schemas/__init__.py @@ -0,0 +1 @@ +# Pydantic模式包 \ No newline at end of file diff --git a/app/schemas/common.py b/app/schemas/common.py new file mode 100644 index 0000000..c914931 --- /dev/null +++ b/app/schemas/common.py @@ -0,0 +1,41 @@ +from pydantic import BaseModel +from typing import Any, Optional + + +class BaseWmsApiResponse(BaseModel): + code: int + message: str + + @staticmethod + def success(message: str = "操作成功"): + return BaseWmsApiResponse( + code=200, + message=message + ) + + @staticmethod + def error(message: str = "操作失败", code: int = 500): + return BaseWmsApiResponse( + code=code, + message=message + ) + + +class WmsApiResponse(BaseWmsApiResponse): + data: Optional[Any] = None + + @staticmethod + def success(data: Any = None, message: str = "操作成功"): + return WmsApiResponse( + code=200, + message=message, + data=data + ) + + @staticmethod + def error(message: str = "操作失败", code: int = 500, data: Any = None): + return WmsApiResponse( + code=code, + message=message, + data=data + ) diff --git a/app/schemas/location.py b/app/schemas/location.py new file mode 100644 index 0000000..e5aa227 --- /dev/null +++ b/app/schemas/location.py @@ -0,0 +1,23 @@ +from pydantic import BaseModel, Field +from typing import Optional + + +class LocationQueryRequest(BaseModel): + l_row: int = Field(..., description="行数", ge=1, le=100) + l_col: int = Field(..., description="列数", ge=1, le=100) + l_layer: int = Field(..., description="层数", ge=1, le=50) + l_depth: int = Field(..., description="深度", ge=1, le=50) + sub_area: str = Field(..., description="子区域标识", min_length=1, max_length=10) + area_id: Optional[int] = Field(1, description="区域ID") + + class Config: + json_schema_extra = { + "example": { + "l_row": 5, + "l_col": 8, + "l_layer": 6, + "l_depth": 2, + "sub_area": "A", + "area_id": 1 + } + } \ No newline at end of file diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..2f90aee --- /dev/null +++ b/app/services/__init__.py @@ -0,0 +1 @@ +# 业务服务包 \ No newline at end of file diff --git a/app/services/location_service.py b/app/services/location_service.py new file mode 100644 index 0000000..e98f73f --- /dev/null +++ b/app/services/location_service.py @@ -0,0 +1,89 @@ +from sqlalchemy.orm import Session +from sqlalchemy import func +from typing import List +import logging + +from app.models.location import TAppLocation +from app.schemas.location import LocationQueryRequest +from app.schemas.common import BaseWmsApiResponse, WmsApiResponse +from app.utils.string_utils import pad_left + +logger = logging.getLogger(__name__) + + +class LocationService: + + def __init__(self, db: Session): + self.db = db + + def gen_locations(self, location_query: LocationQueryRequest) -> BaseWmsApiResponse: + try: + new_location_list: List[TAppLocation] = [] + + for row in range(1, location_query.l_row + 1): + for col in range(1, location_query.l_col + 1): + for layer in range(1, location_query.l_layer + 1): + for depth in range(1, location_query.l_depth + 1): + # 生成库位ID，格式：子区域 + 行(2位) + "-" + 列(2位) + "-" + 层(2位) + "-" + 深度(2位) + location_id = ( + f"{location_query.sub_area}" + f"{pad_left(str(row), '0', 2)}-" + f"{pad_left(str(col), '0', 2)}-" + f"{pad_left(str(layer), '0', 2)}-" + f"{pad_left(str(depth), '0', 2)}" + ) + + new_location = TAppLocation( + location_id=location_id, + location_name="", + location_type=1, + is_occupy=0, + is_enable=0, + equipment_id=0, + aisle_num_left=(row + 1) // 2, + aisle_num_right=(row + 1) // 2, + l_row=row, + l_col=col, + l_layer=layer, + l_depth=depth, + remark="", + area_id=location_query.area_id if location_query.area_id is not None else 1, + sub_area=location_query.sub_area + ) + new_location_list.append(new_location) + + self.db.add_all(new_location_list) + self.db.commit() + + logger.info(f"成功创建了{len(new_location_list)}个库位") + return BaseWmsApiResponse.success(f"成功创建了{len(new_location_list)}个库位。") + + except Exception as e: + self.db.rollback() + logger.error(f"创建库位失败: {str(e)}") + return BaseWmsApiResponse.error(f"创建库位失败: {str(e)}") + + def get_used_locations(self, equipment_id: int, location_type: int) -> WmsApiResponse: + try: + count = self.db.query(func.count(TAppLocation.location_id)).filter( + TAppLocation.equipment_id == equipment_id, + TAppLocation.location_type == location_type, + TAppLocation.is_occupy == 1 + ).scalar() + + # 确保count不为None + if count is None: + count = 0 + + return WmsApiResponse.success(data=count, message="查询成功") + + except Exception as e: + # 打印完整异常信息（包含堆栈） + logger.exception("查询已使用库位时发生异常") + + # 返回错误响应 - 也使用直接实例化 + return WmsApiResponse.error( + code=500, + message=f"查询失败: {type(e).__name__}", + data=None + ) diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 0000000..c6a3a2e --- /dev/null +++ b/app/utils/__init__.py @@ -0,0 +1 @@ +# 工具包 \ No newline at end of file diff --git a/app/utils/database.py b/app/utils/database.py new file mode 100644 index 0000000..442a9e7 --- /dev/null +++ b/app/utils/database.py @@ -0,0 +1,123 @@ +from sqlalchemy import create_engine, text +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker, Session +from sqlalchemy.pool import QueuePool +from app.config.settings import settings +from typing import Generator + +# 数据库引擎 +engine = create_engine( + settings.database_url, + poolclass=QueuePool, + pool_size=10, + max_overflow=20, + pool_pre_ping=True, + pool_recycle=3600, + echo=settings.config.debug # 在调试模式下打印SQL +) + +# 会话工厂 +SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) + +# 基础模型类 +Base = declarative_base() + + +def get_db() -> Generator[Session, None, None]: + db = SessionLocal() + try: + yield db + finally: + db.close() + + +class DatabaseUtils: + """数据库工具类""" + + @staticmethod + def test_connection() -> dict: + """测试数据库连接""" + try: + with engine.connect() as conn: + result = conn.execute(text("SELECT 1 as test")) + row = result.fetchone() + return { + "status": "success", + "message": "数据库连接正常", + "test_result": row[0] if row else None, + "database_url": settings.database_url.replace(settings.config.database.password, "****") + } + except Exception as e: + return { + "status": "error", + "message": f"数据库连接失败: {str(e)}" + } + + @staticmethod + def get_all_tables() -> dict: + """获取所有表名""" + try: + with engine.connect() as conn: + result = conn.execute(text("SHOW TABLES")) + tables = [row[0] for row in result.fetchall()] + return { + "status": "success", + "tables": tables, + "count": len(tables) + } + except Exception as e: + return { + "status": "error", + "message": f"获取表名失败: {str(e)}" + } + + @staticmethod + def execute_query(query: str, limit: int = 100) -> dict: + """执行查询语句""" + try: + with engine.connect() as conn: + # 添加LIMIT限制 + if "LIMIT" not in query.upper(): + query = f"{query} LIMIT {limit}" + + result = conn.execute(text(query)) + + # 获取列名 + columns = list(result.keys()) if result.keys() else [] + + # 获取数据 + rows = result.fetchall() + data = [dict(zip(columns, row)) for row in rows] + + return { + "status": "success", + "columns": columns, + "data": data, + "count": len(data) + } + except Exception as e: + return { + "status": "error", + "message": f"查询执行失败: {str(e)}" + } + + +# 初始化数据库连接 +def init_database(): + """初始化数据库""" + try: + # 测试连接 + test_result = DatabaseUtils.test_connection() + if test_result["status"] == "success": + print(f"✓ 数据库连接成功 - 环境: {settings.environment}") + print(f"✓ 数据库: {settings.config.database.host}:{settings.config.database.port}/{settings.config.database.database}") + else: + print(f"✗ 数据库连接失败: {test_result['message']}") + except Exception as e: + print(f"✗ 数据库初始化错误: {str(e)}") + + +def close_database(): + """关闭数据库连接""" + engine.dispose() + print("数据库连接已关闭") \ No newline at end of file diff --git a/app/utils/string_utils.py b/app/utils/string_utils.py new file mode 100644 index 0000000..6206aef --- /dev/null +++ b/app/utils/string_utils.py @@ -0,0 +1,14 @@ +def pad_left(text: str, pad_char: str = "0", length: int = 2) -> str: + return text.rjust(length, pad_char) + + +def pad_right(text: str, pad_char: str = " ", length: int = 10) -> str: + return text.ljust(length, pad_char) + + +def is_empty(text: str) -> bool: + return text is None or text.strip() == "" + + +def is_not_empty(text: str) -> bool: + return not is_empty(text) diff --git a/config/base.yaml b/config/base.yaml new file mode 100644 index 0000000..305b71e --- /dev/null +++ b/config/base.yaml @@ -0,0 +1,12 @@ +app: + name: wms_main + max_file_size: 104857600 # 100MB + max_request_size: 1048576000 # 1000MB + +server: + context_path: "/" + +database: + driver: mysql+pymysql + port: 3306 + params: characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowMultiQueries=true&rewriteBatchedStatements=true \ No newline at end of file diff --git a/config/development.yaml b/config/development.yaml new file mode 100644 index 0000000..e627f6c --- /dev/null +++ b/config/development.yaml @@ -0,0 +1,13 @@ +app: + name: "wms_main" + debug: true + host: "localhost" + port: 12315 + +database: + host: "localhost" + port: 3306 + username: "root" + password: "root" + database: "wms_ntyc" + charset: "utf8mb4" \ No newline at end of file diff --git a/config/production.yaml b/config/production.yaml new file mode 100644 index 0000000..a17b2d1 --- /dev/null +++ b/config/production.yaml @@ -0,0 +1,13 @@ +app: + name: "wms_main" + debug: false + host: "localhost" + port: 12315 + +database: + host: "10.18.58.21" + port: 3306 + username: "user" + password: "user" + database: "wms_yachi_nantong" + charset: "utf8mb4" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1769e16 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +fastapi +uvicorn +pydantic +pyyaml +python-dotenv +sqlalchemy +pymysql +python-multipart \ No newline at end of file diff --git a/run.py b/run.py new file mode 100644 index 0000000..6e0b1cd --- /dev/null +++ b/run.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +""" +WMS FastAPI 应用启动脚本 +支持通过环境变量切换开发/生产环境 +""" +import os +import sys +import uvicorn +from pathlib import Path + +# 添加项目根目录到Python路径 +project_root = Path(__file__).parent +sys.path.insert(0, str(project_root)) + + +def main(): + """主函数""" + # 获取环境变量，默认为开发环境 + env = os.getenv("ENVIRONMENT", "development") + + print(f"🔧 当前环境: {env}") + print(f"📁 项目根目录: {project_root}") + + # 设置环境变量 + os.environ["ENVIRONMENT"] = env + + try: + from app.config.settings import settings + + print(f"🚀 启动FastAPI应用...") + print(f"📦 应用名称: {settings.config.name}") + print(f"🌐 访问地址: http://{settings.config.host}:{settings.config.port}") + print(f"📚 API文档: http://{settings.config.host}:{settings.config.port}/docs") + + # 根据环境设置不同的启动参数 + if env == "development": + uvicorn.run( + "app.main:app", + host=settings.config.host, + port=settings.config.port, + reload=True, + log_level="debug" + ) + else: + uvicorn.run( + "app.main:app", + host=settings.config.host, + port=settings.config.port, + reload=False, + workers=4 + ) + + except Exception as e: + print(f"❌ 启动失败: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/test_main.http b/test_main.http new file mode 100644 index 0000000..a2d81a9 --- /dev/null +++ b/test_main.http @@ -0,0 +1,11 @@ +# Test your FastAPI endpoints + +GET http://127.0.0.1:8000/ +Accept: application/json + +### + +GET http://127.0.0.1:8000/hello/User +Accept: application/json + +### diff --git a/test_setup.py b/test_setup.py new file mode 100644 index 0000000..b75bcd4 --- /dev/null +++ b/test_setup.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +""" +快速测试脚本，验证项目配置是否正确 +""" +import os +import sys +from pathlib import Path + +# 添加项目根目录到Python路径 +project_root = Path(__file__).parent +sys.path.insert(0, str(project_root)) + + +def test_config(): + """测试配置加载""" + print("🔧 测试配置加载...") + + # 测试开发环境配置 + os.environ["ENVIRONMENT"] = "development" + try: + from app.config.settings import settings + print(f"✅ 开发环境配置加载成功") + print(f" 应用名称: {settings.config.name}") + print( + f" 数据库: {settings.config.database.host}:{settings.config.database.port}/{settings.config.database.database}") + print(f" 调试模式: {settings.config.debug}") + except Exception as e: + print(f"❌ 开发环境配置加载失败: {e}") + return False + + return True + + +def test_database(): + """测试数据库连接""" + print("\n🔍 测试数据库连接...") + + try: + from app.utils.database import DatabaseUtils + result = DatabaseUtils.test_connection() + + if result["status"] == "success": + print(f"✅ 数据库连接成功") + print(f" 测试结果: {result['test_result']}") + + # 测试获取表名 + tables_result = DatabaseUtils.get_all_tables() + if tables_result["status"] == "success": + print(f"✅ 获取表名成功，共 {tables_result['count']} 张表") + if tables_result["tables"]: + print(f" 前5张表: {tables_result['tables'][:5]}") + else: + print(f"⚠️ 获取表名失败: {tables_result['message']}") + + else: + print(f"❌ 数据库连接失败: {result['message']}") + return False + + except Exception as e: + print(f"❌ 数据库测试失败: {e}") + return False + + return True + + +def test_app_start(): + """测试应用是否可以正常启动""" + print("\n🚀 测试应用启动...") + + try: + from app.main import app + print(f"✅ FastAPI应用创建成功") + print(f" 应用标题: {app.title}") + print(f" 路由数量: {len(app.routes)}") + + # 检查路由 + routes = [route.path for route in app.routes if hasattr(route, 'path')] + print(f" 主要路由: {routes}") + + except Exception as e: + print(f"❌ 应用启动测试失败: {e}") + return False + + return True + + +def main(): + """主测试函数""" + print("=" * 50) + print("🧪 WMS FastAPI 项目配置测试") + print("=" * 50) + + tests = [ + ("配置加载", test_config), + ("数据库连接", test_database), + ("应用启动", test_app_start) + ] + + passed = 0 + failed = 0 + + for test_name, test_func in tests: + try: + if test_func(): + passed += 1 + else: + failed += 1 + except Exception as e: + print(f"❌ {test_name}测试异常: {e}") + failed += 1 + + print("\n" + "=" * 50) + print(f"📊 测试结果: 通过 {passed}, 失败 {failed}") + + if failed == 0: + print("🎉 所有测试通过！项目配置正确，可以启动应用。") + print("\n💡 启动命令:") + print(" python run.py") + print("\n📚 访问文档:") + print(" http://localhost:12315/docs") + else: + print("⚠️ 部分测试失败，请检查配置。") + + print("=" * 50) + + +if __name__ == "__main__": + main()