style: lint code

Author: loadept

.github/workflows/pre-production.yml

@@ -0,0 +1,36 @@
+name: Pre Production CI
+
+on:
+  push:
+    branches: [master]
+  pull_request:
+    branches: [master]
+
+jobs:
+  lint_code:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout project
+        uses: actions/checkout@v5
+        with:
+          fetch-depth: 0
+
+      - name: Download python
+        uses: actions/setup-python@v6
+        with:
+          python-version: '3.13' 
+
+      - name: Download uv
+        uses: astral-sh/setup-uv@v6
+        with:
+          version: "latest"
+
+      - name: Install dependencies
+        run: uv sync --group dev
+
+      - name: Lint code
+        run: uv run ruff check
+
+      - name: Check format
+        run: uv run ruff format --check

app/api/v1/api_pdf.py

@@ -7,16 +7,20 @@
 router = APIRouter(prefix="/pdf", tags=["PDF Operations"])
 file_service = FileService()
 
-accepted_quality = {
-    "extreme": "screen",
-    "normal": "ebook",
-    "low": "printer"
-}
+accepted_quality = {"extreme": "screen", "normal": "ebook", "low": "printer"}
 
-@router.post("/compress", summary="Compress PDF", description="Compress a PDF file with specified quality level")
+
+@router.post(
+    "/compress",
+    summary="Compress PDF",
+    description="Compress a PDF file with specified quality level",
+)
 async def pdf_compressor(
     file: UploadFile = File(..., description="PDF file to compress"),
-    quality: Literal["extreme", "normal", "low"] = Form(..., description="Compression level: extreme (max compression), normal (balanced), low (min compression)")
+    quality: Literal["extreme", "normal", "low"] = Form(
+        ...,
+        description="Compression level: extreme (max compression), normal (balanced), low (min compression)",
+    ),
 ):
     try:
         if file.content_type != "application/pdf":
@@ -26,12 +30,14 @@ async def pdf_compressor(
         compressed_file: bytes
         actual_size = file.size
 
-        compressed_file = file_service.compress_pdf_tmp(pdf_bytes, accepted_quality[quality])
+        compressed_file = file_service.compress_pdf_tmp(
+            pdf_bytes, accepted_quality[quality]
+        )
 
         if not compressed_file:
-            raise HTTPException(500, "Compression did not generate any results")    
+            raise HTTPException(500, "Compression did not generate any results")
 
-        file_name = file.filename.removesuffix('.pdf')
+        file_name = file.filename.removesuffix(".pdf")
         compressed_size = len(compressed_file)
         reduction = ((actual_size - compressed_size) / actual_size) * 100
 
@@ -44,44 +50,48 @@ async def pdf_compressor(
                 "X-Original-Size": str(actual_size),
                 "X-Compressed-size": str(compressed_size),
                 "X-Reduction-Percent": f"{reduction:.2f}",
-                "X-Quality-Level": quality
-            }
+                "X-Quality-Level": quality,
+            },
         )
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 
-@router.post("/merge", summary="Merge PDFs", description="Merge multiple PDF files into one")
+
+@router.post(
+    "/merge", summary="Merge PDFs", description="Merge multiple PDF files into one"
+)
 async def pdf_merge(
-    files: List[UploadFile] = File(..., description="List of PDF files to merge (minimum 2)"),
+    files: List[UploadFile] = File(
+        ..., description="List of PDF files to merge (minimum 2)"
+    ),
 ):
     try:
         if len(files) < 2:
             raise HTTPException(
-                status_code=400,
-                detail="At least 2 PDF files are required for merging."
+                status_code=400, detail="At least 2 PDF files are required for merging."
             )
 
         for file in files:
             if file.content_type != "application/pdf":
                 raise HTTPException(
                     status_code=400,
-                    detail=f"Invalid file type: {file.filename} must be a PDF."
+                    detail=f"Invalid file type: {file.filename} must be a PDF.",
                 )
 
         bytes_list = [await f.read() for f in files]
 
         merged_file = file_service.merge_pdf(bytes_list)
 
         if not merged_file:
-            raise HTTPException(500, "Merge did not generate any results")    
+            raise HTTPException(500, "Merge did not generate any results")
 
         return StreamingResponse(
             BytesIO(merged_file),
             media_type="application/pdf",
             headers={
-                "Content-Disposition": f'attachment; filename="merged.pdf"',
+                "Content-Disposition": 'attachment; filename="merged.pdf"',
                 "Content-Length": str(len(merged_file)),
-            }
+            },
         )
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))

app/services/file_service.py

@@ -4,9 +4,12 @@
 import tempfile
 import logging
 
+
 class FileService:
     def compress_pdf_buffer(self, pdf_bytes: bytes, quality: str) -> bytes:
-        logging.debug(f"Compressing {len(pdf_bytes)} bytes with quality={quality} in buffer")
+        logging.debug(
+            f"Compressing {len(pdf_bytes)} bytes with quality={quality} in buffer"
+        )
 
         gs_cmd = [
             "gs",
@@ -24,7 +27,7 @@ def compress_pdf_buffer(self, pdf_bytes: bytes, quality: str) -> bytes:
             "-dFILTERTEXTANNOTATIONS=true",
             "-dFILTERIMAGEANNOTATIONS=true",
             "-sOutputFile=-",
-            "-"
+            "-",
         ]
         gs_process: subprocess.Popen = None
 
@@ -33,12 +36,14 @@ def compress_pdf_buffer(self, pdf_bytes: bytes, quality: str) -> bytes:
                 gs_cmd,
                 stdin=subprocess.PIPE,
                 stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE
