add: FastAPI Tutorial

Author: king04aman

Learn-FastAPI/FastAPI-CURD.py

@@ -0,0 +1,66 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from typing import List, Optional
+from uuid import UUID, uuid4
+
+app = FastAPI()
+
+# Define a Pydantic model for Task
+class Task(BaseModel):
+    id: Optional[UUID] = None
+    title: str
+    description: Optional[str] = None
+    completed: bool = False
+
+# Initialize an empty list to store tasks
+tasks = []
+
+# Create a new task
+@app.post("/tasks/", response_model=Task)
+def create_task(task: Task):
+    task.id = uuid4()  # Assign a UUID to the task
+    tasks.append(task)  # Add the task to the tasks list
+    return task
+
+# Retrieve all tasks
+@app.get("/tasks/", response_model=List[Task])
+def read_tasks():
+    return tasks
+
+# Retrieve a specific task by ID
+@app.get("/tasks/{task_id}", response_model=Task)
+def read_task(task_id: UUID):
+    for task in tasks:
+        if task.id == task_id:
+            return task
+    
+    # Raise HTTPException if task not found
+    raise HTTPException(status_code=404, detail="Task not found")
+
+# Update a specific task by ID
+@app.put("/tasks/{task_id}", response_model=Task)
+def update_task(task_id: UUID, task_update: Task):
+    for idx, task in enumerate(tasks):
+        if task.id == task_id:
+            updated_task = task.copy(update=task_update.dict(exclude_unset=True))
+            tasks[idx] = updated_task
+            return updated_task
+    
+    # Raise HTTPException if task not found
+    raise HTTPException(status_code=404, detail="Task not found")
+
+# Delete a specific task by ID
+@app.delete("/tasks/{task_id}", response_model=Task)
+def delete_task(task_id: UUID):
+    for idx, task in enumerate(tasks):
+        if task.id == task_id:
+            return tasks.pop(idx)
+    
+    # Raise HTTPException if task not found
+    raise HTTPException(status_code=404, detail="Task not found")
+
+# Run the FastAPI app
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)

Learn-FastAPI/FastAPI-Tutorial.py

@@ -0,0 +1,134 @@
+from fastapi import Depends, FastAPI, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from pydantic import BaseModel
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+
+# Constants
+SECRET_KEY = ""
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+# Database simulation
+db = {
+    "aman": {
+        "username": "king04aman",
+        "full_name": "Aman Kumar",
+        "email": "[email protected]",
+        "hashed_password": "$2b$12$HxWHkvMuL7WrZad6lcCfluNFj1/Zp63lvP5aUrKlSTYtoFzPXHOtu",
+        "disabled": False
+    }
+}
+
+# Pydantic models
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: str or None = None
+
+class User(BaseModel):
+    username: str
+    email: str or None = None
+    full_name: str or None = None
+    disabled: bool or None = None
+
+class UserInDB(User):
+    hashed_password: str
+
+# Password hashing context
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+# OAuth2 password bearer
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+# FastAPI instance
+app = FastAPI()
+
+# Password helper functions
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+# User authentication functions
+def get_user(db, username: str):
+    if username in db:
+        user_data = db[username]
+        return UserInDB(**user_data)
+
+def authenticate_user(db, username: str, password: str):
+    user = get_user(db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.hashed_password):
+        return False
+    return user
+
+# Token creation function
+def create_access_token(data: dict, expires_delta: timedelta or None = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+# Dependency to get current user
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    credential_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"}
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credential_exception
+        token_data = TokenData(username=username)
+    except JWTError:
+        raise credential_exception
+
+    user = get_user(db, username=token_data.username)
+    if user is None:
+        raise credential_exception
+
+    return user
+
+# Dependency to get current active user
+async def get_current_active_user(current_user: UserInDB = Depends(get_current_user)):
+    if current_user.disabled:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    return current_user
+
+# Route to get access token
+@app.post("/token", response_model=Token)
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
+    user = authenticate_user(db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"}
+        )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user.username},
+        expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+# Route to get current user
+@app.get("/users/me/", response_model=User)
+async def read_users_me(current_user: User = Depends(get_current_active_user)):
+    return current_user
+
+# Route to get current user's items
+@app.get("/users/me/items")
+async def read_own_items(current_user: User = Depends(get_current_active_user)):
+    return [{"item_id": 1, "owner": current_user}]

Learn-FastAPI/README.md

@@ -0,0 +1,64 @@
+# FastAPI CRUD and Authentication Example
+
+This directory contains two FastAPI applications: one for basic CRUD operations (task management) and another for user authentication using JWT (JSON Web Tokens).
+
+## Overview
+
+1. **Task Management API (`fastapi-curd.py`)**:
+   - Implements basic CRUD operations for managing tasks.
+   - Tasks have attributes like `id`, `title`, `description`, and `completed`.
+
+2. **User Authentication API (`fastapi-tutorial.py`)**:
+   - Implements user registration, login, and token-based authentication.
+   - Uses JWT for secure access and stores user information in a simulated database.
+
+## Installation
+
+To run the applications, ensure you have Python and FastAPI installed. You can install FastAPI and Uvicorn using pip:
+
+```bash
+pip install fastapi uvicorn python-jose passlib[bcrypt]
+```
+
+## Running the Applications
+
+You can run each application separately.
+
+### Task Management API
+
+To run the task management API:
+
+```bash
+python fastapi-curd.py
+```
+
+The API will be available at `http://localhost:8000/tasks/`.
+
+### User Authentication API
+
+To run the user authentication API:
+
+```bash
+python fastapi-tutorial.py
+```
+
+The API will be available at `http://localhost:8000/token` for login and `http://localhost:8000/users/me/` for user information.
+
+## API Endpoints
+
+### Task Management API Endpoints
+
+- **Create Task**: `POST /tasks/`
+- **Retrieve All Tasks**: `GET /tasks/`
+- **Retrieve Specific Task**: `GET /tasks/{task_id}`
+- **Update Task**: `PUT /tasks/{task_id}`
+- **Delete Task**: `DELETE /tasks/{task_id}`
+
+### User Authentication API Endpoints
+
+- **Token Generation**: `POST /token`
+- **Get Current User**: `GET /users/me/`
+- **Get Current User's Items**: `GET /users/me/items`
+
+## Author
+- [king04aman](https://github.com/king04aman)