fastapi接口

admin

==========================================硬件应用部分代码======================================================
mind+接入数据部分
--------------------------------------------------------------------------------------------python模式代码参数部分-------------------------------------------------------------------------------------------------------------------
U_GET:注意需要使用''引起来才行
'http://192.168.3.55/users/'
http://192.168.3.55/users/

U_POST:
网址：'http://192.168.3.55/users/7/items/'
字典：{   "title": "8253",   "description": "825" }






======================================后台部分代码==================================================



APP为根目录下
db目录下有其他个文件。

APP_FASTAPI_825

1. # 第三方类库导入
   
2. from typing import List
   
3. from fastapi import Depends, FastAPI, HTTPException
   
4. from sqlalchemy.orm import Session
   
5. import uvicorn
   
6. # 自定义类导入
   
7. from db import crud, models, schemas
   
8. from db.database import SessionLocal, engine
   
9. 
   
10. models.Base.metadata.create_all(bind=engine)
    
11. app = FastAPI()
    
12. 
    
13. 
    
14. # 依赖
    
15. def get_db():
    
16.     try:
    
17.         db = SessionLocal()
    
18.         yield db
    
19.     finally:
    
20.         db.close()
    
21. 
    
22. 
    
23. @app.post("/users/", response_model=schemas.User)
    
24. def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    
25.     """根据email查找用户"""
    
26.     db_user = crud.get_user_by_email(db, email=user.email)
    
27.     # 如果用户存在，提示该邮箱已经被注册
    
28.     if db_user:
    
29.         raise HTTPException(status_code=400, detail="邮箱已经被注册！")
    
30.     # 返回创建的user对象
    
31.     return crud.create_user(db=db, user=user)
    
32. 
    
33. 
    
34. @app.get("/users/", response_model=List[schemas.User])
    
35. def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    
36.     """读取指定数量用户，默认0-100条数据"""
    
37.     users = crud.get_users(db, skip=skip, limit=limit)
    
38.     return users
    
39. 
    
40. 
    
41. @app.get("/users/{user_id}", response_model=schemas.User)
    
42. def read_user(user_id: int, db: Session = Depends(get_db)):
    
43.     """获取当前id的用户信息"""
    
44.     db_user = crud.get_user(db, user_id=user_id)
    
45.     """如果没有信息，提示用户不存在"""
    
46.     if db_user is None:
    
47.         raise HTTPException(status_code=404, detail="用户不存在")
    
48.     return db_user
    
49. 
    
50. # 根据用户id增加item数据表项目数据
    
51. @app.post("/users/{user_id}/items/", response_model=schemas.Item)
    
52. def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
    
53.     """创建该用户的items，用户关联项目信息"""
    
54.     return crud.create_user_item(db=db, item=item, user_id=user_id)
    
55. 
    
56. 
    
57. @app.get("/items/", response_model=List[schemas.Item])
    
58. def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    
59.     """"获取所有items，默认0-100条数据"""
    
60.     items = crud.get_items(db, skip=skip, limit=limit)
    
61.     return items
    
62. 
    
63. 
    
64. """KM关联数据操作"""
    
65. 
    
66. 
    
67. @app.post("/users/{user_id}/kms/", response_model=schemas.KM)
    
68. def create_km_for_user(user_id: int, km: schemas.KMCreate, db: Session = Depends(get_db)):
    
69.     """创建该用户的items，用户关联项目信息"""
    
70.     return crud.create_user_km(db=db, km=km, user_id=user_id)
    
71. 
    
72. 
    
73. @app.get("/kms/", response_model=List[schemas.KM])
    
