Skip to content
Snippets Groups Projects

improve front

Merged Antoine Gaudron-Desjardins requested to merge collab_front into main
1 file
+ 0
146
Compare changes
  • Side-by-side
  • Inline
from pydantic import BaseModel
from fastapi import HTTPException, APIRouter, Response, Depends
from uuid import UUID, uuid4
from fastapi_sessions.backends.implementations import InMemoryBackend
from fastapi_sessions.session_verifier import SessionVerifier
from fastapi_sessions.frontends.implementations import SessionCookie, CookieParameters
from fastapi.responses import RedirectResponse
from urllib.parse import urlencode
from dotenv import load_dotenv
from sqlalchemy.orm import Session
from requests import get, post
import os
load_dotenv()
class SessionData(BaseModel):
username: str
cookie_params = CookieParameters()
# Uses UUID
cookie = SessionCookie(
cookie_name="cookie",
identifier="general_verifier",
auto_error=True,
secret_key="DONOTUSE",
cookie_params=cookie_params,
)
backend = InMemoryBackend[UUID, SessionData]()
class BasicVerifier(SessionVerifier[UUID, SessionData]):
def __init__(
self,
*,
identifier: str,
auto_error: bool,
backend: InMemoryBackend[UUID, SessionData],
auth_http_exception: HTTPException,
):
self._identifier = identifier
self._auto_error = auto_error
self._backend = backend
self._auth_http_exception = auth_http_exception
@property
def identifier(self):
return self._identifier
@property
def backend(self):
return self._backend
@property
def auto_error(self):
return self._auto_error
@property
def auth_http_exception(self):
return self._auth_http_exception
def verify_session(self, model: SessionData) -> bool:
"""If the session exists, it is valid"""
return True
verifier = BasicVerifier(
identifier="general_verifier",
auto_error=True,
backend=backend,
auth_http_exception=HTTPException(status_code=403, detail="invalid session"),
)
router = APIRouter(prefix="/api", tags=["auth"])
@router.post("/create_session/{name}")
async def create_session(name: str, response: Response):
session = uuid4()
data = SessionData(username=name)
await backend.create(session, data)
cookie.attach_to_response(response, session)
return f"created session for {name}"
@router.get("/whoami", dependencies=[Depends(cookie)])
async def whoami(session_data: SessionData = Depends(verifier)):
return session_data
@router.post("/delete_session")
async def del_session(response: Response, session_id: UUID = Depends(cookie)):
await backend.delete(session_id)
cookie.delete_from_response(response)
return "deleted session"
@router.get("/login")
async def login(code):
if not code:
new_state = crud.create_state(db)
params = urlencode({
"client_id": os.getenv("CLIENT_ID"),
"redirect_uri": f"{os.getenv('API_ROOT')}/login",
"response_type": "code",
"state": new_state.state,
"scope": "default linkcs-event:read"})
return RedirectResponse(f"{os.getenv('AUTH_ROOT')}/oauth/authorize?{params}")
stored_states = crud.get_states(db)
if state not in map(lambda state: state.state, stored_states):
raise HTTPException(status_code=403, detail="State Invalid")
crud.delete_state(state, db)
headers = {"content-type": "application/x-www-form-urlencoded"}
data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": f"{os.getenv('API_ROOT')}/login",
"client_id": os.getenv("CLIENT_ID"),
"client_secret": os.getenv("CLIENT_SECRET"),
}
token_response = post(
f"{os.getenv('AUTH_ROOT')}/oauth/token",
data=data,
headers=headers
)
access_token = token_response.json()["access_token"]
user_info = get(
f"{os.getenv('AUTH_ROOT')}/api/user/show/me",
headers={"Authorization": f"Bearer {access_token}"}
)
return crud.create_user(user_info.json(), db)
Loading