added test&postman requests

Author: Asbuga

app/api/app.py

@@ -3,14 +3,14 @@
 from core.startup import on_startup, on_shutdown
 from core.users import current_active_user
 from models.user import User
-from app.services.convertor_router import file_router
+from services.convert_router import convert_router
 
 from .user_router import api_router
 
 
 app = FastAPI()
 app.include_router(api_router)
-app.include_router(file_router, prefix="/files", tags=["files"])
+app.include_router(convert_router, prefix="/files", tags=["files"])
 
 
 @app.get("/authenticated-route")

app/services/convert_router.py

@@ -0,0 +1,122 @@
+import os
+import shutil
+
+from fastapi import APIRouter, Depends, Form, File, UploadFile, HTTPException
+from fastapi.responses import FileResponse
+
+from core.config import file_storage_settings as settings
+from core.users import current_active_user
+from models.user import User
+from schemas.schemas import ImageProcessingOptions
+
+from .csv_to_xlsx_convertor import convert_csv_xlsx
+from .docx_to_pdf_convertor import write_docx, write_pdf
+from .image_convertor import process_image
+
+
+convert_router = APIRouter()
+UPLOAD_DIR = settings.MEDIA_DIR
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+
+
+@convert_router.post("/convert/image")
+async def convert_image(
+    file: UploadFile = File(...),
+    resize: str = Form(None),
+    convert_to: str = Form(None),
+    grayscale: bool = Form(None),
+    flip: str = Form(None),
+    user: User = Depends(current_active_user)
+):
+    options = ImageProcessingOptions(
+        resize=resize,
+        convert_to=convert_to,
+        grayscale=grayscale,
+        flip=flip
+    )
+
+    user_folder = f"{UPLOAD_DIR}/{user.id}"
+    os.makedirs(user_folder, exist_ok=True)
+
+    file_location = os.path.join(user_folder, file.filename)
+    
+    with open(file_location, "wb") as buffers:
+        shutil.copyfileobj(file.file, buffers)
+
+    processed_file_location = process_image(file_location, options)
+    return {
+        "filename": os.path.basename(processed_file_location), 
+        "location": processed_file_location
+    }
+
+
+@convert_router.post("/convert/csv-xlsx")
+async def convert_csv_xlsx_file(
+    file: UploadFile = File(...), 
+    user: User = Depends(current_active_user)
+):
+    user_folder = os.path.join(str(UPLOAD_DIR), str(user.id))
+    os.makedirs(user_folder, exist_ok=True)
+
+    file_location = os.path.join(user_folder, file.filename)
+
+
+    with open(file_location, "wb") as buffers:
+        shutil.copyfileobj(file.file, buffers)
+    
+    filename = convert_csv_xlsx(file_location)
+    
+    return {"filename": filename}
+
+
+@convert_router.post("/convert/pdf-docs")
+async def convert_pdf_word_file(
+    file: UploadFile = File(...), 
+    user: User = Depends(current_active_user)
+):
+    user_folder = f"{UPLOAD_DIR}/{user.id}"
+    os.makedirs(user_folder, exist_ok=True)
+
+    file_location = os.path.join(user_folder, file.filename)
+    file_docs = file_location.split(".")[0] + ".docs"
+    processed_file_location = write_docx(
+                                file_word=file_docs, file_pdf=file_location
+                                )
+    return {
+        "filename": os.path.basename(processed_file_location), 
+        "location": processed_file_location
+    }
+
+
+@convert_router.post("/convert/docs-pdf")
+async def convert_pdf_word_file(
+    file: UploadFile = File(...), 
+    user: User = Depends(current_active_user)
+):
+    user_folder = f"{UPLOAD_DIR}/{user.id}"
+    os.makedirs(user_folder, exist_ok=True)
+
+    file_location = os.path.join(user_folder, file.filename)
+    file_pdf = file_location.split(".")[0] + ".pdf"
+    processed_file_location = write_pdf(
+                                file_pdf=file_pdf, file_word=file_location
+                                )
+    return {
+        "filename": os.path.basename(processed_file_location), 
+        "location": processed_file_location
+    }
+
+
+@convert_router.get("/download/{filename}", response_class=FileResponse)
+async def download_file(filename: str, user: User = Depends(current_active_user)):
+    user_folder = os.path.join(str(UPLOAD_DIR), str(user.id))
+    file_location = os.path.join(user_folder, filename)
+    if not os.path.exists(file_location):
+        raise HTTPException(status_code=404, detail="File not found")
+    return {
+        "message": [
+            user_folder,
+            file_location
+        ]
+    }
+    return FileResponse(file_location)

app/services/convertor_router.py

@@ -21,20 +21,12 @@ def convert_csv_xlsx(filename: FilePath) -> None:
     if re.findall(_csv_filetype, filename):
         file_xlsx = re.sub(_csv_filetype, ".xlsx", filename)
         pd.read_csv(filename).to_excel(file_xlsx, index=False)
