FastAPI + SQLAlchemy/Alembic + MariaDB backend with Jinja2/htmx/Alpine server-rendered frontend. Multi-user via Google OAuth, daily habit tracking, monthly/weekly history views, Web Push reminders via APScheduler, and PWA support (manifest, service worker, offline caching). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
from authlib.integrations.starlette_client import OAuth
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.models.user import User
|
|
from app.security import SESSION_MAX_AGE_SECONDS, create_session_token
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
oauth = OAuth()
|
|
oauth.register(
|
|
name="google",
|
|
server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
|
|
client_id=settings.google_client_id,
|
|
client_secret=settings.google_client_secret,
|
|
client_kwargs={"scope": "openid email profile"},
|
|
)
|
|
|
|
|
|
@router.get("/google/login")
|
|
async def google_login(request: Request):
|
|
return await oauth.google.authorize_redirect(request, settings.google_redirect_uri)
|
|
|
|
|
|
@router.get("/google/callback")
|
|
async def google_callback(request: Request, db: Session = Depends(get_db)):
|
|
token = await oauth.google.authorize_access_token(request)
|
|
userinfo = token["userinfo"]
|
|
google_sub = userinfo["sub"]
|
|
email = userinfo["email"]
|
|
name = userinfo.get("name")
|
|
picture_url = userinfo.get("picture")
|
|
|
|
user = db.scalar(select(User).where(User.google_sub == google_sub))
|
|
if user is None:
|
|
user = User(google_sub=google_sub, email=email, name=name, picture_url=picture_url)
|
|
db.add(user)
|
|
else:
|
|
user.email = email
|
|
user.name = name
|
|
user.picture_url = picture_url
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
session_token = create_session_token(user.id)
|
|
response = RedirectResponse(url="/today", status_code=303)
|
|
response.set_cookie(
|
|
settings.session_cookie_name,
|
|
session_token,
|
|
httponly=True,
|
|
samesite="lax",
|
|
max_age=SESSION_MAX_AGE_SECONDS,
|
|
)
|
|
return response
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout():
|
|
response = RedirectResponse(url="/login", status_code=303)
|
|
response.delete_cookie(settings.session_cookie_name)
|
|
return response
|