All-In-One-Python-Projects/Learn-FastAPI/FastAPI-Tutorial.py at main · Kshi123/All-In-One-Python-Projects · GitHub

History

134 lines (114 loc) · 4.28 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

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": "aman@gmail.com",

"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}]

All-In-One-Python-Projects/Learn-FastAPI/FastAPI-Tutorial.py at main · Kshi123/All-In-One-Python-Projects · GitHub

Expand file tree

Breadcrumbs

FastAPI-Tutorial.py

Latest commit

History

Breadcrumbs

FastAPI-Tutorial.py

File metadata and controls