+        return file_xlsx
 
     elif re.findall(_xlsx_filetype, filename):
         file_csv = re.sub(_xlsx_filetype, ".csv", filename)
         pd.read_excel(filename).to_csv(file_csv, index=False)
+        return file_csv
 
     else:
         raise ValueError("This file type is not supported.\n")
-
-
-if __name__ == "__main__":
-    file_csv = ".\\data\\2024-04.csv"
-    convert_csv_xlsx(file_csv)
-
-    file_xlsx = ".\\data\\2024-04_1.xlsx"
-    convert_csv_xlsx(file_xlsx)
-
-    print("Convert is successful.")

app/services/csv_to_xlsx_convertor.py

@@ -1,16 +1,11 @@
-"""Домашнє завдання 11
-
-11.2 Зробити конвертор для docx або doc у pdf та навпаки
-"""
-
 from typing import Union
 
 from docx import Document
 from fpdf import FPDF
 from PyPDF2 import PdfReader
 
 
-def _read_docx(filename: Union[str, bytes]) -> str:
+def _read_docx(file_docs: Union[str, bytes]) -> str:
     """Reads DOCX file and return only text content.
 
     Args:
@@ -19,24 +14,28 @@ def _read_docx(filename: Union[str, bytes]) -> str:
     Returns:
         str: file content
     """
-    document = Document(filename)
+    document = Document(file_docs)
     return "\n".join([line.text for line in document.paragraphs])
 
 
-def write_docx(filename: Union[str, bytes], content: str) -> None:
+def write_docx(
+        file_docs: Union[str, bytes], 
+        file_pdf: Union[str, bytes]
+    ) -> None:
     """Writes textual content to a DOCX file.
 
     Args:
-        filename (Union[str, bytes]): filename or file path
-        content (str): textual content to write to a DOCX file
+        file_docs (Union[str, bytes]): filename or file path
+        file_pdf (Union[str, bytes]): filename or file path
     """
+    content=_read_pdf(file_pdf)
     document = Document()
     for line in content.split("\n"):
         document.add_paragraph(line)
-    document.save(filename)
+    document.save(file_docs)
 
 
-def _read_pdf(filename: Union[str, bytes]) -> str:
+def _read_pdf(file_pdf: Union[str, bytes]) -> str:
     """Reads DOCX file and return only text content.
 
     Args:
@@ -45,36 +44,25 @@ def _read_pdf(filename: Union[str, bytes]) -> str:
     Returns:
         str: file content
     """
-    reader = PdfReader(stream=filename)
+    reader = PdfReader(stream=file_pdf)
     return "\n".join(
         [reader.pages[i].extract_text() for i in range(0, len(reader.pages))]
     )
 
 
-def write_pdf(filename: Union[str, bytes], content: str) -> None:
+def write_pdf(
+        file_pdf: Union[str, bytes], 
+        file_word: Union[str, bytes]
+    ) -> None:
     """Writes textual content to a PDF file.
 
     Args:
-        filename (Union[str, bytes]): filename or file path
-        content (str): textual content to write to a PDF file
+        file_pdf (Union[str, bytes]): filename or file path
+        file_word (Union[str, bytes]): filename or file path
     """
+    content = _read_docx(file_word)
     pdf = FPDF()
     pdf.set_font("Arial", size=12)
     pdf.add_page()
     pdf.write(5, content)
-    pdf.output(filename)
-
-
-if __name__ == "__main__":
-    # Let's set the path to the files for the presentation
-    file_docs = ".\\data\\data.docx"
-    file_pdf = ".\\data\\data.pdf"
-    _file_to_docs = ".\\data\\to_docx.docx"
-
-    # Convert DOCX to PDF
-    write_pdf(filename=file_pdf, content=_read_docx(file_docs))
-    print("Convert DOCX to PDF is successfully!")
-
-    # Convert PDF to DOCX
-    write_docx(filename=_file_to_docs, content=_read_pdf(file_pdf))
-    print("Convert PDF to DOCX is successfully!")
+    pdf.output(file_pdf)

app/services/docx_to_pdf_convertor.py

@@ -165,21 +165,6 @@ async def test_delete_another_user_data(
     assert not user, "User was not deleted from the database"
 
 
-async def test_upload_file(async_client, get_token):
-    with open("tests/media/python.png", "rb") as f:
-        files = {"file": ("python.png", f, "image/png")}
-        data = {"resize": "500x500"}
-
-        headers = {"Authorization": f"Bearer {get_token}"}
-        image_upload_response = await async_client.post(
-            "/files/upload/", files=files, data=data, headers=headers
-        )
-
-    assert image_upload_response.status_code == status.HTTP_200_OK, (
-        "Upload file failed: " + image_upload_response.text
-    )
-
-
 async def test_logout(async_client, get_token):
     headers = {
         "Authorization": f"Bearer {get_token}",

tests/media/hello_world.xlsx

@@ -0,0 +1,4 @@
+from fastapi import status
+from sqlalchemy import select
+
+from tests.conftest import async_session_maker