generateur_v3/backend/api/routes/exercices/routes.py
Kilton937342 4941b5a154 tests
2023-02-22 12:43:39 +01:00

251 lines
11 KiB
Python

import csv
import io
from enum import Enum
from typing import List
from fastapi import APIRouter, Depends, Query, UploadFile, HTTPException, status
from fastapi.responses import FileResponse, StreamingResponse
from fastapi_pagination.ext.sqlalchemy_future import paginate as p
from pydantic import BaseModel
from sqlmodel import Session, select
from database.auth.models import User
from database.db import get_session
from database.exercices.crud import add_tags_db, check_exercice_author, check_private, check_tag_author, create_exo_db, \
delete_exo_db, get_exo_dependency, clone_exo_db, remove_tag_db, serialize_exo, update_exo_db, get_tags_dependency
from database.exercices.models import Exercice, ExerciceCreate, ExerciceEdit, ExerciceReadFull, ExercicesTagLink, Tag, \
TagCreate, TagRead, ExerciceRead
from generateur.generateur_csv import Csv_generator
from services.auth import get_current_user, get_current_user_optional
from services.exoValidation import validate_file, validate_file_optionnal
from services.io import add_fast_api_root, get_filename_from_path
from services.models import Page
router = APIRouter(tags=['exercices'])
class ExoType(str, Enum):
csv = "csv"
pdf = "pdf"
web = "web"
def filter_exo_by_tags(exos: List[tuple[Exercice, str]], tags: List[Tag]):
valid_exos = [exo for exo, tag in exos if all(
str(t) in tag.split(',') for t in tags)]
return valid_exos
def queryFilters_dependency(search: str = "", tags: List[str] | None = Depends(get_tags_dependency),
type: ExoType | None = Query(default=None)):
return search, tags, type
@router.post('/exercices', response_model=ExerciceReadFull, status_code=status.HTTP_201_CREATED)
def create_exo(exercice: ExerciceCreate = Depends(ExerciceCreate.as_form), file: UploadFile = Depends(validate_file),
user: User = Depends(get_current_user), db: Session = Depends(get_session)):
file_obj = file['file'].file._file
file_obj.name = file['file'].filename
exo_obj = create_exo_db(exercice=exercice, user=user,
exo_source=file_obj, supports=file['supports'], db=db)
return serialize_exo(exo=exo_obj, user_id=user.id, db=db)
@router.post('/clone/{id_code}', response_model=ExerciceReadFull)
def clone_exo(exercice: Exercice | None = Depends(check_private), user: User = Depends(get_current_user),
db: Session = Depends(get_session)):
if not exercice:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail={
"Exercice introuvable"})
exo_obj = clone_exo_db(exercice, user, db)
if type(exo_obj) == str:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail=exo_obj)
return serialize_exo(exo=exo_obj, user_id=user.id, db=db)
@router.get('/exercices/user', response_model=Page[ExerciceRead | ExerciceReadFull])
def get_user_exercices(user: User = Depends(get_current_user),
queryFilters: tuple[str, List[int] | None, ExoType | None] = Depends(queryFilters_dependency),
db: Session = Depends(get_session)):
search, tags, type = queryFilters
statement = select(Exercice)
statement = statement.where(Exercice.author_id == user.id)
statement = statement.where(Exercice.name.startswith(search))
if type == ExoType.csv:
statement = statement.where(Exercice.csv == True)
if type == ExoType.pdf:
statement = statement.where(Exercice.pdf == True)
if type == ExoType.web:
statement = statement.where(Exercice.web == True)
for t in tags:
sub = select(ExercicesTagLink).where(ExercicesTagLink.exercice_id == Exercice.id).where(
ExercicesTagLink.tag_id == t).exists()
statement = statement.where(sub)
page = p(db, statement)
exercices = page.items
page.items = [
serialize_exo(exo=e, user_id=user.id, db=db) for e in exercices]
return page
@router.get('/exercices/public', response_model=Page[ExerciceRead | ExerciceReadFull])
def get_public_exercices(user: User | None = Depends(get_current_user_optional),
queryFilters: tuple[str, List[int] | None] = Depends(queryFilters_dependency),
db: Session = Depends(get_session)):
search, tags, type = queryFilters
if user is not None:
statement = select(Exercice)
statement = statement.where(Exercice.author_id != user.id)
statement = statement.where(Exercice.private == False)
statement = statement.where(Exercice.origin_id == None)
statement = statement.where(Exercice.name.startswith(search))
if type == ExoType.csv:
statement = statement.where(Exercice.csv == True)
if type == ExoType.pdf:
statement = statement.where(Exercice.pdf == True)
if type == ExoType.web:
statement = statement.where(Exercice.web == True)
for t in tags:
sub = select(ExercicesTagLink).where(ExercicesTagLink.exercice_id == Exercice.id).where(
ExercicesTagLink.tag_id == t).exists()
statement = statement.where(sub)
page = p(db, statement)
print('¨PAGE', page)
exercices = page.items
page.items = [
serialize_exo(exo=e, user_id=user.id, db=db) for e in exercices]
return page
else:
statement = select(Exercice)
statement = statement.where(Exercice.private == False)
statement = statement.where(Exercice.name.startswith(search))
if type == ExoType.csv:
statement = statement.where(Exercice.csv == True)
if type == ExoType.pdf:
statement = statement.where(Exercice.pdf == True)
if type == ExoType.web:
statement = statement.where(Exercice.web == True)
page = p(db, statement)
exercices = page.items
page.items = [
serialize_exo(exo=e, user_id=None, db=db) for e in exercices]
return page
@router.get('/exercice/{id_code}', response_model=ExerciceReadFull)
def get_exercice(exo: Exercice = Depends(check_private), user: User | None = Depends(get_current_user_optional),
db: Session = Depends(get_session)):
return serialize_exo(exo=exo, user_id=getattr(user, 'id', None), db=db)
@router.put('/exercice/{id_code}', response_model=ExerciceReadFull)
def update_exo(file: UploadFile = Depends(validate_file_optionnal), exo: Exercice = Depends(check_exercice_author),
exercice: ExerciceEdit = Depends(ExerciceEdit.as_form), db: Session = Depends(get_session)):
if exo is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail='Exercice introuvable')
if exo == False:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
detail='Cet exercice ne vous appartient pas')
file_obj = None
if file:
file_obj = file["file"].file._file
file_obj.name = file['file'].filename
exo_obj = update_exo_db(exo, exercice, file['supports'] if file is not None else None, file_obj, db)
return serialize_exo(exo=exo_obj, user_id=exo_obj.author_id, db=db)
@router.delete('/exercice/{id_code}')
def delete_exercice(exercice: Exercice | bool | None = Depends(check_exercice_author),
db: Session = Depends(get_session)):
if exercice is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail="Exercice introuvable")
if exercice == False:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
detail='Cet exercice ne vous appartient pas')
delete_exo_db(exercice, db)
return {'detail': 'Exercice supprimé avec succès'}
class NewTags(BaseModel):
exo: ExerciceReadFull
tags: list[TagRead]
@router.post('/exercice/{id_code}/tags', response_model=NewTags, tags=['tags'])
def add_tags(tags: List[TagCreate], exo: Exercice | None = Depends(get_exo_dependency),
db: Session = Depends(get_session), user: User = Depends(get_current_user)):
if exo is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail='Exercice introuvable')
exo_obj, new = add_tags_db(exo, tags, user, db)
return {"exo": serialize_exo(exo=exo_obj, user_id=user.id, db=db), "tags": new}
@router.delete('/exercice/{id_code}/tags/{tag_id}', response_model=ExerciceReadFull, tags=['tags'])
def remove_tag(exo: Exercice | None = Depends(get_exo_dependency), tag: Tag | None | bool = Depends(check_tag_author),
db: Session = Depends(get_session)):
if exo is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail='Exercice introuvable')
if tag is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail='Tag introuvable')
if tag == False:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
detail="Vous n'êtes pas le propriétaire du tag")
exo_obj = remove_tag_db(exo, tag, db)
return serialize_exo(exo=exo_obj, user_id=tag.author_id, db=db)
@router.get('/exercice/{id_code}/exo_source')
async def get_exo_source(exo: Exercice = Depends(check_exercice_author)):
if exo is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
detail='Exercice introuvable')
if exo == False:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
detail='Cet exercice ne vous appartient pas')
path = add_fast_api_root(exo.exo_source)
filename = get_filename_from_path(path)
return FileResponse(path, headers={'content-disposition': 'attachment;filename=' + filename})
@router.get('/tags', response_model=List[TagRead], tags=['tags'])
def get_tags(user: User = Depends(get_current_user), db: Session = Depends(get_session)):
return user.tags
@router.get('/generator/csv/{id_code}')
async def generate_csv(*, exo: Exercice | None = Depends(get_exo_dependency), filename: str):
if exo.csv is False:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
detail='Impossible de générer cet exercice sur dans ce format')
source_path = add_fast_api_root(exo.exo_source)
consigne = exo.consigne
buffer = io.StringIO()
writer = csv.writer(buffer, delimiter=',',
quotechar=',', quoting=csv.QUOTE_MINIMAL, dialect='excel') # mettre | comme sep un jour
Csv_generator(source_path, 10, 10, 12, consigne, writer)
return StreamingResponse(iter([buffer.getvalue()]),
headers={"Content-Disposition": f'attachment;filename={filename}'}, media_type='text/csv')