|
| 1 | +from datetime import datetime, timedelta, timezone |
| 2 | +from secrets import token_urlsafe |
| 3 | +from typing import Annotated |
| 4 | + |
| 5 | +from fastapi import Depends, FastAPI, Form, HTTPException, status |
| 6 | +from fastapi.security import OAuth2PasswordBearer |
| 7 | +from pydantic import BaseModel |
| 8 | +from typing import Union, Literal |
| 9 | +import uvicorn |
| 10 | + |
| 11 | + |
| 12 | +app = FastAPI() |
| 13 | + |
| 14 | +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login") |
| 15 | + |
| 16 | +# Demo users only. Use a database and password hashing in a real application. |
| 17 | +USERS = { |
| 18 | + "admin": {"password": "admin123", "role": "admin"}, |
| 19 | + "pawan": {"password": "user123", "role": "user"}, |
| 20 | +} |
| 21 | +active_tokens = {} |
| 22 | +TOKEN_EXPIRE_MINUTES = 30 |
| 23 | + |
| 24 | + |
| 25 | +@app.post("/login") |
| 26 | +async def login(username: str = Form(...), password: str = Form(...)): |
| 27 | + user = USERS.get(username) |
| 28 | + |
| 29 | + if user is None or user["password"] != password: |
| 30 | + raise HTTPException( |
| 31 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 32 | + detail="Incorrect username or password", |
| 33 | + headers={"WWW-Authenticate": "Bearer"}, |
| 34 | + ) |
| 35 | + |
| 36 | + token = token_urlsafe(32) |
| 37 | + active_tokens[token] = { |
| 38 | + "username": username, |
| 39 | + "role": user["role"], |
| 40 | + "expires_at": datetime.now(timezone.utc) |
| 41 | + + timedelta(minutes=TOKEN_EXPIRE_MINUTES), |
| 42 | + } |
| 43 | + |
| 44 | + return {"access_token": token, "token_type": "bearer"} |
| 45 | + |
| 46 | + |
| 47 | +def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): |
| 48 | + user = active_tokens.get(token) |
| 49 | + |
| 50 | + if user is None or user["expires_at"] <= datetime.now(timezone.utc): |
| 51 | + raise HTTPException( |
| 52 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 53 | + detail="Invalid or expired authentication token", |
| 54 | + headers={"WWW-Authenticate": "Bearer"}, |
| 55 | + ) |
| 56 | + |
| 57 | + return user |
| 58 | + |
| 59 | + |
| 60 | +def require_admin(user: Annotated[dict, Depends(get_current_user)]): |
| 61 | + if user["role"] != "admin": |
| 62 | + raise HTTPException( |
| 63 | + status_code=status.HTTP_403_FORBIDDEN, |
| 64 | + detail="Admin access required", |
| 65 | + ) |
| 66 | + |
| 67 | + return user |
| 68 | + |
| 69 | + |
| 70 | +@app.get("/") |
| 71 | +async def index(user: Annotated[dict, Depends(get_current_user)]): |
| 72 | + #return {"message": "Hello World"} |
| 73 | + return f"Mathematics Claculator" |
| 74 | + |
| 75 | +# Calculator input schema |
| 76 | +class CalculatorInput(BaseModel): |
| 77 | + num1: Union[int, float] |
| 78 | + symbol: Literal['+', '-', '*', '/'] |
| 79 | + num2: Union[int, float] |
| 80 | + |
| 81 | +# Calculator logic |
| 82 | +class Calculator: |
| 83 | + def __init__(self, num1: Union[int, float], symbol: str, num2: Union[int, float]): |
| 84 | + self.num1 = num1 |
| 85 | + self.symbol = symbol |
| 86 | + self.num2 = num2 |
| 87 | + |
| 88 | + def calculate(self) -> float: |
| 89 | + if self.symbol == '+': |
| 90 | + return self.num1 + self.num2 |
| 91 | + elif self.symbol == '-': |
| 92 | + return self.num1 - self.num2 |
| 93 | + elif self.symbol == '*': |
| 94 | + return self.num1 * self.num2 |
| 95 | + elif self.symbol == '/': |
| 96 | + if self.num2 == 0: |
| 97 | + raise ValueError("Division by zero is not allowed") |
| 98 | + return self.num1 / self.num2 |
| 99 | + else: |
| 100 | + raise ValueError("Invalid operation") |
| 101 | + |
| 102 | +@app.delete("/users/{user_id}") |
| 103 | +def delete_user( |
| 104 | + user_id: int, |
| 105 | + user: Annotated[dict, Depends(require_admin)], |
| 106 | +): |
| 107 | + return { |
| 108 | + "message": f"User {user_id} deleted" |
| 109 | + } |
| 110 | + |
| 111 | +@app.post("/calculate") |
| 112 | +async def calculate( |
| 113 | + data: CalculatorInput, |
| 114 | + user: Annotated[dict, Depends(get_current_user)], |
| 115 | +): |
| 116 | + try: |
| 117 | + calc = Calculator(data.num1, data.symbol, data.num2) |
| 118 | + result = calc.calculate() |
| 119 | + return {"result": result} |
| 120 | + except ValueError as e: |
| 121 | + raise HTTPException(status_code=400, detail=str(e)) |
| 122 | + |
| 123 | +# form-data endpoint |
| 124 | +@app.post("/calculate-form") |
| 125 | +async def calculate_form( |
| 126 | + user: Annotated[dict, Depends(get_current_user)], |
| 127 | + num1: float = Form(...), |
| 128 | + symbol: str = Form(...), |
| 129 | + num2: float = Form(...), |
| 130 | +): |
| 131 | + try: |
| 132 | + calc = Calculator(num1, symbol, num2) |
| 133 | + result = calc.calculate() |
| 134 | + return {"result": result} |
| 135 | + except ValueError as e: |
| 136 | + raise HTTPException(status_code=400, detail=str(e)) |
| 137 | + |
| 138 | +if __name__ == "__main__": |
| 139 | + uvicorn.run("Ganesh:app", host="127.0.0.1", port=8000, reload=True) |
0 commit comments