python 利用flask+sqlalchemy 实现简朴数据查询接口

数据库表结构和部门数据

1. SET NAMES utf8mb4;
2. SET FOREIGN_KEY_CHECKS = 0;
4. -- ----------------------------
5. -- Table structure for user
6. -- ----------------------------
7. DROP TABLE IF EXISTS `user`;
8. CREATE TABLE `user`  (
9.   `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键ID',
10.   `name` varchar(60) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '用户名',
11.   `password` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '密码',
12.   `avatar` varchar(60) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '头像',
13.   `sex` smallint(6) NULL DEFAULT 0 COMMENT '性别: 0 未知 1 男 2 女',
14.   `phone` varchar(30) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL COMMENT '手机号',
15.   `is_deleted` smallint(6) NULL DEFAULT 0 COMMENT '是否删除: 0 未删除 1 已删除',
16.   `create_time` datetime NULL DEFAULT NULL COMMENT '创建时间',
17.   `update_time` datetime NULL DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
18.   PRIMARY KEY (`id`) USING BTREE
19. ) ENGINE = InnoDB AUTO_INCREMENT = 31 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = DYNAMIC;
21. -- ----------------------------
22. -- Records of user
23. -- ----------------------------
24. INSERT INTO `user` VALUES (1, 'admin', '30780cc6f2e56945aaf9c9578c932e22', '1', 0, NULL, 0, '2023-03-29 02:34:58', '2023-03-29 02:34:58');
25. INSERT INTO `user` VALUES (2, 'user', '30780cc6f2e56945aaf9c9578c932e22', '', 1, NULL, 0, '2023-03-29 02:34:58', '2023-03-29 02:34:58');
26. INSERT INTO `user` VALUES (3, 'guest', '30780cc6f2e56945aaf9c9578c932e22', '2', 2, NULL, 0, '2023-03-29 02:34:58', '2023-03-29 02:34:58');
27. INSERT INTO `user` VALUES (4, 'guest', '30780cc6f2e56945aaf9c9578c932e22', '2', 2, NULL, 0, '2023-03-29 02:34:58', '2023-03-29 02:34:58');
28. INSERT INTO `user` VALUES (5, 'guest', '30780cc6f2e56945aaf9c9578c932e22', '2', 2, NULL, 0, '2023-03-29 02:34:58', '2023-03-29 02:34:58');
30. SET FOREIGN_KEY_CHECKS = 1;

复制代码

python 代码

1. import logging
2. from datetime import datetime
3. from typing import List, Optional
5. import uvicorn
6. from fastapi import FastAPI, HTTPException, Query
7. from pydantic import BaseModel
8. from pydantic import create_model
9. from sqlalchemy import create_engine, Column, Integer, String, SmallInteger, DateTime, and_
10. from sqlalchemy.ext.declarative import declarative_base
11. from sqlalchemy.future import select
12. from sqlalchemy.orm import sessionmaker
14. # 配置 SQLAlchemy 日志，打印 SQL 语句
15. logging.basicConfig()
16. logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
18. # 修改为您的数据库连接信息
19. DATABASE_URL = "mysql://root:wonderful2021@127.0.0.1:3306/demo"
21. engine = create_engine(DATABASE_URL, echo=True)  # echo=True 打印 SQL 语句
22. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
24. Base = declarative_base()
27. class User(Base):
28.     __tablename__ = "user"
30.     id = Column(Integer, primary_key=True, index=True)
31.     name = Column(String(60), nullable=False)
32.     password = Column(String(64), nullable=False)
33.     avatar = Column(String(60), nullable=True)
34.     sex = Column(SmallInteger, default=0)
35.     phone = Column(String(30), nullable=True)
36.     is_deleted = Column(SmallInteger, default=0)
37.     create_time = Column(DateTime, default=datetime.utcnow)
38.     update_time = Column(DateTime, default=datetime.utcnow)
41. class UserResponse(BaseModel):
42.     id: int
43.     name: str
44.     avatar: Optional[str] = None
45.     sex: int
46.     phone: Optional[str] = None
47.     is_deleted: int
48.     create_time: datetime
49.     update_time: datetime
51.     class Config:
52.         orm_mode = True
53.         json_encoders = {
54.             datetime: lambda v: v.strftime('%Y-%m-%d %H:%M:%S')  # 自定义日期时间格式
55.         }
58. app = FastAPI()
60. # # 固定返回字段 字段与类型的映射
61. field_type_mapping = {
62.     'id': int,
63.     'name': str,
64.     'phone': str,
65.     'avatar': str,
66.     'create_time': str,  # 这里使用 str，因为我们会在返回时格式化
67.     'update_time': str  # 同上
68. }
70. # 动态生成 Pydantic 模型
71. DynamicUserResponse = create_model(
72.     'DynamicUserResponse',
73.     **{field: (field_type_mapping[field], ...) for field in field_type_mapping.keys()}
74. )
77. def get_selected_fields(model, fields):
78.     """根据字段列表动态选择模型字段"""
79.     return [getattr(model, field.strip()) for field in fields]
82. def handle_none_values(data: dict) -> dict:
83.     """处理返回值中的 None 值，转换为空字符串"""
84.     return {k: (v if v is not None else "") for k, v in data.items()}
87. def format_datetime(dt: Optional[datetime]) -> str:
88.     """格式化日期时间，返回字符串"""
89.     return dt.strftime('%Y-%m-%d %H:%M:%S') if dt else ""
92. @app.get("/users/", response_model=List[DynamicUserResponse])
93. async def read_users(
94.         skip: int = Query(0, ge=0),
95.         limit: int = Query(10, ge=1),
96.         name: Optional[str] = None,
97.         sex: Optional[int] = None,
98.         phone: Optional[str] = None
99. ):
100.     with SessionLocal() as session:
101.         filters = [User.is_deleted == 0]  # 默认条件：未删除
103.         if name:
104.             filters.append(User.name.like(f"%{name}%"))  # 模糊查询
105.         if sex is not None:
106.             filters.append(User.sex == sex)  # 精确查询
107.         if phone:
108.             filters.append(User.phone.like(f"%{phone}%"))  # 模糊查询
110.         # 固定返回字段
111.         selected_fields = get_selected_fields(User, field_type_mapping.keys())
112.         stmt = select(*selected_fields).where(and_(*filters)).offset(skip).limit(limit)
114.         result = session.execute(stmt)
115.         users = result.all()
117.         # 返回字典列表，处理 None 值并格式化日期时间
118.         return [
119.             handle_none_values({
120.                 **dict(zip(field_type_mapping.keys(), user)),
121.                 'create_time': format_datetime(user.create_time),
122.                 'update_time': format_datetime(user.update_time)
123.             }) for user in users
124.         ]
127. @app.get("/users/{user_id}", response_model=DynamicUserResponse)
128. async def read_user(
129.         user_id: int
130. ):
131.     with SessionLocal() as session:
132.         # 固定返回字段
133.         selected_fields = get_selected_fields(User, field_type_mapping.keys())
134.         stmt = select(*selected_fields).where(User.id == user_id, User.is_deleted == 0)
136.         result = session.execute(stmt)
137.         user = result.first()
139.         if user is None:
140.             raise HTTPException(status_code=404, detail="User not found")
142.         # 返回字典，处理 None 值并格式化日期时间
143.         return handle_none_values({
144.             **dict(zip(field_type_mapping.keys(), user)),
145.             'create_time': format_datetime(user.create_time),
146.             'update_time': format_datetime(user.update_time)
147.         })
150. @app.get("/usersallfiled/", response_model=List[UserResponse])
151. async def read_users(
152.         skip: int = Query(0, ge=0),
153.         limit: int = Query(10, ge=1),
154.         name: Optional[str] = None,
155.         sex: Optional[int] = None,
156.         phone: Optional[str] = None
157. ):
158.     with SessionLocal() as session:
159.         filters = [User.is_deleted == 0]  # 默认条件：未删除
161.         if name:
162.             filters.append(User.name.like(f"%{name}%"))  # 模糊查询
163.         if sex is not None:
164.             filters.append(User.sex == sex)  # 精确查询
165.         if phone:
166.             filters.append(User.phone.like(f"%{phone}%"))  # 模糊查询
168.         stmt = select(User).where(and_(*filters)).offset(skip).limit(limit)
170.         result = session.execute(stmt)
171.         users = result.scalars().all()
173.         return users
176. @app.get("/usersallfiled/{user_id}", response_model=UserResponse)
177. async def read_user(
178.         user_id: int
179. ):
180.     with SessionLocal() as session:
181.         stmt = select(User).where(User.id == user_id, User.is_deleted == 0)
183.         result = session.execute(stmt)
184.         user = result.scalars().first()
186.         if user is None:
187.             raise HTTPException(status_code=404, detail="User not found")
189.         return user
192. if __name__ == "__main__":
193.     uvicorn.run(app, host="127.0.0.1", port=8000)

复制代码

调用方式

  get请求http://127.0.0.1:8000/usersallfiled/?name=admin&skip=0&limit=10
返回效果[{"id":1,"name":"admin","avatar":"1","sex":0,"phone":null,"is_deleted":0,"create_time":"2023-03-29 02:34:58","update_time":"2023-03-29 02:34:58"}]

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。