chore: Add extensive linting

Makefile

@@ -1,7 +1,7 @@
 
-all: docs test
+all: docs test format lint
 
-.PHONY: docs test
+.PHONY: docs test format lint
 
 docs:
 	@echo "Generating documentation"
@@ -10,3 +10,9 @@ docs:
 
 test:
 	pdm run python -m pytest --cov=whatsapp_transcribe tests/
+
+format:
+	pdm run ruff format .
+
+lint:
+	pdm run ruff check --fix .

bruno_collection/Whatsapp Voice Transcriber/Transcribe.bru

@@ -7,9 +7,13 @@ meta {
 post {
   url: {{baseURL}}/transcribe
   body: multipartForm
-  auth: none
+  auth: bearer
+}
+
+auth:bearer {
+  token: {{apiKey}}
 }
 
 body:multipart-form {
-  voice_message: @file(WhatsApp Ptt 2024-02-26 at 12.08.03.ogg)
+  voice_message: @file(/Users/sebbo/Downloads/vani_transcription/WhatsApp Audio 2024-02-08 at 11.00.32.ogg)
 }

pyproject.toml

@@ -36,3 +36,7 @@ dev = [
     "pytest-cov>=5.0.0",
     "pdoc>=14.5.0",
 ]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "UP"]
+ignore = ["D203", "D212", "N803"]

src/whatsapp_transcribe/__init__.py

@@ -0,0 +1,4 @@
+"""
+Whatsapp Transcribe Python Moudle implements a FastAPI server  for
+whatsap voice message transcription.
+"""

src/whatsapp_transcribe/__main__.py

@@ -1,23 +1,24 @@
+import logging
+from dataclasses import dataclass
+from typing import Annotated
+
 from fastapi import (
+    BackgroundTasks,
     FastAPI,
-    UploadFile,
-    Request,
-    HTTPException,
-    Header,
     Form,
+    Header,
+    HTTPException,
+    Request,
     Response,
-    BackgroundTasks,
     Security,
+    UploadFile,
 )
-from .transcribe import TranscriptionService
-from .twilio_client import TwilioClient
-from typing import Annotated
-import logging
-from dataclasses import dataclass
-from .summarize import Summarizer
-from .authentication import setup_api_key_auth
 from fastapi.security.api_key import APIKey
 
+from .authentication import setup_api_key_auth
+from .summarize import Summarizer
+from .transcribe import TranscriptionService
+from .twilio_client import TwilioClient
 
 transcription_service = TranscriptionService()
 summarization_service = Summarizer()
@@ -75,7 +76,7 @@ async def twilio_whatsapp(
 ):
     logger.info(f"Received message from {From}")
     form_ = await req.form()
-    twilio_client.validateRequest(form_, x_twilio_signature)
+    twilio_client.validate_request(form_, x_twilio_signature)
 
     if MediaUrl0 is None and not Body:
         return Response(content="No message or media found", status_code=400)

src/whatsapp_transcribe/authentication.py

@@ -2,11 +2,13 @@
 Authentication module
 
 This module contains functions for API key authentication in FastAPI.
-As we are building an API only service without any user login, we just use a fixed API key
+As we are building an API only service without any user login,
+we just use a fixed API key
 that could be rotated frequently.
 """
 
 import os
+
 from fastapi import Depends, HTTPException
 from fastapi.security.api_key import APIKeyHeader
 
@@ -15,28 +17,31 @@ def setup_api_key_auth():
     """
     Setup API key authentication
 
-    Loads the API key from the environment variable API_KEY and returns a dependency for API key authentication.
+    Loads the API key from the environment variable API_KEY and
+    returns a dependency for API key authentication.
     Allows access to the API only if the API key is correct
 
-    Returns:
+    Returns
+    -------
         Dependency for API key authentication
 
-    Raises:
+    Raises
+    ------
         ValueError: If API_KEY environment variable is not set
 
     """
     # Load API key from environment variable
-    API_KEY = os.getenv("API_KEY")
-    if not API_KEY:
+    api_key = os.getenv("API_KEY")
+    if not api_key:
         raise ValueError("API_KEY environment variable not set")
 
-    API_KEY_NAME = "Authorization"
-    api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
+    api_key_name = "Authorization"
+    api_key_header = APIKeyHeader(name=api_key_name, auto_error=False)
 
     async def get_api_key(
         api_key_header: str = Depends(api_key_header),
     ):
-        if api_key_header == f"Bearer {API_KEY}":
+        if api_key_header == f"Bearer {api_key}":
             return api_key_header
         else:
             raise HTTPException(

src/whatsapp_transcribe/summarize.py

@@ -1,21 +1,22 @@
 #!/usr/bin/env python3
-from langchain_core.prompts import PromptTemplate
-
-from textwrap import dedent
 import logging
-from langchain_community.chat_models import ChatOllama
 from os import environ
+from textwrap import dedent
 from typing import Optional
 
+from langchain_community.chat_models import ChatOllama
+from langchain_core.prompts import PromptTemplate
+
 logger = logging.getLogger(__name__)
 
 
 class Summarizer:
-    PROMPT_TEMPLATE = dedent("""Schreibe eine kurze deutsche Version der folgenden Sprachnachricht:
-                                 "{text}"
-                                 Wenn möglich, gliedere die Nachricht in Abschnitte und ergänze Überschriften.
-                                 Gebe nur den gekürzten Text zurück:
-                                 """)
+    PROMPT_TEMPLATE = dedent("""
+    Schreibe eine kurze deutsche Version der folgenden Sprachnachricht:
+    "{text}"
+    Wenn möglich, gliedere die Nachricht in Abschnitte und ergänze Überschriften.
+    Gebe nur den gekürzten Text zurück:
+    """)
 
     def __init__(self, model: Optional[str] = None):
         model = (

src/whatsapp_transcribe/transcribe.py

@@ -1,14 +1,16 @@
 """
 # Transcription
 
