Add recursive directory creation utility: Implement create_directory_recursive function in utils.py to ensure directories are created recursively if they do not exist. Update upload controllers to utilize this new function for improved error handling and code clarity.
Some checks failed
Docker Build / build (push) Has been cancelled
Some checks failed
Docker Build / build (push) Has been cancelled
This commit is contained in:
parent
571a80b725
commit
e19a43fcd5
@ -2,7 +2,7 @@
|
|||||||
from fastapi import APIRouter, HTTPException, UploadFile, File, Form
|
from fastapi import APIRouter, HTTPException, UploadFile, File, Form
|
||||||
from fastapi.responses import JSONResponse
|
from fastapi.responses import JSONResponse
|
||||||
from nc_py_api import NextcloudException
|
from nc_py_api import NextcloudException
|
||||||
from utils import get_nextcloud_client
|
from utils import get_nextcloud_client, create_directory_recursive
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
router = APIRouter()
|
router = APIRouter()
|
||||||
@ -42,22 +42,15 @@ async def upload_file(
|
|||||||
destination_path = f"/{clean_path}/{final_filename}"
|
destination_path = f"/{clean_path}/{final_filename}"
|
||||||
directory_path = f"/{clean_path}"
|
directory_path = f"/{clean_path}"
|
||||||
|
|
||||||
# Vérifier si le dossier existe, sinon le créer
|
# Créer le dossier de destination récursivement s'il n'existe pas
|
||||||
if directory_path != "/":
|
if directory_path != "/":
|
||||||
try:
|
try:
|
||||||
# Vérifier si le dossier existe
|
create_directory_recursive(nc, directory_path)
|
||||||
nc.files.by_path(directory_path)
|
except Exception as mkdir_error:
|
||||||
except NextcloudException as e:
|
raise HTTPException(
|
||||||
if "404" in str(e):
|
status_code=500,
|
||||||
# Le dossier n'existe pas, on le crée
|
detail=f"Impossible de créer le dossier '{directory_path}': {str(mkdir_error)}"
|
||||||
try:
|
)
|
||||||
nc.files.mkdir(directory_path)
|
|
||||||
print(f"📁 Dossier créé : {directory_path}")
|
|
||||||
except Exception as mkdir_error:
|
|
||||||
raise HTTPException(
|
|
||||||
status_code=500,
|
|
||||||
detail=f"Impossible de créer le dossier '{directory_path}': {str(mkdir_error)}"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Lire le contenu du fichier
|
# Lire le contenu du fichier
|
||||||
file_content = await file.read()
|
file_content = await file.read()
|
||||||
|
|||||||
@ -2,7 +2,7 @@
|
|||||||
from fastapi import APIRouter, HTTPException, Form
|
from fastapi import APIRouter, HTTPException, Form
|
||||||
from fastapi.responses import JSONResponse
|
from fastapi.responses import JSONResponse
|
||||||
from nc_py_api import NextcloudException
|
from nc_py_api import NextcloudException
|
||||||
from utils import get_nextcloud_client
|
from utils import get_nextcloud_client, create_directory_recursive
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
import base64
|
import base64
|
||||||
|
|
||||||
@ -42,22 +42,15 @@ async def upload_file_base64(
|
|||||||
destination_path = f"/{clean_path}/{filename}"
|
destination_path = f"/{clean_path}/{filename}"
|
||||||
directory_path = f"/{clean_path}"
|
directory_path = f"/{clean_path}"
|
||||||
|
|
||||||
# Vérifier si le dossier existe, sinon le créer
|
# Créer le dossier de destination récursivement s'il n'existe pas
|
||||||
if directory_path != "/":
|
if directory_path != "/":
|
||||||
try:
|
try:
|
||||||
# Vérifier si le dossier existe
|
create_directory_recursive(nc, directory_path)
|
||||||
nc.files.by_path(directory_path)
|
except Exception as mkdir_error:
|
||||||
except NextcloudException as e:
|
raise HTTPException(
|
||||||
if "404" in str(e):
|
status_code=500,
|
||||||
# Le dossier n'existe pas, on le crée
|
detail=f"Impossible de créer le dossier '{directory_path}': {str(mkdir_error)}"
|
||||||
try:
|
)
|
||||||
nc.files.mkdir(directory_path)
|
|
||||||
print(f"📁 Dossier créé : {directory_path}")
|
|
||||||
except Exception as mkdir_error:
|
|
||||||
raise HTTPException(
|
|
||||||
status_code=500,
|
|
||||||
detail=f"Impossible de créer le dossier '{directory_path}': {str(mkdir_error)}"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Décoder le fichier base64
|
# Décoder le fichier base64
|
||||||
try:
|
try:
|
||||||
|
|||||||
35
utils.py
35
utils.py
@ -1,7 +1,7 @@
|
|||||||
"""Fonctions utilitaires pour l'application"""
|
"""Fonctions utilitaires pour l'application"""
|
||||||
from typing import Dict, Any
|
from typing import Dict, Any
|
||||||
from fastapi import HTTPException
|
from fastapi import HTTPException
|
||||||
from nc_py_api import Nextcloud
|
from nc_py_api import Nextcloud, NextcloudException
|
||||||
from config import settings
|
from config import settings
|
||||||
|
|
||||||
|
|
||||||
@ -21,6 +21,39 @@ def get_nextcloud_client() -> Nextcloud:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def create_directory_recursive(nc: Nextcloud, directory_path: str) -> None:
|
||||||
|
"""
|
||||||
|
Crée un dossier et tous ses parents récursivement si nécessaire
|
||||||
|
|
||||||
|
Args:
|
||||||
|
nc: Client Nextcloud
|
||||||
|
directory_path: Chemin complet du dossier à créer (ex: /GED/Mouvements/Octobre 2025)
|
||||||
|
"""
|
||||||
|
if directory_path == "/" or directory_path == "":
|
||||||
|
return
|
||||||
|
|
||||||
|
# Vérifier si le dossier existe déjà
|
||||||
|
try:
|
||||||
|
nc.files.by_path(directory_path)
|
||||||
|
return # Le dossier existe déjà, rien à faire
|
||||||
|
except NextcloudException:
|
||||||
|
pass # Le dossier n'existe pas, on continue
|
||||||
|
|
||||||
|
# Créer le dossier parent d'abord (récursif)
|
||||||
|
parent_path = "/".join(directory_path.rstrip("/").split("/")[:-1])
|
||||||
|
if parent_path and parent_path != "/":
|
||||||
|
create_directory_recursive(nc, parent_path)
|
||||||
|
|
||||||
|
# Créer le dossier actuel
|
||||||
|
try:
|
||||||
|
nc.files.mkdir(directory_path)
|
||||||
|
print(f"📁 Dossier créé : {directory_path}")
|
||||||
|
except NextcloudException as e:
|
||||||
|
# Ignorer l'erreur 405 (Method Not Allowed) qui signifie que le dossier existe déjà
|
||||||
|
if "405" not in str(e):
|
||||||
|
raise
|
||||||
|
|
||||||
|
|
||||||
def format_file_info(file_info) -> Dict[str, Any]:
|
def format_file_info(file_info) -> Dict[str, Any]:
|
||||||
"""Formate les informations d'un fichier/dossier"""
|
"""Formate les informations d'un fichier/dossier"""
|
||||||
result = {
|
result = {
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user