107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
import os
|
|
import shutil
|
|
from typing import BinaryIO, List
|
|
from schemas.exercices import TagIn, TagIn_schema
|
|
from services.database import generate_unique_code
|
|
from services.io import add_fast_api_root, get_ancestor, get_filename_from_path, get_parent_dir, remove_fastapi_root, remove_if_exists, get_or_create_dir
|
|
|
|
from database.exercices.models import Exercice, Tag
|
|
from schemas.exercices import Exercice_schema
|
|
|
|
|
|
|
|
async def create_exo_db(*args, **kwargs) -> Exercice:
|
|
code = await generate_unique_code(Exercice)
|
|
return await Exercice.create(*args, **{**kwargs, 'id_code': code})
|
|
|
|
|
|
async def update_exo_db(id_code: str,exo_source: BinaryIO | None = None, **kwargs) -> Exercice:
|
|
exo = await Exercice.get(id_code=id_code)
|
|
if exo_source != None:
|
|
path = add_fast_api_root(exo.exo_source)
|
|
|
|
remove_if_exists(path)
|
|
kwargs['exo_source'] = exo_source
|
|
|
|
await exo.update_from_dict({**kwargs, 'origin_id': exo.origin_id, 'isOriginal': exo.isOriginal}).save()
|
|
return exo
|
|
|
|
|
|
async def delete_exo_db(id_code: str):
|
|
exo = await Exercice.get(id_code=id_code)
|
|
path = add_fast_api_root(exo.exo_source)
|
|
|
|
parent = get_parent_dir(path)
|
|
|
|
shutil.rmtree(parent)
|
|
|
|
return await exo.delete()
|
|
|
|
|
|
def clone_exo_source(path, id_code):
|
|
upload_root = get_ancestor(path, 2)
|
|
path = add_fast_api_root(path)
|
|
new_path = add_fast_api_root(
|
|
os.path.join(upload_root, id_code))
|
|
dir_path = get_or_create_dir(new_path)
|
|
final_path = shutil.copy(path, dir_path)
|
|
return final_path
|
|
|
|
def exclude_dict_key(dict, key):
|
|
return {k: v for k, v in dict.items() if k != key}
|
|
|
|
|
|
async def clone_exo_db(id_code: str, user_id):
|
|
exo = await Exercice.get(id_code=id_code)
|
|
new_id_code = await generate_unique_code(Exercice)
|
|
|
|
exo_obj = await Exercice_schema.from_tortoise_orm(exo)
|
|
exo_obj = exo_obj.dict(exclude_unset=True)
|
|
print("eaoiuezaoiueza", exo_obj)
|
|
exo_obj.pop('tags')
|
|
exo_obj = exclude_dict_key(exo_obj, 'tags')
|
|
exo_obj = exclude_dict_key(exo_obj, 'author')
|
|
#exo_obj.pop('exercices')
|
|
|
|
path = clone_exo_source(exo.exo_source, new_id_code)
|
|
print('\n\npatiuar', path, exo_obj, '\n',
|
|
exclude_dict_key(exo_obj, 'tags'))
|
|
new_exo = Exercice(**{**exo_obj, 'id_code': new_id_code, 'exo_source': path,
|
|
"isOriginal": False, 'author_id': user_id}, origin_id=exo.id)
|
|
|
|
await new_exo.save()
|
|
return new_exo
|
|
|
|
|
|
async def get_exo_source_path(id_code: str):
|
|
exo = await Exercice.get(id_code=id_code)
|
|
path = add_fast_api_root(exo.exo_source)
|
|
return path
|
|
|
|
#tags
|
|
async def get_or_create_tag(id_code: str, data: List[TagIn_schema]):
|
|
tag = await Tag.get_or_none(id_code=id_code)
|
|
if tag == None:
|
|
code = await generate_unique_code(Tag)
|
|
return await Tag.create(**{**data, 'id_code': code})
|
|
return tag
|
|
|
|
|
|
async def add_tag_db(exo: Exercice, tags_data: List[TagIn], user_id: int) -> Exercice:
|
|
for t in tags_data:
|
|
tag = await get_or_create_tag(t.id_code, {**t.dict(exclude_unset=True), 'owner_id': user_id})
|
|
await exo.tags.add(tag)
|
|
return exo
|
|
|
|
'''
|
|
async def add_tag_db(id_code: str, tag: Tag) -> Exercice:
|
|
exo = await Exercice.get(id_code=id_code)
|
|
await exo.tags.add(tag)
|
|
return exo
|
|
'''
|
|
async def delete_tag_db(exo_id: str, tag_id: str) -> Exercice:
|
|
exo = await Exercice.get(id_code=exo_id)
|
|
tag = await exo.tags.all().get(id_code=tag_id)
|
|
await exo.tags.remove(tag)
|
|
return exo
|