adding current revision information

horoscope/backend/main.py

@@ -0,0 +1,123 @@
+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"}