Enable mypy strict mode

.pre-commit-config.yaml

@@ -37,8 +37,12 @@ repos:
       - id: mypy
         additional_dependencies:
           [
+            anyio==4.11.0,
+            fastapi==0.117.1,
+            joserfc==1.3.4,
             pydantic==2.11.9,
             pydantic-settings==2.10.1,
+            pytest-asyncio==1.2.0,
             types-pyyaml==6.0.12.20250915,
           ]
   - repo: local

conftest.py

@@ -2,17 +2,19 @@
 import os
 import tempfile
 from pathlib import Path
+from typing import Any, AsyncGenerator, Dict
 
 import pytest_asyncio
 from aiocache import caches
+from fastapi import FastAPI
 from httpx import ASGITransport, AsyncClient
 from tortoise import Tortoise
 from tortoise.contrib.fastapi import RegisterTortoise
 
 from goosebit import app
 from goosebit.auth.permissions import GOOSEBIT_PERMISSIONS
 from goosebit.db.models import UpdateModeEnum, UpdateStateEnum
-from goosebit.settings import PWD_CXT
+from goosebit.settings import PWD_CXT  # type: ignore[attr-defined]
 
 # Configure logging
 logging.basicConfig(level=logging.WARN)
@@ -28,13 +30,13 @@
 
 
 @pytest_asyncio.fixture(scope="function", autouse=True)
-async def clear_cache():
+async def clear_cache() -> AsyncGenerator[None, None]:
     await caches.get("default").clear()
     yield
 
 
 @pytest_asyncio.fixture(scope="function")
-async def test_app():
+async def test_app() -> AsyncGenerator[FastAPI, None]:
     from goosebit.users import create_initial_user
 
     async with RegisterTortoise(
@@ -48,7 +50,7 @@ async def test_app():
 
 
 @pytest_asyncio.fixture(scope="function")
-async def async_client(test_app):
+async def async_client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]:
     async with AsyncClient(
         transport=ASGITransport(app=test_app), base_url="http://test", follow_redirects=True
     ) as client:
@@ -63,7 +65,7 @@ async def async_client(test_app):
 
 
 @pytest_asyncio.fixture(scope="function")
-async def db():
+async def db() -> AsyncGenerator[None, None]:
     await Tortoise.init(config=TORTOISE_CONF)
     await Tortoise.generate_schemas()
     yield
@@ -72,7 +74,7 @@ async def db():
 
 
 @pytest_asyncio.fixture(scope="function")
-async def test_data(db):
+async def test_data(db: None) -> AsyncGenerator[Dict[str, Any], None]:
     from goosebit.db.models import Device, Hardware, Rollout, Software, User
 
     # Create a temporary directory

docker/demo/device/myscript.py

@@ -3,7 +3,7 @@
 from time import sleep
 
 
-def main():
+def main() -> None:
     while True:
         print("Hello!", flush=True)
         sleep(5)

goosebit/__init__.py

@@ -1,13 +1,13 @@
 import importlib.metadata
 from contextlib import asynccontextmanager
 from logging import getLogger
-from typing import Annotated
+from typing import Annotated, AsyncGenerator, Awaitable, Callable
 
 from fastapi import Depends, FastAPI, HTTPException
 from fastapi.exception_handlers import http_exception_handler
 from fastapi.openapi.docs import get_swagger_ui_html
 from fastapi.requests import Request
-from fastapi.responses import RedirectResponse
+from fastapi.responses import RedirectResponse, Response
 from fastapi.security import OAuth2PasswordRequestForm
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor as Instrumentor
 from starlette.exceptions import HTTPException as StarletteHTTPException
@@ -16,7 +16,7 @@
 from goosebit import api, db, plugins, ui, updater
 from goosebit.auth import get_user_from_request, login_user, redirect_if_authenticated
 from goosebit.device_manager import DeviceManager
-from goosebit.settings import PWD_CXT, config
+from goosebit.settings import PWD_CXT, config  # type: ignore[attr-defined]
 from goosebit.ui.nav import nav
 from goosebit.ui.static import static
 from goosebit.ui.templates import templates
@@ -26,7 +26,7 @@
 
 
 @asynccontextmanager
-async def lifespan(_: FastAPI):
+async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     db_ready = await db.init()
     if not db_ready:
         logger.exception("DB does not exist, try running `poetry run aerich upgrade`.")
@@ -57,16 +57,16 @@ async def lifespan(_: FastAPI):
         }
     ],
 )
-app.include_router(updater.router)
-app.include_router(ui.router)
-app.include_router(api.router)
+app.include_router(updater.router)  # type: ignore[attr-defined]
+app.include_router(ui.router)  # type: ignore[attr-defined]
+app.include_router(api.router)  # type: ignore[attr-defined]
 app.mount("/static", static, name="static")
 Instrumentor.instrument_app(app)
 
 for plugin in plugins.load():
     if plugin.middleware is not None:
         logger.info(f"Adding middleware for plugin: {plugin.name}")
