629测试程序

37ck

1. import uvicorn as uvicorn
   
2. from fastapi import FastAPI, HTTPException
   
3. from motor.motor_asyncio import AsyncIOMotorClient
   
4. 
   
5. app = FastAPI()
   
6. 
   
7. # 连接MongoDB数据库
   
8. client = AsyncIOMotorClient("mongodb://root:37ck2020@iot.37ck.cn:27017")
   
9. db = client["myforum"]
   
10. collection = db["posts"]
    
11. 
    
12. 
    
13. # 单独数据查询
    
14. @app.get("/posts/{post_id}")
    
15. async def get_post(post_id: str):
    
16.     try:
    
17.         post = await collection.find_one({"_id": post_id})
    
18.         if not post:
    
19.             raise HTTPException(status_code=404, detail="Post not found")
    
20.         return post
    
21.     except Exception as e:
    
22.         raise HTTPException(status_code=500, detail=str(e))
    
23. 
    
24. 
    
25. # 批量数据查询
    
26. @app.get("/posts")
    
27. async def get_posts():
    
28.     try:
    
29.         posts = []
    
30.         async for post in collection.find():
    
31.             posts.append(post)
    
32.         return posts
    
33.     except Exception as e:
    
34.         raise HTTPException(status_code=500, detail=str(e))
    
35. 
    
36. 
    
37. # 单独数据创建
    
38. @app.post("/posts/{post_id}")
    
39. async def create_post(post_id: str, post_data: dict):
    
40.     try:
    
41.         post = await collection.find_one({"_id": post_id})
    
42.         if post:
    
43.             raise HTTPException(status_code=409, detail="Post already exists")
    
44.         await collection.insert_one({"_id": post_id, **post_data})
    
45.         return {"message": "Post created successfully"}
    
46.     except HTTPException:
    
47.         raise
    
48.     except Exception as e:
    
49.         raise HTTPException(status_code=500, detail=str(e))
    
50. 
    
51. 
    
52. # 批量数据创建
    
53. @app.post("/posts")
    
54. async def create_posts(posts_data: list):
    
55.     try:
    
56.         for post_data in posts_data:
    
57.             post_id = post_data["_id"]
    
58.             post = await collection.find_one({"_id": post_id})
    
59.             if post:
    
60.                 raise HTTPException(status_code=409, detail=f"Post with id {post_id} already exists")
    
61.         await collection.insert_many(posts_data)
    
62.         return {"message": "Posts created successfully"}
    
63.     except HTTPException:
    
64.         raise
    
65.     except Exception as e:
    
66.         raise HTTPException(status_code=500, detail=str(e))
    
67. 
    
68. 
    
69. # 单独数据更新
    
70. @app.put("/posts/{post_id}")
    
71. async def update_post(post_id: str, post_data: dict):
    
72.     try:
    
73.         existing_post = await collection.find_one({"_id": post_id})
    
74.         if not existing_post:
    
75.             raise HTTPException(status_code=404, detail="Post not found")
    
76.         await collection.update_one({"_id": post_id}, {"$set": post_data})
    
77.         return {"message": "Post updated successfully"}
    
78.     except Exception as e:
    
79.         raise HTTPException(status_code=500, detail=str(e))
    
80. 
    
81. 
    
82. # 批量数据更新
    
83. @app.put("/posts")
    
84. async def update_posts(posts_data: list):
    
85.     try:
    
86.         for post_data in posts_data:
    
87.             post_id = post_data["_id"]
    
88.             existing_post = await collection.find_one({"_id": post_id})
    
89.             if not existing_post:
    
90.                 raise HTTPException(status_code=404, detail=f"Post with id {post_id} not found")
    
91.             await collection.update_one({"_id": post_id}, {"$set": post_data})
    
92.         return {"message": "Posts updated successfully"}
    
93.     except Exception as e:
    
94.         raise HTTPException(status_code=500, detail=str(e))
    
95. 
    
96. 
    
97. # 单独数据删除
    
98. @app.delete("/posts/{post_id}")
    
99. async def delete_post(post_id: str):
    
100.     try:
     
101.         existing_post = await collection.find_one({"_id": post_id})
     
102.         if not existing_post:
     
103.             raise HTTPException(status_code=404, detail="Post not found")
     
104.         await collection.delete_one({"_id": post_id})
     
105.         return {"message": "Post deleted successfully"}
     
106.     except Exception as e:
     
107.         raise HTTPException(status_code=500, detail=str(e))
     
108. 
     
109. 
     
110. # 批量数据删除
     
111. @app.delete("/posts")
     
112. async def delete_posts(post_ids: list):
     
113.     try:
     
114.         delete_results = await collection.delete_many({"_id": {"$in": post_ids}})
     
115.         return {"message": "Posts deleted successfully", "deleted_count": delete_results.deleted_count}
     
116.     except Exception as e:
     
117.         raise HTTPException(status_code=500, detail=str(e))
     
118. 
     
119. 
     
120. # 主程序入口
     
121. if __name__ == '__main__':
     
122.     uvicorn.run(app, host="0.0.0.0", port=81)
     

复制代码