motor mongodb python案例

37ck

database.py

1. #  @bekbrace
   
2. #  FARMSTACK Tutorial - Sunday 13.06.2021
   
3. 
   
4. import motor.motor_asyncio
   
5. from model import Todo
   
6. 
   
7. client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017/')
   
8. database = client.TodoList
   
9. collection = database.todo
   
10. 
    
11. async def fetch_one_todo(title):
    
12.     document = await collection.find_one({"title": title})
    
13.     return document
    
14. 
    
15. async def fetch_all_todos():
    
16.     todos = []
    
17.     cursor = collection.find({})
    
18.     async for document in cursor:
    
19.         todos.append(Todo(**document))
    
20.     return todos
    
21. 
    
22. async def create_todo(todo):
    
23.     document = todo
    
24.     result = await collection.insert_one(document)
    
25.     return document
    
26. 
    
27. 
    
28. async def update_todo(title, desc):
    
29.     await collection.update_one({"title": title}, {"$set": {"description": desc}})
    
30.     document = await collection.find_one({"title": title})
    
31.     return document
    
32. 
    
33. async def remove_todo(title):
    
34.     await collection.delete_one({"title": title})
    
35.     return True

复制代码

model.py

1. #  @bekbrace
   
2. #  FARMSTACK Tutorial - Sunday 13.06.2021
   
3. 
   
4. # Pydantic allows auto creation of JSON Schemas from models
   
5. from pydantic import BaseModel
   
6. 
   
7. class Todo(BaseModel):
   
8.     title: str
   
9.     description: str

复制代码

main.py

1. #  @bekbrace
   
2. #  FARMSTACK Tutorial - Sunday 13.06.2021
   
3. 
   
4. from fastapi import FastAPI, HTTPException
   
5. 
   
6. from model import Todo
   
7. 
   
8. from database import (
   
9.     fetch_one_todo,
   
10.     fetch_all_todos,
    
11.     create_todo,
    
12.     update_todo,
    
13.     remove_todo,
    
14. )
    
15. 
    
16. # an HTTP-specific exception class  to generate exception information
    
17. 
    
18. from fastapi.middleware.cors import CORSMiddleware
    
19. app = FastAPI()
    
20. 
    
21. origins = [
    
22.     "http://localhost:3000",
    
23. ]
    
24. 
    
25. # what is a middleware?
    
26. # software that acts as a bridge between an operating system or database and applications, especially on a network.
    
27. 
    
28. app.add_middleware(
    
29.     CORSMiddleware,
    
30.     allow_origins=origins,
    
31.     allow_credentials=True,
    
32.     allow_methods=["*"],
    
33.     allow_headers=["*"],
    
34. )
    
35. 
    
36. @app.get("/")
    
37. async def read_root():
    
38.     return {"Hello": "World"}
    
39. 
    
40. @app.get("/api/todo")
    
41. async def get_todo():
    
42.     response = await fetch_all_todos()
    
43.     return response
    
44. 
    
45. @app.get("/api/todo/{title}", response_model=Todo)
    
46. async def get_todo_by_title(title):
    
47.     response = await fetch_one_todo(title)
    
48.     if response:
    
49.         return response
    
50.     raise HTTPException(404, f"There is no todo with the title {title}")
    
51. 
    
52. @app.post("/api/todo/", response_model=Todo)
    
53. async def post_todo(todo: Todo):
    
54.     response = await create_todo(todo.dict())
    
55.     if response:
    
56.         return response
    
57.     raise HTTPException(400, "Something went wrong")
    
58. 
    
59. @app.put("/api/todo/{title}/", response_model=Todo)
    
60. async def put_todo(title: str, desc: str):
    
61.     response = await update_todo(title, desc)
    
62.     if response:
    
63.         return response
    
64.     raise HTTPException(404, f"There is no todo with the title {title}")
    
65. 
    
66. @app.delete("/api/todo/{title}")
    
67. async def delete_todo(title):
    
68.     response = await remove_todo(title)
    
69.     if response:
    
70.         return "Successfully deleted todo"
    
71.     raise HTTPException(404, f"There is no todo with the title {title}")

复制代码