74. def read_kms(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    
75.     """"获取所有items，默认0-100条数据"""
    
76.     items = crud.get_kms(db, skip=skip, limit=limit)
    
77.     return items
    
78. 
    
79. """PM关联数据操作"""
    
80. 
    
81. 
    
82. @app.post("/users/{user_id}/pms/", response_model=schemas.PM)
    
83. def create_pm_for_user(user_id: int, pm: schemas.PMCreate, db: Session = Depends(get_db)):
    
84.     """创建该用户的items，用户关联项目信息"""
    
85.     return crud.create_user_pm(db=db, pm=pm, user_id=user_id)
    
86. 
    
87. 
    
88. @app.get("/pms/", response_model=List[schemas.PM])
    
89. def read_pms(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    
90.     """"获取所有items，默认0-100条数据"""
    
91.     items = crud.get_pms(db, skip=skip, limit=limit)
    
92.     return items
    
93. 
    
94. 
    
95. 
    
96. if __name__ == '__main__':
    
97.     uvicorn.run(app, host="0.0.0.0", port=80)
    

复制代码

crud

1. from sqlalchemy.orm import Session
   
2. 
   
3. # from . import models, schemas
   
4. from db import models, schemas
   
5. 
   
6. 
   
7. def get_user(db: Session, user_id: int):
   
8.     """
   
9.     根据id获取用户信息
   
10.     :param db: 数据库会话
    
11.     :param user_id: 用户id
    
12.     :return: 用户信息
    
13.     """
    
14.     return db.query(models.User).filter(models.User.id == user_id).first()
    
15. 
    
16. 
    
17. def get_user_by_email(db: Session, email: str):
    
18.     """
    
19.     根据email获取用户信息
    
20.     :param db: 数据库会话
    
21.     :param email: 用户email
    
22.     :return: 用户信息
    
23.     """
    
24.     return db.query(models.User).filter(models.User.email == email).first()
    
25. 
    
26. 
    
27. def get_users(db: Session, skip: int = 0, limit: int = 100):
    
28.     """
    
29.     获取特定数量的用户
    
30.     :param db: 数据库会话
    
31.     :param skip: 开始位置
    
32.     :param limit: 限制数量
    
33.     :return: 用户信息列表
    
34.     """
    
35.     return db.query(models.User).offset(skip).limit(limit).all()
    
36. 
    
37. 
    
38. def create_user(db: Session, user: schemas.UserCreate):
    
39.     """
    
40.     创建用户
    
41.     :param db: 数据库会话
    
42.     :param user: 用户模型
    
43.     :return: 根据email和password登录的用户信息
    
44.     """
    
45.     fake_hashed_password = user.password + "notreallyhashed"
    
46.     db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    
47.     db.add(db_user)  # 添加到会话
    
48.     db.commit()  # 提交到数据库
    
49.     db.refresh(db_user)  # 刷新数据库
    
50.     return db_user
    
51. 
    
52. 
    
53. def get_items(db: Session, skip: int = 0, limit: int = 100):
    
54.     """
    
55.     获取指定数量的item
    
56.     :param db: 数据库会话
    
57.     :param skip: 开始位置
    
58.     :param limit: 限制数量
    
59.     :return: item列表
    
60.     """
    
61.     return db.query(models.Item).offset(skip).limit(limit).all()
    
62. 
    
63. 
    
64. def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    
65.     """
    
66.     创建用户item
    
67.     :param db: 数据库会话
    
68.     :param item: Item对象
    
69.     :param user_id: 用户id
    
70.     :return: Item模型对象
    
71.     """
    
72.     db_item = models.Item(**item.dict(), owner_id=user_id)
    
73.     db.add(db_item)
    
74.     db.commit()
    
75.     db.refresh(db_item)
    
76.     return db_item
    
77. 
    
78. 
    
79. """KM关联数据操作"""
    
80. 
    
81. 
    
82. def get_kms(db: Session, skip: int = 0, limit: int = 100):
    
83.     """
    
84.     获取指定数量的item
    
85.     :param db: 数据库会话
    
86.     :param skip: 开始位置
    
87.     :param limit: 限制数量
    
88.     :return: item列表
    
89.     """
    
90.     return db.query(models.KM).offset(skip).limit(limit).all()
    
91. 
    
92. 
    
93. def create_user_km(db: Session, km: schemas.KMCreate, user_id: int):
    
94.     """
    
95.     创建用户item
    
96.     :param db: 数据库会话
    
97.     :param km: km对象
    
98.     :param user_id: 用户id
    
99.     :return: km模型对象
    
100.     """
     
101.     db_km = models.KM(**km.dict(), km_id=user_id)
     
102.     db.add(db_km)
     
103.     db.commit()
     
104.     db.refresh(db_km)
     
105.     return db_km
     
106. 
     
107. 
     
108. """PM关联数据操作"""
     
109. 
     
110. 
     
111. def get_pms(db: Session, skip: int = 0, limit: int = 100):
     
112.     """
     
113.     获取指定数量的item
     
114.     :param db: 数据库会话
     
115.     :param skip: 开始位置
     
116.     :param limit: 限制数量
     
117.     :return: item列表
     
118.     """
     
119.     return db.query(models.PM).offset(skip).limit(limit).all()
     
120. 
     
121. 
     
122. def create_user_pm(db: Session, pm: schemas.PMCreate, user_id: int):
     
123.     """
     
124.     创建用户item
     
125.     :param db: 数据库会话
     
126.     :param km: km对象
     
127.     :param user_id: 用户id
     
128.     :return: km模型对象
     
129.     """
     
130.     db_pm = models.PM(**pm.dict(), pm_id=user_id)
     
131.     db.add(db_pm)
     
132.     db.commit()
     
133.     db.refresh(db_pm)
     
134.     return db_pm
     

复制代码

schemas

1. from typing import List
   
2. from pydantic import BaseModel
   
3. 
   
4. """PM模型类定义"""
   
5. 
   
6. 
   
7. class PMBase(BaseModel):
   
8.     pname: str
   
9.     remark: str = None
   
10. 
    
11. 
    
12. class PMCreate(PMBase):
    
13.     pass
    
14. 
    
15. 
    
16. class PM(PMBase):
    
17.     id: int
    
18.     pm_id: int
    
19. 
    
20.     class Config:
    
21.         orm_mode = True
    
22. 
    
23. 
    
24. """KM模型类定义"""
    
25. 
    
26. 
    
27. class KMBase(BaseModel):
    
28.     title: str
    
29.     description: str = None
    
30. 
    
31. 
    
32. class KMCreate(KMBase):
    
33.     pass
    
34. 
    
35. 
    
36. class KM(KMBase):
    
37.     id: int
    
38.     km_id: int
    
39. 
    
40.     class Config:
    
41.         orm_mode = True
    
42. 
    
43. 
    
44. """Item模型类定义"""
    
45. 
    
46. 
    
47. class ItemBase(BaseModel):
    
48.     title: str
    
49.     description: str = None
    
50. 
    
51. 
    
52. class ItemCreate(ItemBase):
    
53.     pass
    
54. 
    
55. 
    
56. class Item(ItemBase):
    
57.     id: int
    
58.     owner_id: int
    
59. 
    
60.     class Config:
    
61.         orm_mode = True
    
62. 
    
63. 
    
64. """User模型类定义"""
    
65. 
    
66. 
    
67. class UserBase(BaseModel):
    
68.     email: str
    
69.     name: str
    
70. 
    
71. 
    
72. class UserCreate(UserBase):
    
73.     password: str
    
74.     username: str
    
75. 
    
76. 
    
77. class User(UserBase):
    
78.     id: int
    
79.     is_active: bool
    
80. 
    
81.     items: List[Item] = []
    
82.     kms: List[KM] = []
    
83.     pms: List[PM] = []
    
84.     # wts: List[PM] = []
    
85.     # emps: List[PM] = []
    
86. 
    
87.     class Config:
    
88.         orm_mode = True
    

复制代码

models

1. # 导入第三类
   
2. from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
   
3. from sqlalchemy.orm import relationship
   
4. # 导入数据库设置类，模型基础类
   
5. from db.database import Base
   
6. 
   
7. 
   
8. # 定义User类
   
9. class User(Base):
   
10.     __tablename__ = "users"  # 定义表名
    
11.     # 定义属性
    
12.     id = Column(Integer, primary_key=True, index=True)
    
13.     name = Column(String, index=True)
    
14.     username = Column(String, index=True)
    
15.     email = Column(String(255), unique=True, index=True)
    
16.     hashed_password = Column(String(255))
    
17.     is_active = Column(Boolean, default=True)
    
18.     # 关联item模型，同时关联items表
    
19.     items = relationship("Item", back_populates="owner")  # 关联Item表
    
20.     kms = relationship("KM", back_populates="km")  # 关联Item表
    
21.     pms = relationship("PM", back_populates="pm")
    
22.     # wts = relationship("PM", back_populates="pm")
    
23.     # emps = relationship("PM", back_populates="pm")
    
24. 
    
25. 
    
26. # 定义Item类
    
27. class KM(Base):
    
28.     __tablename__ = "kms"  # 定义表名
    
29.     # 定义属性
    
30.     id = Column(Integer, primary_key=True, index=True)
    
31.     title = Column(String(255), index=True)
    
32.     description = Column(String(255), index=True)
    
33.     # 关联user模型，同时关联users表
    
34.     km_id = Column(Integer, ForeignKey("users.id"))  # 关联km表
    
35.     km = relationship("User", back_populates="kms")
    
36. 
    
37. 
    
38. class PM(Base):
    
39.     __tablename__ = "pms"  # 定义表名
    
40.     # 定义属性
    
41.     id = Column(Integer, primary_key=True, index=True)
    
42.     pname = Column(String(255), index=True)
    
43.     remark = Column(String(255), index=True)
    
44.     # 关联user模型，同时关联users表
    
45.     pm_id = Column(Integer, ForeignKey("users.id"))  # 关联km表
    
46.     pm = relationship("User", back_populates="pms")
    
47. 
    
48. 
    
49. # 定义Item类
    
50. class Item(Base):
    
51.     __tablename__ = "items"  # 定义表名
    
52.     # 定义属性
    
53.     id = Column(Integer, primary_key=True, index=True)
    
54.     title = Column(String(255), index=True)
    
55.     description = Column(String(255), index=True)
    
56.     # 关联user模型，同时关联users表
    
57.     owner_id = Column(Integer, ForeignKey("users.id"))
    
58.     owner = relationship("User", back_populates="items")  # 关联User表
    

复制代码

database

1. # 导入第三方库文件
   
2. from sqlalchemy import create_engine
   
3. from sqlalchemy.ext.declarative import declarative_base
   
4. from sqlalchemy.orm import sessionmaker
   
5. 
   
6. # 数据库连接配置
   
7. # SQLALCHEMY_DATABASE_URL = "sqlite:///sqlitefastapi.db"
   
8. SQLALCHEMY_DATABASE_URL = "sqlite:///826.db"
   
9. # SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:andy123456@localhost/fastapi?charset=utf8mb4"
   
10. # 创建引擎
    
11. engine = create_engine(SQLALCHEMY_DATABASE_URL)
    
12. # 创建数据库会话
    
13. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    
14. # 声明基类；该类为后面models定义模型类使用
    
15. Base = declarative_base()
    

复制代码