Update code

Author: Youngestdev

ch06/planner/main.py

@@ -1,5 +1,6 @@
 from fastapi import FastAPI
 from fastapi.responses import RedirectResponse
+from database.database import Settings
 
 from routes.users import user_router
 from routes.events import event_router
@@ -8,11 +9,16 @@
 
 app = FastAPI()
 
+settings = Settings()
+
 # Register routes
 
 app.include_router(user_router,  prefix="/user")
 app.include_router(event_router, prefix="/event")
 
+@app.on_event("startup")
+async def init_db():
+    await settings.initialize_database()
 
 @app.get("/")
 async def home():

ch06/planner/models/events.py

@@ -1,10 +1,9 @@
-from typing import List
-
+from beanie import Document
+from typing import Optional, List
 from pydantic import BaseModel
 
 
-class Event(BaseModel):
-    id: int
+class Event(Document):
     title: str
     image: str
     description: str
@@ -21,3 +20,25 @@ class Config:
                 "location": "Google Meet"
             }
         }
+
+    class Collection:
+        name = "events"
+
+
+class EventUpdate(BaseModel):
+    title: Optional[str]
+    image: Optional[str]
+    description: Optional[str]
+    tags: Optional[List[str]]
+    location: Optional[str]
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "title": "FastAPI BookLaunch",
+                "image": "https://linktomyimage.com/image.png",
+                "description": "We will be discussing the contents of the FastAPI book in this event.Ensure to come with your own copy to win gifts!",
+                "tags": ["python", "fastapi", "book", "launch"],
+                "location": "Google Meet"
+            }
+        }

ch06/planner/models/users.py

@@ -1,58 +1,28 @@
 from typing import Optional, List
+from beanie import Document, Link
 
-from fastapi import Form
 from pydantic import BaseModel, EmailStr
 
 from models.events import Event
 
 
-class User(BaseModel):
+class User(Document):
     email: EmailStr
-    username: str
-    events: Optional[List[Event]]
+    password: str
+    events: Optional[List[Link[Event]]]
 
-    @classmethod
-    def as_form(
-            cls,
-            email: EmailStr = Form(...),
-            username: str = Form(...),
-            password: str = Form(...)
-    ):
-        return cls(email=email, username=username, password=password)
+    class Collection:
+        name = "users"
 
     class Config:
         schema_extra = {
             "example": {
                 "email": "fastapi@packt.com",
-                "username": "fastapipackt001",
+                "password": "strong!!!",
                 "events": [],
             }
         }
 
-
-class NewUser(User):
-    password: str
-
-    class Config:
-        schema_extra = {
-            "example": {
-                "email": "fastapi@packt.com",
-                "password": "Stro0ng!",
-                "username": "FastPackt"
-            }
-        }
-
-
 class UserSignIn(BaseModel):
     email: EmailStr
     password: str
-
-    @classmethod
-    def as_form(
-            cls,
-            email: EmailStr = Form(...),
-            password: str = Form(...)
-    ):
-        return cls(email=email, password=password)
-
-

ch06/planner/routes/events.py

@@ -1,66 +1,60 @@
-from fastapi import APIRouter, Body, HTTPException, Request, status
-from fastapi.responses import HTMLResponse
-from fastapi.templating import Jinja2Templates
+from urllib import response
+from beanie import PydanticObjectId
+from fastapi import APIRouter, HTTPException, status
+from database.database import Database
 
-from models.events import Event
+from models.events import Event, EventUpdate
 from typing import List
 
 event_router = APIRouter(
     tags=["Events"]
 )
 
-templates = Jinja2Templates(directory="templates/")
+event_database = Database(Event)
 
-events = []
+@event_router.get("/", response_model=List[Event])
+async def retrieve_all_events():
+    events = await event_database.get_all()
+    return events
 
 
-@event_router.get("/", response_class=HTMLResponse)
-async def retrieve_all_events(request: Request):
-    return templates.TemplateResponse(
-        "event.html",
-        {
-            "request": request,
-            "events": events
-        }
-    )
-
-
-@event_router.get("/{id}", response_class=HTMLResponse)
-async def retrieve_event(id: int, request: Request):
-    for event in events:
-        if event.id == id:
-            return templates.TemplateResponse(
-                "event.html",
-                {
-                    "request": request,
-                    "event": event
-                }
-            )
-
-    raise HTTPException(
-        status_code=status.HTTP_404_NOT_FOUND,
-        detail="Event with supplied ID does not exist"
-    )
+@event_router.get("/{id}", response_model=Event)
+async def retrieve_event(id: PydanticObjectId):
+    event = await event_database.get(id)
+    if not event:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Event with supplied ID does not exist"
+        )
+    return event
 
 
 @event_router.post("/new")
-async def create_event(body: Event = Body(...)):
-    events.append(body)
+async def create_event(body: Event):
+    await event_database.save(body)
     return {
         "message": "Event created successfully"
     }
 
+@event_router.put("/{id}", response_model=Event)
+async def update_event(id: PydanticObjectId, body: EventUpdate):
+    updated_event = await event_database.update(id, body)
+    if not updated_event:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Event with supplied ID does not exist"
+        )
+    return updated_event
 
-@event_router.delete("/{id}")
-async def delete_event(id: int):
-    for event in events:
-        if event.id == id:
-            events.remove(event)
-            return {
-                "message": "Event deleted successfully"
-            }
 
-    raise HTTPException(
-        status_code=status.HTTP_404_NOT_FOUND,
-        detail="Event with supplied ID does not exist"
-    )
+@event_router.delete("/{id}")
+async def delete_event(id: PydanticObjectId):
+    event = await event_database.delete(id)
+    if not event:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Event with supplied ID does not exist"
+        )
+    return {
+        "message": "Event deleted successfully."
+    }

ch06/planner/routes/users.py

@@ -1,69 +1,43 @@
-from fastapi import APIRouter, HTTPException, Request, status, Depends
-from fastapi.templating import Jinja2Templates
+from fastapi import APIRouter, HTTPException, status
+from database.database import Database
 
-from models.users import NewUser, UserSignIn
+from models.users import User, UserSignIn
 
 user_router = APIRouter(
     tags=["User"],
 )
 
-users = {}
-templates = Jinja2Templates(directory="templates/")
-
+user_database = Database(User)
 
 @user_router.post("/signup")
-async def sign_user_up(request: Request, data: NewUser = Depends(NewUser.as_form)):
-    if data.email in users:
+async def sign_user_up(user: User):
+    user_exist = await User.find_one(User.email == user.email)
+    print(user_exist)
+    if user_exist:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
-            detail="User with supplied username exists"
+            detail="User with email provided exists already."
         )
-
-    users[data.email] = data
-
-    return templates.TemplateResponse("user.html", {
-        "request": request,
-        "signed_in": True,
-    })
+    await user_database.save(user)
+    return {
+        "message": "User created successfully"
+    }
 
 
 @user_router.post("/signin")
-async def sign_user_in(request: Request, user: UserSignIn = Depends(UserSignIn.as_form)):
-    if user.email not in users:
+async def sign_user_in(user: UserSignIn):
+    user_exist = await User.find_one(User.email == user.email)
+    if not user_exist:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail="User does not exist"
-        )
-
-    if users[user.email].password != user.password:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Wrong credential passed"
+            detail="User with email does not exist."
         )
-    return templates.TemplateResponse(
-        "index.html",
-        {
-            "request": request,
-            "signed_in": True
-        }
-    )
-
-
-@user_router.get("/")
-async def render_login_page(request: Request):
-    return templates.TemplateResponse(
-        "user.html", {
-            "request": request,
-            "sign_in": True
-        }
-    )
-
-
-@user_router.get("/signup")
-async def render_signup_page(request: Request):
-    return templates.TemplateResponse(
-        "user.html", {
-            "request": request,
-            "sign_in": False
+    if user_exist.password == user.password:
+        return {
+            "message": "User signed in successfully."
         }
+    
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid details passed."
     )