chore: Satisfy mypy (71 errors in 16 files) #217

app/aws.py

@@ -8,6 +8,7 @@
 
 import boto3
 from mypy_boto3_cognito_idp import CognitoIdentityProviderClient, literals, type_defs
+from mypy_boto3_cognito_idp.client import Exceptions
 from mypy_boto3_ses.client import SESClient
 
 from app import mail
@@ -55,7 +56,7 @@ def __init__(
         #: A function reference that generates a password
         self.generate_password = generate_password_fn
 
-    def exceptions(self):
+    def exceptions(self) -> Exceptions:
         return self.client.exceptions
 
     def get_secret_hash(self, username: str) -> str:

app/commands.py

@@ -21,15 +21,17 @@
 app = typer.Typer()
 
 
-def _create_complete_application(award_response, db_provider: Callable[[], Generator[Session, None, None]]) -> None:
+def _create_complete_application(
+    award_entry: dict[str, str], db_provider: Callable[[], Generator[Session, None, None]]
+) -> None:
     with contextmanager(db_provider)() as session:
         with handle_skipped_award(session, "Error creating application"):
-            award = util.create_award_from_data_source(session, award_response)
+            award = util.create_award_from_data_source(session, award_entry)
 
             # Create a new borrower or update an existing borrower based on the entry data.
-            documento_proveedor = data_access.get_documento_proveedor(award_response)
+            documento_proveedor = data_access.get_documento_proveedor(award_entry)
             borrower_identifier = util.get_secret_hash(documento_proveedor)
-            data = data_access.get_borrower(borrower_identifier, documento_proveedor, award_response)
+            data = data_access.get_borrower(borrower_identifier, documento_proveedor, award_entry)
             if borrower := models.Borrower.first_by(session, "borrower_identifier", borrower_identifier):
                 if borrower.status == models.BorrowerStatus.DECLINE_OPPORTUNITIES:
                     raise SkippedAwardError(

app/db.py

@@ -18,7 +18,7 @@
 
 
 @contextmanager
-def rollback_on_error(session: Session) -> Generator[Session, None, None]:
+def rollback_on_error(session: Session) -> Generator[None, None, None]:
     try:
         yield
     except Exception:
@@ -27,7 +27,7 @@ def rollback_on_error(session: Session) -> Generator[Session, None, None]:
 
 
 @contextmanager
-def transaction_session(session: Session) -> Generator[Session, None, None]:
+def transaction_session(session: Session) -> Generator[None, None, None]:
     """
     Context manager for database transactions. It takes a Session instance, commits the transaction if it is
     successful, and rolls back the transaction if any exception is raised.
@@ -45,7 +45,7 @@ def transaction_session(session: Session) -> Generator[Session, None, None]:
 
 
 @contextmanager
-def handle_skipped_award(session: Session, msg: str) -> Generator[Session, None, None]:
+def handle_skipped_award(session: Session, msg: str) -> Generator[None, None, None]:
     try:
         yield
     except SkippedAwardError as e:

app/exceptions.py

@@ -1,11 +1,16 @@
+from typing import Any
+
+import httpx
+
+
 class CredereError(Exception):
     """Base class for exceptions from within this application"""
 
 
 class SkippedAwardError(CredereError):
     """Raised if an award needs to be skipped due to a data quality issue"""
 
-    def __init__(self, message, url="", data=None):
+    def __init__(self, message: str, url: str | httpx.URL = "", data: Any = None):
         self.category = "SKIPPED_AWARD"
         self.message = message
         self.url = url

app/mail.py

@@ -405,7 +405,9 @@ def send_mail_to_reset_password(ses: SESClient, username: str, temp_password: st
     return send_email(ses, username, prepare_html("Reset_password", html_data), False)
 
 
-def get_invitation_email_parameters(borrower_name, tender_title, buyer_name, uuid):
+def get_invitation_email_parameters(
+    borrower_name: str, tender_title: str, buyer_name: str, uuid: str
+) -> dict[str, str]:
     images_base_url = get_images_base_url()
     base_application_url = f"{app_settings.frontend_url}/application/{quote(uuid)}"
     base_fathom_url = "?utm_source=credere-intro&utm_medium=email&utm_campaign="

app/models.py

@@ -4,10 +4,10 @@
 from typing import Any, Optional, Self
 
 from pydantic import BaseModel, ConfigDict, PlainSerializer
-from sqlalchemy import DECIMAL, Column, DateTime, and_, desc, or_, select
+from sqlalchemy import DECIMAL, Boolean, Column, DateTime, and_, desc, or_, select
 from sqlalchemy.dialects.postgresql import JSON
 from sqlalchemy.orm import Query, Session
-from sqlalchemy.sql import Select, func
+from sqlalchemy.sql import ColumnElement, Select, func
 from sqlalchemy.sql.expression import nulls_last, true
 from sqlmodel import Field, Relationship, SQLModel, col
 from typing_extensions import Annotated
@@ -111,7 +111,7 @@ def update(self, session: Session, **data: Any) -> Self:
         return self
 
     @classmethod
-    def create_or_update(cls, session: Session, filters: list, **data: Any) -> Self:
+    def create_or_update(cls, session: Session, filters: list[bool | ColumnElement[Boolean]], **data: Any) -> Self:
         if obj := session.query(cls).filter(*filters).first():
             return obj.update(session, **data)
         return cls.create(session, **data)

app/settings.py

@@ -1,14 +1,17 @@
 import logging.config
 import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import sentry_sdk
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+if TYPE_CHECKING:
+    from sentry_sdk._types import Event
+
 VERSION: str = "0.1.1"
 
 
-def sentry_filter_transactions(event: dict[str, Any], hint: dict[str, Any]) -> dict[str, Any] | None:
+def sentry_filter_transactions(event: Event, hint: dict[str, Any]) -> Event | None:
     """
     Filter transactions to be sent to Sentry.
     This function prevents transactions that interact with AWS Cognito from being sent to Sentry.

app/sources/colombia.py

@@ -21,7 +21,7 @@
 
 
 def _get_remote_contract(
-    proceso_de_compra: str, proveedor_adjudicado: str, previous=False
+    proceso_de_compra: str, proveedor_adjudicado: str, previous: bool = False
 ) -> tuple[list[dict[str, str]], str]:
     params = f"proceso_de_compra='{proceso_de_compra}' AND documento_proveedor='{proveedor_adjudicado}'"
     contract_url = f"{URLS['CONTRACTS']}?$where={params}"
@@ -222,7 +222,7 @@ def get_borrower(borrower_identifier: str, documento_proveedor: str, entry: dict
     }
 
 
-def _get_email(borrower_response: dict):
+def _get_email(borrower_response: dict[str, str]) -> str:
     if "correo_entidad" in borrower_response:
         return borrower_response["correo_entidad"]
     return borrower_response.get("correo_electr_nico", "")

app/util.py

@@ -6,8 +6,9 @@
 from contextlib import contextmanager
 from datetime import datetime
 from enum import StrEnum
-from typing import Any, Callable, Generator
+from typing import Any, Callable, Generator, TypeVar
 
+import httpx
 import orjson
 from fastapi import File, HTTPException, UploadFile, status
 from sqlalchemy.orm import Session
@@ -19,6 +20,7 @@
 from app.settings import app_settings
 from app.sources import colombia as data_access
 
+T = TypeVar("T")
 MAX_FILE_SIZE = app_settings.max_file_size_mb * 1024 * 1024  # MB in bytes
 ALLOWED_EXTENSIONS = {".png", ".pdf", ".jpeg", ".jpg"}
 
@@ -31,17 +33,17 @@ class ERROR_CODES(StrEnum):
 
 
 # In future, httpx.Client might allow custom decoders. https://github.com/encode/httpx/issues/717
-def loads(response):
+def loads(response: httpx.Response) -> Any:
     return orjson.loads(response.text)
 
 
-def commit_and_refresh(session, instance):
+def commit_and_refresh(session: Session, instance: T) -> T:
     session.commit()
     session.refresh(instance)
     return instance
 
 
-def get_object_or_404(session: Session, model: type[models.ActiveRecordMixin], field: str, value: Any):
+def get_object_or_404(session: Session, model: type[T], field: str, value: Any) -> T:
     obj = model.first_by(session, field, value)
     if not obj:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"{model.__name__} not found")