-        app.add_middleware(plugin.middleware)
+        app.add_middleware(plugin.middleware)  # type: ignore[arg-type]
     if plugin.router is not None:
         logger.info(f"Adding routing handler for plugin: {plugin.name}")
         app.include_router(router=plugin.router, prefix=plugin.url_prefix)
@@ -78,7 +78,7 @@ async def lifespan(_: FastAPI):
         app.mount(f"{plugin.url_prefix}/static", plugin.static_files, name=plugin.static_files_name)
     if plugin.templates is not None:
         logger.info(f"Adding template handler for plugin: {plugin.name}")
-        templates.add_template_handler(plugin.templates)
+        templates.add_template_handler(plugin.templates)  # type: ignore[attr-defined]
     if plugin.update_source_hook is not None:
         DeviceManager.add_update_source(plugin.update_source_hook)
     if plugin.config_data_hook is not None:
@@ -87,70 +87,70 @@ async def lifespan(_: FastAPI):
 
 # Custom exception handler for Tortoise ValidationError
 @app.exception_handler(ValidationError)
-async def tortoise_validation_exception_handler(request: Request, exc: ValidationError):
+async def tortoise_validation_exception_handler(request: Request, exc: ValidationError) -> None:
     raise HTTPException(422, str(exc))
 
 
 # Extend default handler to do logging
 @app.exception_handler(StarletteHTTPException)
-async def custom_http_exception_handler(request, exc):
+async def custom_http_exception_handler(request: Request, exc: StarletteHTTPException) -> Response:
     logger.warning(f"HTTPException, request={request.url}, status={exc.status_code}, detail={exc.detail}")
     return await http_exception_handler(request, exc)
 
 
 @app.middleware("http")
