Skip to content
Extraits de code Groupes Projets
main.py 16,9 ko
Newer Older
  • Learn to ignore specific revisions
  • from collections import defaultdict
    
    import datetime
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    from typing import Annotated
    
    from fastapi import APIRouter, FastAPI, Form, status, Depends, HTTPException, BackgroundTasks, Header
    
    from sqlalchemy.orm import Session
    
    from fastapi.security import OAuth2PasswordRequestForm
    
    from fastapi.middleware.cors import CORSMiddleware
    
    from fastapi.websockets import WebSocket
    
    from contextlib import asynccontextmanager
    
    import json
    
    import schemas
    import crud
    
    import models
    
    from database import Base, engine, get_db, SessionLocal
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    import hashing
    
    from utils import check_user_level
    
    websocket_users = defaultdict(lambda: defaultdict(set))
    
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        db = SessionLocal()
        try:
    
            if crud.get_user_by_email(db, config.ADMIN_EMAIL) is None:
    
                crud.create_user(db, schemas.UserCreate(email=config.ADMIN_EMAIL, nickname=config.ADMIN_NICKNAME,
                                 password=config.ADMIN_PASSWORD, type=models.UserType.ADMIN.value))
    
    Base.metadata.create_all(bind=engine)
    
    app = FastAPI(lifespan=lifespan)
    
    
    app.add_middleware(
        CORSMiddleware,
        allow_origins=config.ALLOWED_ORIGINS,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    apiRouter = APIRouter(prefix="/api")
    v1Router = APIRouter(prefix="/v1")
    authRouter = APIRouter(prefix="/auth", tags=["auth"])
    usersRouter = APIRouter(prefix="/users", tags=["users"])
    sessionsRouter = APIRouter(prefix="/sessions", tags=["sessions"])
    
    websocketRouter = APIRouter(prefix="/ws", tags=["websocket"])
    
    webhookRouter = APIRouter(prefix="/webhooks", tags=["webhook"])
    
    stats = APIRouter(prefix="/stats", tags=["stats"])
    
    @app.get("/health", status_code=status.HTTP_204_NO_CONTENT)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def health():
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @authRouter.post("/login", response_model=schemas.Token)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def login(email: Annotated[str, Form()], password: Annotated[str, Form()], db: Session = Depends(get_db)):
        db_user = crud.get_user_by_email_and_password(db, email, password)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
        if db_user is None:
    
            raise HTTPException(
                status_code=401, detail="Incorrect email or password")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        return {
            "access_token": hashing.create_access_token(db_user),
            "refresh_token": hashing.create_refresh_token(db_user),
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @authRouter.post("/register", status_code=status.HTTP_201_CREATED)
    def register(email: Annotated[str, Form()], password: Annotated[str, Form()], nickname: Annotated[str, Form()], db: Session = Depends(get_db)):
        db_user = crud.get_user_by_email(db, email=email)
        if db_user:
            raise HTTPException(status_code=400, detail="User already registered")
    
    
        user_data = schemas.UserCreate(
            email=email, password=password, nickname=nickname)
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
        user = crud.create_user(db=db, user=user_data)
    
        return user.id
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @authRouter.post("/refresh", response_model=schemas.Token)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def refresh_token(current_user: models.User = Depends(hashing.get_jwt_user_from_refresh_token)):
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
        return {
            "access_token": hashing.create_access_token(current_user),
            "refresh_token": hashing.create_refresh_token(current_user),
        }
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @usersRouter.post("", status_code=status.HTTP_201_CREATED)
    
    def create_user(user: schemas.UserCreate, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to create a user")
    
        db_user = crud.get_user_by_email(db, email=user.email)
    
        if db_user:
            raise HTTPException(status_code=400, detail="User already registered")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
        user = crud.create_user(db=db, user=user)
    
        return user.id
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @usersRouter.get("/{user_id}", response_model=schemas.User)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def read_user(user_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user.id != user_id:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to view this user")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
        db_user = crud.get_user(db, user_id=user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
        return db_user
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @usersRouter.get("", response_model=list[schemas.User])
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to view users")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        return crud.get_users(db, skip=skip, limit=limit)
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @usersRouter.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def delete_user(user_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to delete a user")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        if not crud.delete_user(db, user_id):
            raise HTTPException(status_code=404, detail="User not found")
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @usersRouter.get("/{user_id}/sessions", response_model=list[schemas.Session])
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def read_user_sessions(user_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user.id != user_id:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to view this user's sessions")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        db_user = crud.get_user(db, user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
        return db_user.sessions
    
    
    @usersRouter.post("/{user_id}/metadata", status_code=status.HTTP_201_CREATED)
    def create_user_metadata(user_id: int, metadata: schemas.UserMetadataCreate, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user.id != user_id:
            raise HTTPException(
                status_code=401, detail="You do not have permission to create metadata for this user")
    
        db_user = crud.get_user(db, user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
        return crud.create_user_metadata(db, db_user.id, metadata).id
    
    
    
    @usersRouter.get("/{user_id}/metadata", response_model=schemas.UserMetadata)
    def read_user_metadata(user_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user.id != user_id:
            raise HTTPException(
                status_code=401, detail="You do not have permission to view this user's metadata")
    
        db_user = crud.get_user(db, user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
        metadata = crud.get_user_metadata(db, db_user.id)
    
        if metadata is None:
            raise HTTPException(status_code=404, detail="Metadata not found")
    
        return metadata
    
    
    
    @usersRouter.post("/{user_id}/tests/typing", status_code=status.HTTP_201_CREATED)
    def create_test_typing(user_id: int, test: schemas.TestTypingCreate, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user.id != user_id:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to create a test for this user")
    
    
        db_user = crud.get_user(db, user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
        return crud.create_test_typing(db, test, db_user).id
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.post("", response_model=schemas.Session)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def create_session(db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.TUTOR):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to create a session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        return crud.create_session(db, current_user)
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.get("/{session_id}", response_model=schemas.Session)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def read_session(session_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user not in db_session.users:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to view this session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        return db_session
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.delete("/{session_id}", status_code=status.HTTP_204_NO_CONTENT)
    
    def delete_session(session_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        if not check_user_level(current_user, models.UserType.ADMIN):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to delete a session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
        crud.delete_session(db, session_id)
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.patch("/{session_id}", status_code=status.HTTP_204_NO_CONTENT)
    def update_session(session_id: int, session: schemas.SessionUpdate, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        if not check_user_level(current_user, models.UserType.ADMIN) and (current_user.type != models.UserType.TUTOR or current_user not in db_session.users):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to update this session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        crud.update_session(db, session, session_id)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.post("/{session_id}/users/{user_id}", status_code=status.HTTP_201_CREATED)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def add_user_to_session(session_id: int, user_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN) and (current_user.id != user_id or current_user.type != models.UserType.TUTOR):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to add a user to a session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        db_user = crud.get_user(db, user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
        db_session.users.append(db_user)
        db.commit()
        db.refresh(db_session)
    
    
    @sessionsRouter.delete("/{session_id}/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def remove_user_from_session(session_id: int, user_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if not check_user_level(current_user, models.UserType.ADMIN) and (current_user.id != user_id or current_user.type != models.UserType.TUTOR):
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to remove a user from a session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        db_user = crud.get_user(db, user_id)
        if db_user is None:
            raise HTTPException(status_code=404, detail="User not found")
    
        db_session.users.remove(db_user)
        db.commit()
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.get("", response_model=list[schemas.Session])
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    def read_sessions(skip: int = 0, limit: int = 100, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
        if check_user_level(current_user, models.UserType.ADMIN):
            return crud.get_all_sessions(db, skip=skip, limit=limit)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
        return crud.get_sessions(db, current_user, skip=skip, limit=limit)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.get("/{session_id}/messages", response_model=list[schemas.Message])
    def read_messages(session_id: int, skip: int = 0, limit: int = 100, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user not in db_session.users:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to view messages in this session")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    
        return crud.get_messages(db, session_id, skip=skip, limit=limit)
    
    
    async def send_websoket_message(session_id: int, message: schemas.Message):
    
    
        content = json.dumps({
            "type": "message",
            "action": "create",
    
            "data": message.to_dict()
    
        })
    
        for _, user_websockets in websocket_users[session_id].items():
            for user_websocket in user_websockets:
    
                await user_websocket.send_text(content)
    
    def store_metadata(db, message_id, metadata: list[schemas.MessageMetadataCreate]):
        for m in metadata:
            crud.create_message_metadata(db, message_id, m)
        pass
    
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    @sessionsRouter.post("/{session_id}/messages", status_code=status.HTTP_201_CREATED)
    
    def create_message(session_id: int, entryMessage: schemas.MessageCreate, background_tasks: BackgroundTasks, db: Session = Depends(get_db), current_user: schemas.User = Depends(hashing.get_jwt_user)):
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user not in db_session.users:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to create a message in this session")
    
        message = crud.create_message(db, entryMessage, current_user, db_session)
    
        background_tasks.add_task(
            store_metadata, db, message.id, entryMessage.metadata)
        background_tasks.add_task(
            send_websoket_message, session_id, schemas.Message.model_validate(message))
    
    @websocketRouter.websocket("/{token}/{session_id}")
    async def websocket_session(token: str, session_id: int, websocket: WebSocket, db: Session = Depends(get_db)):
        current_user = hashing.get_jwt_user(token=token, db=db)
    
        if current_user is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    
    
        db_session = crud.get_session(db, session_id)
        if db_session is None:
            raise HTTPException(status_code=404, detail="Session not found")
    
        if not check_user_level(current_user, models.UserType.ADMIN) and current_user not in db_session.users:
    
            raise HTTPException(
                status_code=401, detail="You do not have permission to access this session")
    
    
        await websocket.accept()
    
        websocket_users[session_id][current_user.id].add(websocket)
    
        try:
            while True:
                data = await websocket.receive_text()
                for user_id, user_websockets in websocket_users[session_id].items():
                    if user_id == current_user.id:
                        continue
    
                    for user_websocket in user_websockets:
                        await user_websocket.send_text(data)
        except:
            websocket_users[session_id][current_user.id].remove(websocket)
    
            try:
                await websocket.close()
            except:
                pass
    
    
    @webhookRouter.post("/sessions", status_code=status.HTTP_202_ACCEPTED)
    
    async def webhook_session(webhook: schemas.CalComWebhook, x_cal_signature_256: Annotated[str | None, Header()] = None, db: Session = Depends(get_db)):
        if x_cal_signature_256 != config.CALCOM_SECRET:
            raise HTTPException(status_code=401, detail="Invalid secret")
    
        if webhook.triggerEvent == "BOOKING_CREATED":
    
            start_time = datetime.datetime.fromisoformat(
                webhook.payload["startTime"])
    
            start_time -= datetime.timedelta(hours=1)
    
            end_time = datetime.datetime.fromisoformat(webhook.payload["endTime"])
    
            end_time += datetime.timedelta(hours=1)
            attendes = webhook.payload["attendees"]
    
            emails = [attendee["email"]
                      for attendee in attendes if attendee != None]
            db_users = [crud.get_user_by_email(db, email)
                        for email in emails if email != None]
            users = [user for user in db_users if user != None]
    
            if users:
                db_session = crud.create_session_with_users(
                    db, users, start_time, end_time)
            else:
                raise HTTPException(status_code=404, detail="Users not found")
    
    
        raise HTTPException(status_code=400, detail="Invalid trigger event")
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    v1Router.include_router(authRouter)
    v1Router.include_router(usersRouter)
    v1Router.include_router(sessionsRouter)
    
    v1Router.include_router(websocketRouter)
    
    v1Router.include_router(webhookRouter)
    
    Brieuc Dubois's avatar
    Brieuc Dubois a validé
    apiRouter.include_router(v1Router)
    app.include_router(apiRouter)