Initialize chapter 9.

Author: Youngestdev

ch09/planner/auth/__init__.py

@@ -0,0 +1,17 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+
+from auth.jwt_handler import verify_access_token
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/user/signin")
+
+
+async def authenticate(token: str = Depends(oauth2_scheme)) -> str:
+    if not token:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Sign in for access"
+        )
+
+    decoded_token = await verify_access_token(token)
+    return decoded_token["user"]

ch09/planner/auth/authenticate.py

@@ -0,0 +1,11 @@
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+class HashPassword:
+    def create_hash(self, password: str) -> str:
+        return pwd_context.hash(password)
+
+    def verify_hash(self, plain_password: str, hashed_password: str) -> bool:
+        return pwd_context.verify(plain_password, hashed_password)

ch09/planner/auth/hash_password.py

@@ -0,0 +1,52 @@
+import time
+from datetime import datetime
+
+from fastapi import HTTPException, status
+from jose import jwt, JWTError
+
+from database.database import Settings
+from models.users import User
+
+settings = Settings()
+
+
+def create_access_token(user: str) -> str:
+    payload = {
+        "user": user,
+        "expires": time.time() + 3600
+    }
+
+    token = jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")
+    return token
+
+
+async def verify_access_token(token: str) -> dict:
+    try:
+        data = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
+
+        expire = data.get("expires")
+
+        if expire is None:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="No access token supplied"
+            )
+        if datetime.utcnow() > datetime.utcfromtimestamp(expire):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Token expired!"
+            )
+        user_exist = await User.find_one(User.email == data["user"])
+        if not user_exist:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid token"
+            )
+
+        return data
+
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid token"
+        )

ch09/planner/auth/jwt_handler.py

@@ -0,0 +1,61 @@
+from beanie import init_beanie, PydanticObjectId
+from motor.motor_asyncio import AsyncIOMotorClient
+from typing import Optional, Any, List
+from pydantic import BaseSettings, BaseModel
+
+from models.users import User
+from models.events import Event
+
+
+class Settings(BaseSettings):
+    DATABASE_URL: Optional[str] = None
+    SECRET_KEY: Optional[str] = "default"
+
+    async def initialize_database(self):
+        client = AsyncIOMotorClient(self.DATABASE_URL)
+        await init_beanie(database=client.get_default_database(),
+                          document_models=[Event, User])
+
+    class Config:
+        env_file = ".env"
+
+
+class Database:
+    def __init__(self, model):
+        self.model = model
+
+    async def save(self, document):
+        await document.create()
+        return
+
+    async def get(self, id: PydanticObjectId) -> bool:
+        doc = await self.model.get(id)
+        if doc:
+            return doc
+        return False
+
+    async def get_all(self) -> List[Any]:
+        docs = await self.model.find_all().to_list()
+        return docs
+
+    async def update(self, id: PydanticObjectId, body: BaseModel) -> Any:
+        doc_id = id
+        des_body = body.dict()
+
+        des_body = {k: v for k, v in des_body.items() if v is not None}
+        update_query = {"$set": {
+            field: value for field, value in des_body.items()
+        }}
+
+        doc = await self.get(doc_id)
+        if not doc:
+            return False
+        await doc.update(update_query)
+        return doc
+
+    async def delete(self, id: PydanticObjectId) -> bool:
+        doc = await self.get(id)
+        if not doc:
+            return False
+        await doc.delete()
+        return True

ch09/planner/database/__init__.py

@@ -0,0 +1,30 @@
+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
+
+import uvicorn
+
+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():
+    return RedirectResponse(url="/event/")
+
+if __name__ == '__main__':
+    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)

ch09/planner/database/database.py

@@ -0,0 +1,45 @@
+from beanie import Document
+from typing import Optional, List
+from pydantic import BaseModel
+
+
+class Event(Document):
+    creator: Optional[str]
+    title: str
+    image: str
+    description: str
+    tags: List[str]
+    location: str
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "title": "FastAPI Book Launch",
+                "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"
+            }
+        }
+
+    class Settings:
+        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"
+            }
+        }

ch09/planner/main.py

@@ -0,0 +1,24 @@
+from beanie import Document
+
+from pydantic import BaseModel, EmailStr
+
+
+class User(Document):
+    email: EmailStr
+    password: str
+
+    class Settings:
+        name = "users"
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "email": "fastapi@packt.com",
+                "password": "strong!!!"
+            }
+        }
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str