diff --git a/app/config.py b/app/config.py index 9f6a411f..f794629e 100644 --- a/app/config.py +++ b/app/config.py @@ -269,7 +269,7 @@ class Settings(BaseSettings): @field_validator("notification_urls", mode="before") @classmethod - def parse_notification_urls(cls, v): + def parse_notification_urls(cls, v: str | list[str]) -> list[str]: """Parse notification URLs from string or list""" if isinstance(v, str): if "," in v: @@ -281,7 +281,7 @@ class Settings(BaseSettings): @field_validator("session_secret") @classmethod - def validate_session_secret(cls, v, info): + def validate_session_secret(cls, v: str | None, info: object) -> str | None: """Validate that session_secret is set and has sufficient length when auth is enabled""" if info.data.get("auth_enabled") and not v: raise ValueError("SESSION_SECRET must be set when AUTH_ENABLED=True") diff --git a/app/database.py b/app/database.py index cd17bcf3..516ab41c 100644 --- a/app/database.py +++ b/app/database.py @@ -3,9 +3,12 @@ import logging import os +from collections.abc import Generator +from typing import Any + from sqlalchemy import create_engine, exc from sqlalchemy.engine.url import make_url -from sqlalchemy.orm import declarative_base, sessionmaker +from sqlalchemy.orm import Session, declarative_base, sessionmaker from app.config import settings @@ -19,7 +22,7 @@ engine = create_engine(DB_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) -def init_db(): +def init_db() -> None: """ Ensures the SQLite database file and its parent directory exist (if using sqlite). Then runs Base.metadata.create_all(bind=engine) to initialize tables. @@ -55,7 +58,7 @@ def init_db(): raise -def _run_schema_migrations(engine): +def _run_schema_migrations(engine: Any) -> None: """ Apply lightweight schema migrations for columns added after the initial release. Each migration is idempotent and safe to run multiple times. @@ -117,7 +120,7 @@ def _run_schema_migrations(engine): logger.warning(f"Skipping filehash unique index drop: {exc}") -def get_db(): +def get_db() -> Generator[Session, None, None]: """ Dependency for FastAPI routes or general DB usage. Yields a SQLAlchemy session, and closes it upon exit. diff --git a/app/utils/config_loader.py b/app/utils/config_loader.py index 2efe846a..6b91ea3e 100644 --- a/app/utils/config_loader.py +++ b/app/utils/config_loader.py @@ -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. diff --git a/app/utils/config_validator/masking.py b/app/utils/config_validator/masking.py index 9ea441b4..7c7f1efc 100644 --- a/app/utils/config_validator/masking.py +++ b/app/utils/config_validator/masking.py @@ -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 """ diff --git a/app/utils/config_validator/providers.py b/app/utils/config_validator/providers.py index 102d1781..3f37117d 100644 --- a/app/utils/config_validator/providers.py +++ b/app/utils/config_validator/providers.py @@ -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 """ diff --git a/app/utils/config_validator/settings_display.py b/app/utils/config_validator/settings_display.py index 24b75024..ee2a06ba 100644 --- a/app/utils/config_validator/settings_display.py +++ b/app/utils/config_validator/settings_display.py @@ -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. diff --git a/app/utils/config_validator/validators.py b/app/utils/config_validator/validators.py index 8c9ba047..394e6952 100644 --- a/app/utils/config_validator/validators.py +++ b/app/utils/config_validator/validators.py @@ -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 diff --git a/app/utils/encryption.py b/app/utils/encryption.py index 28957947..9c8cc55d 100644 --- a/app/utils/encryption.py +++ b/app/utils/encryption.py @@ -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. diff --git a/app/utils/file_operations.py b/app/utils/file_operations.py index 0dae508f..485da36d 100644 --- a/app/utils/file_operations.py +++ b/app/utils/file_operations.py @@ -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. diff --git a/app/utils/filename_utils.py b/app/utils/filename_utils.py index 17f23cde..3c1b7906 100644 --- a/app/utils/filename_utils.py +++ b/app/utils/filename_utils.py @@ -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. diff --git a/app/utils/logging.py b/app/utils/logging.py index b7ef29da..ad16765b 100644 --- a/app/utils/logging.py +++ b/app/utils/logging.py @@ -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. diff --git a/pyproject.toml b/pyproject.toml index 42792249..5646f35a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -194,6 +194,12 @@ disable_error_code = [ "call-arg", # Dynamic call signatures in framework code ] +# Stricter type-checking for utility modules +[[tool.mypy.overrides]] +module = "app.utils.*" +disallow_untyped_defs = true +disallow_incomplete_defs = true + # Coverage configuration [tool.coverage.run] source = ["app"]