-async def attach_user(request: Request, call_next):
+async def attach_user(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
     request.scope["user"] = await get_user_from_request(request)
     return await call_next(request)
 
 
 @app.middleware("http")
-async def attach_nav(request: Request, call_next):
+async def attach_nav(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
     request.scope["nav"] = nav.get()
     return await call_next(request)
 
 
 @app.middleware("http")
-async def attach_config(request: Request, call_next):
+async def attach_config(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
     request.scope["config"] = config
     return await call_next(request)
 
 
 @app.get("/", include_in_schema=False)
-def root_redirect(request: Request):
+def root_redirect(request: Request) -> RedirectResponse:
     return RedirectResponse(request.url_for("ui_root"))
 
 
 @app.get("/login", include_in_schema=False, dependencies=[Depends(redirect_if_authenticated)])
-async def login_get(request: Request):
+async def login_get(request: Request) -> Response:
     return templates.TemplateResponse(request, "login.html.jinja")
 
 
 @app.post("/login", tags=["login"])
-async def login_post(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
+async def login_post(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]) -> dict[str, str]:
     return {"access_token": await login_user(form_data.username, form_data.password), "token_type": "bearer"}
 
 
 @app.get("/setup", include_in_schema=False)
-async def setup_get(request: Request):
+async def setup_get(request: Request) -> Response:
     return templates.TemplateResponse(request, "setup.html.jinja")
 
 
 @app.post("/setup", include_in_schema=False)
-async def setup_post(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
+async def setup_post(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]) -> dict[str, str]:
     await create_initial_user(form_data.username, PWD_CXT.hash(form_data.password))
     return {"access_token": await login_user(form_data.username, form_data.password), "token_type": "bearer"}
 
 
 @app.get("/logout", include_in_schema=False)
-async def logout(request: Request):
+async def logout(request: Request) -> RedirectResponse:
     resp = RedirectResponse(request.url_for("login_get"), status_code=302)
     resp.delete_cookie(key="session_id")
     return resp
 
 
 @app.get("/docs", include_in_schema=False)
-async def swagger_docs(request: Request):
+async def swagger_docs(request: Request) -> Response:
     return get_swagger_ui_html(
         title="gooseBit docs",
         openapi_url="/openapi.json",

goosebit/api/routes.py

@@ -5,5 +5,5 @@
 from . import telemetry, v1
 
 router = APIRouter(prefix="/api", dependencies=[Depends(validate_current_user)])
-router.include_router(telemetry.router)
-router.include_router(v1.router)
+router.include_router(telemetry.router)  # type: ignore[attr-defined]
+router.include_router(v1.router)  # type: ignore[attr-defined]

goosebit/api/telemetry/metrics.py

@@ -9,7 +9,7 @@
 readers = []
 
 if config.metrics.prometheus.enable:
-    readers.append(prometheus.reader)
+    readers.append(prometheus.reader)  # type: ignore[attr-defined]
 
 
 resource = Resource(attributes={SERVICE_NAME: "goosebit"})

goosebit/api/telemetry/routes.py

@@ -6,4 +6,4 @@
 
 router = APIRouter(prefix="/telemetry")
 if config.metrics.prometheus.enable:
-    router.include_router(prometheus.router)
+    router.include_router(prometheus.router)  # type: ignore[attr-defined]

goosebit/api/v1/devices/device/routes.py

@@ -6,7 +6,7 @@
 from goosebit.api.v1.devices.device.responses import DeviceLogResponse
 from goosebit.auth import validate_user_permissions
 from goosebit.auth.permissions import GOOSEBIT_PERMISSIONS
-from goosebit.db import Device
+from goosebit.db import Device  # type: ignore[attr-defined]
 from goosebit.device_manager import get_device
 from goosebit.schema.devices import DeviceSchema
 

goosebit/api/v1/devices/routes.py

@@ -29,7 +29,7 @@ async def devices_get(_: Request) -> DevicesResponse:
     devices = await Device.all().prefetch_related("hardware", "assigned_software", "assigned_software__compatibility")
     response = DevicesResponse(devices=devices)
 
-    async def set_assigned_sw(d: DeviceSchema):
+    async def set_assigned_sw(d: DeviceSchema) -> DeviceSchema:
         device = await get_device(d.id)
         _, target = await DeviceManager.get_update(device)
         if target is not None:
@@ -108,4 +108,4 @@ async def devices_put(_: Request, config: DevicesPutRequest) -> StatusResponse:
     return StatusResponse(success=True)
 
 
-router.include_router(device.router)
+router.include_router(device.router)  # type: ignore[attr-defined]

goosebit/api/v1/download/routes.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Response
 from fastapi.requests import Request
 from fastapi.responses import FileResponse, RedirectResponse, StreamingResponse
 
@@ -9,7 +9,7 @@
 
 
 @router.get("/{file_id}")
-async def download_file(_: Request, file_id: int):
+async def download_file(_: Request, file_id: int) -> Response:
     software = await Software.get_or_none(id=file_id)
     if software is None:
         raise HTTPException(404)

goosebit/api/v1/routes.py

@@ -3,8 +3,8 @@
 from . import devices, download, rollouts, settings, software
 
 router = APIRouter(prefix="/v1")
-router.include_router(software.router)
-router.include_router(devices.router)
-router.include_router(rollouts.router)
-router.include_router(download.router)
-router.include_router(settings.router)
+router.include_router(software.router)  # type: ignore[attr-defined]
+router.include_router(devices.router)  # type: ignore[attr-defined]
+router.include_router(rollouts.router)  # type: ignore[attr-defined]
+router.include_router(download.router)  # type: ignore[attr-defined]
+router.include_router(settings.router)  # type: ignore[attr-defined]

goosebit/api/v1/settings/routes.py

@@ -6,7 +6,7 @@
 
 router = APIRouter(prefix="/settings", tags=["settings"])
 
-router.include_router(users.router)
+router.include_router(users.router)  # type: ignore[attr-defined]
 
 
 @router.get("/permissions", response_model_exclude_none=True)

goosebit/api/v1/software/routes.py

@@ -3,14 +3,15 @@
 import random
 import string
 
-from anyio import Path, open_file
+from anyio import open_file
 from fastapi import APIRouter, File, Form, HTTPException, Security, UploadFile
 from fastapi.requests import Request
 
 from goosebit.api.responses import StatusResponse
 from goosebit.auth import validate_user_permissions
 from goosebit.auth.permissions import GOOSEBIT_PERMISSIONS
 from goosebit.db.models import Rollout, Software
+from goosebit.schema.software import SoftwareSchema
 from goosebit.storage import storage
 from goosebit.updates import create_software_update
 from goosebit.util.path import validate_filename
@@ -27,7 +28,7 @@
 )
 async def software_get(_: Request) -> SoftwareResponse:
     software = await Software.all().prefetch_related("compatibility")
-    return SoftwareResponse(software=software)
+    return SoftwareResponse(software=[SoftwareSchema.model_validate(s) for s in software])
 
 
 @router.delete(
@@ -61,7 +62,7 @@ async def software_delete(_: Request, delete_req: SoftwareDeleteRequest) -> Stat
     "",
     dependencies=[Security(validate_user_permissions, scopes=[GOOSEBIT_PERMISSIONS["software"]["write"]()])],
 )
-async def post_update(_: Request, file: UploadFile | None = File(None), url: str | None = Form(None)):
+async def post_update(_: Request, file: UploadFile | None = File(None), url: str | None = Form(None)) -> dict[str, int]:
     if url is not None:
         # remote file
         software = await Software.get_or_none(uri=url)
@@ -75,9 +76,9 @@ async def post_update(_: Request, file: UploadFile | None = File(None), url: str
         software = await create_software_update(url, None)
     elif file is not None:
         # local file
-        temp_dir = Path(storage.get_temp_dir())
+        temp_dir = await storage.get_temp_dir()
         try:
-            file_path = await validate_filename(file.filename, temp_dir)
+            file_path = await validate_filename(file.filename or "unknown", temp_dir)
         except ValueError as e:
             raise HTTPException(400, f"Invalid filename: {e}")
         tmp_file_path = temp_dir.joinpath("".join(random.choices(string.ascii_lowercase, k=12)) + ".tmp")