+                stderr=subprocess.PIPE,
             )
 
             gs_output, gs_err = gs_process.communicate(input=pdf_bytes, timeout=60)
             if gs_process.returncode != 0:
-                raise RuntimeError(f"Error to compress file: {gs_err.decode('utf-8', 'ignore')}")
+                raise RuntimeError(
+                    f"Error to compress file: {gs_err.decode('utf-8', 'ignore')}"
+                )
 
             if not gs_output:
                 raise RuntimeError("Compression failed: no output generated")
@@ -53,7 +58,9 @@ def compress_pdf_buffer(self, pdf_bytes: bytes, quality: str) -> bytes:
             raise RuntimeError(f"Unexpected error: {str(e)}")
 
     def compress_pdf_tmp(self, pdf_bytes: bytes, quality: str) -> bytes:
-        logging.debug(f"Compressing {len(pdf_bytes)} bytes with quality={quality} in tmp disk")
+        logging.debug(
+            f"Compressing {len(pdf_bytes)} bytes with quality={quality} in tmp disk"
+        )
 
         input_path = None
         output_path = None
@@ -64,10 +71,14 @@ def compress_pdf_tmp(self, pdf_bytes: bytes, quality: str) -> bytes:
                 input_file.write(pdf_bytes)
                 input_path = input_file.name
 
-            with tempfile.NamedTemporaryFile(suffix="_gs.pdf", delete=False) as output_file:
+            with tempfile.NamedTemporaryFile(
+                suffix="_gs.pdf", delete=False
+            ) as output_file:
                 output_path = output_file.name
 
-            with tempfile.NamedTemporaryFile(suffix="_qpdf.pdf", delete=False) as output_compress_file:
+            with tempfile.NamedTemporaryFile(
+                suffix="_qpdf.pdf", delete=False
+            ) as output_compress_file:
                 output_compress_path = output_compress_file.name
 
             gs_cmd = [
@@ -86,18 +97,17 @@ def compress_pdf_tmp(self, pdf_bytes: bytes, quality: str) -> bytes:
                 "-dFILTERTEXTANNOTATIONS=true",
                 "-dFILTERIMAGEANNOTATIONS=true",
                 f"-sOutputFile={output_path}",
-                input_path
+                input_path,
             ]
 
             gs_process = subprocess.run(
-                gs_cmd,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                timeout=60
+                gs_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60
             )
 
             if gs_process.returncode != 0:
-                raise RuntimeError(f"Error to compress file: {gs_process.stderr.decode('utf-8', 'ignore')}")
+                raise RuntimeError(
+                    f"Error to compress file: {gs_process.stderr.decode('utf-8', 'ignore')}"
+                )
 
             qpdf_cmd = [
                 "qpdf",
@@ -109,14 +119,13 @@ def compress_pdf_tmp(self, pdf_bytes: bytes, quality: str) -> bytes:
                 output_compress_path,
             ]
             qpdf_process = subprocess.run(
-                qpdf_cmd,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                timeout=60
+                qpdf_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60
             )
 
             if qpdf_process.returncode != 0:
-                raise RuntimeError(f"Error to compress file: {qpdf_process.stderr.decode('utf-8', 'ignore')}")
+                raise RuntimeError(
+                    f"Error to compress file: {qpdf_process.stderr.decode('utf-8', 'ignore')}"
+                )
 
             with open(output_compress_path, "rb") as file:
                 return file.read()
@@ -132,38 +141,44 @@ def compress_pdf_tmp(self, pdf_bytes: bytes, quality: str) -> bytes:
                         unlink(file_path)
                     except Exception:
                         pass
-    
+
     def merge_pdf(self, bytes_list: List[bytes]):
         input_paths = []
         output_path = None
 
         try:
             for i, pdf_bytes in enumerate(bytes_list):
-                with tempfile.NamedTemporaryFile(suffix=f"_{i}.pdf", delete=False) as input_file:
+                with tempfile.NamedTemporaryFile(
+                    suffix=f"_{i}.pdf", delete=False
+                ) as input_file:
                     input_file.write(pdf_bytes)
                     input_paths.append(input_file.name)
 
-            with tempfile.NamedTemporaryFile(suffix="_gs.pdf", delete=False) as output_file:
+            with tempfile.NamedTemporaryFile(
+                suffix="_gs.pdf", delete=False
+            ) as output_file:
                 output_path = output_file.name
 
             qpdf_cmd = [
                 "qpdf",
                 "--linearize",
                 "--empty",
-                "--pages", *input_paths, "--", output_path,
+                "--pages",
+                *input_paths,
+                "--",
+                output_path,
                 "--object-streams=generate",
                 "--compress-streams=y",
-                "--recompress-flate"
+                "--recompress-flate",
             ]
             qpdf_process = subprocess.run(
-                qpdf_cmd,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                timeout=60
+                qpdf_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60
             )
 
             if qpdf_process.returncode != 0:
-                raise RuntimeError(f"Error merging PDFs: {qpdf_process.stderr.decode('utf-8', 'ignore')}")
+                raise RuntimeError(
+                    f"Error merging PDFs: {qpdf_process.stderr.decode('utf-8', 'ignore')}"
+                )
 
             with open(output_path, "rb") as merged:
                 return merged.read()

pyproject.toml

@@ -7,3 +7,8 @@ requires-python = ">=3.13"
 dependencies = [
     "fastapi[standard]>=0.118.0",
 ]
+
+[dependency-groups]
+dev = [
+    "ruff>=0.13.3",
+]