Skip to content
Snippets Groups Projects
Commit c364ef80 authored by Aymeric Chaumont's avatar Aymeric Chaumont
Browse files

delete accidently gitted file

parent bd249ceb
No related branches found
No related tags found
2 merge requests!30Restaurants route,!28improve front
Pipeline #43977 passed with warnings
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)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment