feat: add strict mypy type-checking for app/utils/ module
- Add [[tool.mypy.overrides]] section for app/utils/** with disallow_untyped_defs=true - Add type annotations to all functions in app/utils/ (12 files) - Fix type annotations in app/config.py and app/database.py (imported by utils) - All 85 source files now pass mypy type checking Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -17,7 +17,7 @@ from app.models import ApplicationSettings
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def load_settings_from_db(settings_obj, db_session: Session) -> None:
|
||||
def load_settings_from_db(settings_obj: object, db_session: Session) -> None:
|
||||
"""
|
||||
Load settings from database and apply them to the settings object.
|
||||
|
||||
@@ -111,7 +111,7 @@ def convert_setting_value(value: Optional[str], field_type: Any) -> Any:
|
||||
return str(value)
|
||||
|
||||
|
||||
def reload_settings_from_db(settings_obj) -> bool:
|
||||
def reload_settings_from_db(settings_obj: object) -> bool:
|
||||
"""
|
||||
Reload settings from database.
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ Module for masking sensitive information in configuration values
|
||||
"""
|
||||
|
||||
|
||||
def mask_sensitive_value(value):
|
||||
def mask_sensitive_value(value: str | None) -> str | None:
|
||||
"""
|
||||
Masks sensitive values like API keys in logs and output
|
||||
"""
|
||||
|
||||
@@ -6,7 +6,7 @@ from app.config import settings
|
||||
from app.utils.config_validator.masking import mask_sensitive_value
|
||||
|
||||
|
||||
def get_provider_status():
|
||||
def get_provider_status() -> dict[str, dict[str, object]]:
|
||||
"""
|
||||
Returns status information for all configured providers
|
||||
"""
|
||||
|
||||
@@ -13,7 +13,7 @@ logger = logging.getLogger(__name__)
|
||||
_PYDANTIC_INTERNALS = {"model_computed_fields", "model_config", "model_extra", "model_fields", "model_fields_set"}
|
||||
|
||||
|
||||
def dump_all_settings():
|
||||
def dump_all_settings() -> None:
|
||||
"""Log all settings values for diagnostic purposes"""
|
||||
logger.info("--- DUMPING ALL SETTINGS FOR DIAGNOSTIC PURPOSES ---")
|
||||
for key in dir(settings):
|
||||
@@ -60,7 +60,7 @@ def dump_all_settings():
|
||||
logger.info("--- END OF SETTINGS DUMP ---")
|
||||
|
||||
|
||||
def get_settings_for_display(show_values=False):
|
||||
def get_settings_for_display(show_values: bool = False) -> dict[str, list[dict[str, object]]]:
|
||||
"""
|
||||
Group settings into logical categories and check if they are configured.
|
||||
Returns a dictionary with categories as keys and lists of setting items as values.
|
||||
|
||||
@@ -9,7 +9,7 @@ from app.config import settings
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def validate_email_config():
|
||||
def validate_email_config() -> list[str]:
|
||||
"""Validates email configuration settings"""
|
||||
issues = []
|
||||
|
||||
@@ -36,7 +36,7 @@ def validate_email_config():
|
||||
return issues
|
||||
|
||||
|
||||
def validate_auth_config():
|
||||
def validate_auth_config() -> list[str]:
|
||||
"""Validates authentication configuration settings"""
|
||||
issues = []
|
||||
|
||||
@@ -69,7 +69,7 @@ def validate_auth_config():
|
||||
return issues
|
||||
|
||||
|
||||
def validate_storage_configs():
|
||||
def validate_storage_configs() -> dict[str, list[str]]:
|
||||
"""Validates configuration for all storage providers"""
|
||||
issues = {}
|
||||
|
||||
@@ -178,7 +178,7 @@ def validate_storage_configs():
|
||||
return issues
|
||||
|
||||
|
||||
def validate_notification_config():
|
||||
def validate_notification_config() -> list[str]:
|
||||
"""Check notification configuration"""
|
||||
issues = []
|
||||
|
||||
@@ -210,7 +210,7 @@ def validate_notification_config():
|
||||
return issues
|
||||
|
||||
|
||||
def check_all_configs():
|
||||
def check_all_configs() -> dict[str, list[str] | dict[str, list[str]]]:
|
||||
"""Run all configuration validations and log results"""
|
||||
from app.utils.config_validator.settings_display import dump_all_settings
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
|
||||
_cipher_suite = None
|
||||
|
||||
|
||||
def _get_cipher_suite():
|
||||
def _get_cipher_suite() -> object | None:
|
||||
"""
|
||||
Get or create the Fernet cipher suite for encryption/decryption.
|
||||
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def hash_file(filepath, chunk_size=65536):
|
||||
def hash_file(filepath: str | Path, chunk_size: int = 65536) -> str:
|
||||
"""
|
||||
Returns the SHA-256 hash of the file at 'filepath'.
|
||||
Reads the file in chunks to handle large files efficiently.
|
||||
|
||||
@@ -2,13 +2,14 @@ import logging
|
||||
import os
|
||||
import re
|
||||
import uuid
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_unique_filename(original_path, check_exists_func=None):
|
||||
def get_unique_filename(original_path: str, check_exists_func: Callable[[str], bool] | None = None) -> str:
|
||||
"""
|
||||
Generates a unique filename by appending a timestamp or counter when a collision occurs.
|
||||
|
||||
@@ -69,7 +70,7 @@ def get_unique_filename(original_path, check_exists_func=None):
|
||||
return new_path
|
||||
|
||||
|
||||
def get_unique_filepath_with_counter(directory, base_filename, extension=".pdf"):
|
||||
def get_unique_filepath_with_counter(directory: str, base_filename: str, extension: str = ".pdf") -> str:
|
||||
"""
|
||||
Returns a unique filepath in the specified directory using a numeric counter suffix.
|
||||
If 'base_filename.pdf' exists, it will append '-0001', '-0002', etc.
|
||||
@@ -120,7 +121,7 @@ def get_unique_filepath_with_counter(directory, base_filename, extension=".pdf")
|
||||
return candidate
|
||||
|
||||
|
||||
def sanitize_filename(filename):
|
||||
def sanitize_filename(filename: str) -> str:
|
||||
r"""
|
||||
Sanitize a filename to ensure it's valid across different file systems
|
||||
and prevent path traversal attacks.
|
||||
@@ -162,7 +163,7 @@ def sanitize_filename(filename):
|
||||
return sanitized
|
||||
|
||||
|
||||
def extract_remote_path(file_path, base_dir, remote_base=""):
|
||||
def extract_remote_path(file_path: str, base_dir: str, remote_base: str = "") -> str:
|
||||
"""
|
||||
Extract a remote path for a file by preserving its directory structure
|
||||
relative to the base directory, but with a new remote base path.
|
||||
|
||||
+10
-3
@@ -16,9 +16,9 @@ class TaskLogCollector(logging.Handler):
|
||||
This captures all logger.info/error/warning output automatically.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._buffers = defaultdict(list)
|
||||
self._buffers: defaultdict[str, list[str]] = defaultdict(list)
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
@@ -60,7 +60,14 @@ def _ensure_collector_installed() -> None:
|
||||
_collector_installed = True
|
||||
|
||||
|
||||
def log_task_progress(task_id, step_name, status, message=None, file_id=None, detail=None):
|
||||
def log_task_progress(
|
||||
task_id: str,
|
||||
step_name: str,
|
||||
status: str,
|
||||
message: str | None = None,
|
||||
file_id: int | None = None,
|
||||
detail: str | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Logs the progress of a Celery task to the database.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user