124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
from datetime import datetime
|
|
from typing import List
|
|
from fastapi import Depends, FastAPI, HTTPException, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
|
from sqlalchemy.orm import Session
|
|
|
|
import auth
|
|
import models
|
|
import database
|
|
import schemas
|
|
import horoscope
|
|
|
|
models.Base.metadata.create_all(bind=database.engine)
|
|
|
|
app = FastAPI()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
|
|
|
|
|
# Dependency
|
|
def get_db():
|
|
db = database.SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def get_user(db: Session, username: str):
|
|
return db.query(models.User).filter(models.User.username == username).first()
|
|
|
|
|
|
async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
|
|
credentials_exception = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
try:
|
|
payload = auth.jwt.decode(token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM])
|
|
username: str = payload.get("sub")
|
|
if username is None:
|
|
raise credentials_exception
|
|
token_data = schemas.TokenData(username=username)
|
|
except auth.JWTError:
|
|
raise credentials_exception
|
|
user = get_user(db, username=token_data.username)
|
|
if user is None:
|
|
raise credentials_exception
|
|
return user
|
|
|
|
|
|
async def get_current_admin_user(current_user: models.User = Depends(get_current_user)):
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="The user doesn't have enough privileges")
|
|
return current_user
|
|
|
|
|
|
@app.post("/token", response_model=schemas.Token)
|
|
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
|
|
user = get_user(db, form_data.username)
|
|
if not user or not auth.verify_password(form_data.password, user.hashed_password):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
access_token = auth.create_access_token(data={"sub": user.username})
|
|
return {"access_token": access_token, "token_type": "bearer"}
|
|
|
|
|
|
@app.get("/users/me/", response_model=schemas.User)
|
|
async def read_users_me(current_user: models.User = Depends(get_current_user)):
|
|
return current_user
|
|
|
|
|
|
@app.get("/signs", response_model=List[schemas.Sign])
|
|
def get_signs(db: Session = Depends(get_db)):
|
|
return db.query(models.Sign).all()
|
|
|
|
|
|
@app.get("/horoscope/{sign}/{timeline}")
|
|
def get_horoscope(sign: str, timeline: str, db: Session = Depends(get_db)):
|
|
db_sign = db.query(models.Sign).filter(models.Sign.name == sign).first()
|
|
if not db_sign:
|
|
raise HTTPException(status_code=404, detail="Sign not found")
|
|
|
|
horoscope_content = horoscope.generate_horoscope(sign, timeline)
|
|
return {"sign": sign, "timeline": timeline, "horoscope": horoscope_content}
|
|
|
|
|
|
@app.post("/horoscope/{sign}/{timeline}")
|
|
def create_horoscope(sign: str, timeline: str, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_admin_user)):
|
|
db_sign = db.query(models.Sign).filter(models.Sign.name == sign).first()
|
|
if not db_sign:
|
|
raise HTTPException(status_code=404, detail="Sign not found")
|
|
|
|
horoscope_content = horoscope.generate_horoscope(sign, timeline)
|
|
|
|
db_horoscope = models.Horoscope(
|
|
sign_id=db_sign.id,
|
|
timeline=timeline,
|
|
content=horoscope_content,
|
|
date=datetime.utcnow().date()
|
|
)
|
|
db.add(db_horoscope)
|
|
db.commit()
|
|
db.refresh(db_horoscope)
|
|
return db_horoscope
|
|
|
|
|
|
@app.get("/")
|
|
def read_root():
|
|
return {"message": "Welcome to the Horoscope API"}
|