-This module contains the TranscriptionService class which is responsible for transcribing voice messages.
+This module contains the TranscriptionService class
+which is responsible for transcribing voice messages.
 """
 
-from faster_whisper import WhisperModel
 from io import BytesIO
 from os import environ
 from typing import Optional
 
+from faster_whisper import WhisperModel
+
 
 class TranscriptionService:
     """
@@ -21,7 +23,9 @@ def __init__(self, model_size: Optional[str] = None):
         Create a transcription service and load the transcription model into RAM.
 
         Args:
+        ----
             model_size: Whisper model size possible options: "small", "medium", "large"
+
         """
         model_size = (
             model_size

src/whatsapp_transcribe/twilio_client.py

@@ -1,31 +1,80 @@
-from twilio.rest import Client
-from twilio.request_validator import RequestValidator
-import os
-from fastapi import HTTPException, Response
-from twilio.twiml.messaging_response import MessagingResponse
-import requests
+"""
+# Twilio Client Module
+
+This module contains the TwilioClient class which is responsible for
+sending and receiving messages from Twilio.
+This encapsulates the Twilio API and provides utility functions to format messages.
+
+If we want to exchange Twilio for a custom implementation or another provider
+ in the future, we only need to change this module.
+"""
+
 import io
 import logging
+import os
 from base64 import b64encode
 
-logger = logging.getLogger(__name__)
+import requests
+from fastapi import HTTPException, Response
+from twilio.request_validator import RequestValidator
+from twilio.rest import Client
+from twilio.twiml.messaging_response import MessagingResponse
 
 
 class TwilioClient:
+    """
+    Twilio client for sending and receiving messages.
+
+    This class is responsible for sending and receiving messages from Twilio.
+    Additionally, it provides utility functions to format messages.
+
+    Use this instead of the Twilio API directly, to allow better testing and mocking.
+    """
+
     def __init__(self):
         self._auth_token = os.environ["TWILIO_AUTH_TOKEN"]
         self._account_sid = os.environ["TWILIO_ACCOUNT_SID"]
         self._webhook_url = os.environ["TWILIO_WEBHOOK_URL"]
         self.client = Client(self._account_sid, self._auth_token)
         self.validator = RequestValidator(self._auth_token)
+        self._max_len = 1500
+        self._from_number = os.environ["TWILIO_PHONE_NUMBER"]
+        self._logger = logging.getLogger(__name__)
+
+    def validate_request(self, body, signature):
+        """
+        Validate a Twilio request.
+
+        A signature is invalid  due to a malicious request or
+        wrongly configured URL, Token or Account SID.
+
+        Args:
+        ----
+            body: The body of the request.
+            signature: The signature of the request given by Twilio.
 
-    def validateRequest(self, body, signature):
+        Returns:
+        -------
+            bool: True if the request is valid, False otherwise.
+
+        Raises:
+        ------
+            HTTPException: If the request signature is invalid.
+        """
         if not self.validator.validate(str(self._webhook_url), body, signature):
-            logger.error("Invalid Twilio signature")
+            self._logger.error("Invalid Twilio signature")
             raise HTTPException(status_code=403, detail="Invalid Twilio signature")
         return True
 
-    def create_return_message(self, message):
+    def create_return_message(self, message: str):
+        """
+        Create a Twilio response message.
+
+        Args:
+        ----
+              message: The message to send. Is shortened to max_len characters.
+
+        """
         response = MessagingResponse()
         response.message(self.shorten_message(message))
         return Response(content=str(response), media_type="application/xml")
@@ -39,16 +88,32 @@ def _auth_header(self):
     def download_media(self, url):
         response = requests.get(url, headers=self._auth_header())
         if response.status_code != 200:
-            logger.error(f"Error downloading media: {response.text}")
+            self._logger.error(f"Error downloading media: {response.text}")
             raise HTTPException(status_code=500, detail="Error downloading media")
         return io.BytesIO(response.content)
 
-    def shorten_message(self, message):
-        return message[:1500] + "..." if len(message) > 1500 else message
+    def shorten_message(self, message: str):
+        """
+        Utility method to shorten strings to the maximum length defined in the class
+
+        Args:
+        ----
+            message: Message to shorten
+
+        Returns:
+        -------
+
+        """
+        return (
+            message[: self._max_len] + "..."
+            if len(message) > self._max_len
+            else message
+        )
 
-    def send_message(self, to, body):
+    def send_message(self, to: str, body: str):
+        self._logger.verbose(f"Sending message to {to}")
         self.client.messages.create(
             to=to,
-            from_=os.environ["TWILIO_PHONE_NUMBER"],
+            from_=self._from_number,
             body=self.shorten_message(body),
         )

tests/test_endpoint.py

@@ -1,7 +1,8 @@
+from os import environ
+from pathlib import Path
+
 from fastapi.testclient import TestClient
 from whatsapp_transcribe.__main__ import app
-from pathlib import Path
-from os import environ
 
 test_dir = Path(__file__).parent