90c1cdef34
- Fix code linting issues - Update README with detailed documentation - Configure database paths for the current environment - Create necessary directory structure The News Aggregation Service is now ready to use with FastAPI and SQLite.
82 lines
2.1 KiB
Python
82 lines
2.1 KiB
Python
from typing import Generator
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from jose import jwt, JWTError
|
|
from pydantic import ValidationError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.core.security import ALGORITHM, oauth2_scheme
|
|
from app.db.session import SessionLocal
|
|
from app.models.user import User
|
|
from app.schemas.token import TokenPayload
|
|
from app.services.user import get_user
|
|
|
|
|
|
def get_db() -> Generator:
|
|
"""
|
|
Dependency to get a database session.
|
|
"""
|
|
try:
|
|
db = SessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def get_current_user(
|
|
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
|
|
) -> User:
|
|
"""
|
|
Dependency to get the current authenticated user.
|
|
"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token, settings.SECRET_KEY, algorithms=[ALGORITHM]
|
|
)
|
|
token_data = TokenPayload(**payload)
|
|
except (JWTError, ValidationError):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user = get_user(db, user_id=token_data.sub)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found",
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
def get_current_active_user(
|
|
current_user: User = Depends(get_current_user),
|
|
) -> User:
|
|
"""
|
|
Dependency to get the current active user.
|
|
"""
|
|
if not current_user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Inactive user",
|
|
)
|
|
|
|
return current_user
|
|
|
|
|
|
def get_current_active_superuser(
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> User:
|
|
"""
|
|
Dependency to get the current active superuser.
|
|
"""
|
|
if not current_user.is_superuser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="The user doesn't have enough privileges",
|
|
)
|
|
|
|
return current_user |