chore(ci): configure ruff

.pre-commit-config.yaml

@@ -1,29 +1,12 @@
 repos:
-  - repo: https://github.com/asottile/pyupgrade
-    rev: v2.31.1
-    hooks:
-      - id: pyupgrade
-        args:
-          - --py311-plus
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.8.0'
+    rev: 'v0.8.4'
     hooks:
       - id: ruff
         args: [--fix]
-  - repo: local
-    hooks:
-      - id: isort
-        name: isort
-        language: system
-        entry: isort
-        types: [python]
-      - id: black
-        name: black
-        language: system
-        entry: black
-        types: [python]
+      - id: ruff-format
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v3.4.0
+    rev: v4.5.0
     hooks:
       - id: trailing-whitespace
         exclude: ^src/api/client.js$

catalog-info.yaml

@@ -3,7 +3,7 @@ apiVersion: backstage.io/v1alpha1
 kind: Component
 metadata:
   name: nowplaying
-  title: Nowplaging Service
+  title: Nowplaying Service
   description: |
     The nowplaying daemon is the python server central to our songticker.
   annotations:

docs/gen_ref_pages.py

@@ -16,20 +16,23 @@
 
     parts = list(module_path.parts)
 
-    if parts[-1] == "__init__":
-        continue
-    elif parts[-1] == "__main__":
+    if parts[-1] in ["__init__", "__main__"]:
         continue
 
     with mkdocs_gen_files.open(full_doc_path, "w") as fd:
         identifier = ".".join(parts)
         print("::: " + identifier, file=fd)
 
-    mkdocs_gen_files.set_edit_path(full_doc_path, path)  #
+    mkdocs_gen_files.set_edit_path(full_doc_path, path)
 
 with mkdocs_gen_files.open("reference/SUMMARY.md", "w") as nav_file:
     nav_file.writelines(nav.build_literate_nav())
 
-readme = Path("README.md").open("r")
-with mkdocs_gen_files.open("index.md", "w") as index_file:
+with (
+    Path("README.md").open("r") as readme,
+    mkdocs_gen_files.open(
+        "index.md",
+        "w",
+    ) as index_file,
+):
     index_file.writelines(readme.read())

nowplaying/__main__.py

@@ -9,7 +9,8 @@
 from .main import NowPlaying
 
 
-def main():
+def main() -> None:
+    """Run nowplaying."""
     NowPlaying().run()
 
 

nowplaying/api.py

@@ -1,15 +1,27 @@
+"""Nowplaying ApiServer."""
+
+from __future__ import annotations
+
 import json
 import logging
-from queue import Queue
+from typing import TYPE_CHECKING, Self
 
-import cherrypy
+import cherrypy  # type: ignore[import-untyped]
 import cridlib
 from cloudevents.exceptions import GenericException as CloudEventException
 from cloudevents.http import from_http
 from werkzeug.exceptions import BadRequest, HTTPException, UnsupportedMediaType
 from werkzeug.routing import Map, Rule
 from werkzeug.wrappers import Request, Response
 
+if TYPE_CHECKING:  # pragma: no cover
+    from collections.abc import Iterable
+    from queue import Queue
+    from wsgiref.types import StartResponse, WSGIEnvironment
+
+    from nowplaying.options import Options
+
+
 logger = logging.getLogger(__name__)
 
 _RABE_CLOUD_EVENTS_SUBS = (
@@ -25,21 +37,27 @@
 class ApiServer:
     """The API server."""
 
-    def __init__(self, options, event_queue: Queue, realm: str = "nowplaying"):
+    def __init__(
+        self: Self,
+        options: Options,
+        event_queue: Queue,
+        realm: str = "nowplaying",
+    ) -> None:
+        """Create ApiServer."""
         self.options = options
         self.event_queue = event_queue
         self.realm = realm
 
         self.url_map = Map([Rule("/webhook", endpoint="webhook")])
 
-    def run_server(self):
+    def run_server(self: Self) -> None:
         """Run the API server."""
         if self.options.debug:
             from werkzeug.serving import run_simple
 
-            self._server = run_simple(
-                self.options.apiBindAddress,
-                self.options.apiPort,
+            run_simple(
+                self.options.api_bind_address,
+                self.options.api_port,
                 self,
                 use_debugger=True,
                 use_reloader=True,
@@ -48,25 +66,35 @@ def run_server(self):
             cherrypy.tree.graft(self, "/")
             cherrypy.server.unsubscribe()
 
-            self._server = cherrypy._cpserver.Server()
+            self._server = cherrypy._cpserver.Server()  # noqa: SLF001
 
-            self._server.socket_host = self.options.apiBindAddress
-            self._server.socket_port = self.options.apiPort
+            self._server.socket_host = self.options.api_bind_address
+            self._server.socket_port = self.options.api_port
 
             self._server.subscribe()
 
             cherrypy.engine.start()
             cherrypy.engine.block()
 
-    def stop_server(self):
+    def stop_server(self: Self) -> None:
         """Stop the server."""
         self._server.stop()
         cherrypy.engine.exit()
 
-    def __call__(self, environ, start_response):
+    def __call__(
+        self: Self,
+        environ: WSGIEnvironment,
+        start_response: StartResponse,
+    ) -> Iterable[bytes]:
+        """Forward calls to wsgi_app."""
         return self.wsgi_app(environ, start_response)
 
-    def wsgi_app(self, environ, start_response):
+    def wsgi_app(
+        self: Self,
+        environ: WSGIEnvironment,
+        start_response: StartResponse,
+    ) -> Iterable[bytes]:
+        """Return a wsgi app."""
         request = Request(environ)
         auth = request.authorization
         if auth and self.check_auth(auth.username, auth.password):
@@ -75,21 +103,27 @@ def wsgi_app(self, environ, start_response):
             response = self.auth_required(request)
         return response(environ, start_response)
 
-    def check_auth(self, username, password):
-        return (
-            username in self.options.apiAuthUsers
-            and self.options.apiAuthUsers[username] == password
+    def check_auth(self: Self, username: str | None, password: str | None) -> bool:
+        """Check if auth is valid."""
+        return str(
+            username,
+        ) in self.options.api_auth_users and self.options.api_auth_users[
+            str(username)
+        ] == str(
+            password,
         )
 
-    def auth_required(self, request):
+    def auth_required(self: Self, _: Request) -> Response:
+        """Check if auth is required."""
         return Response(
             "Could not verify your access level for that URL.\n"
             "You have to login with proper credentials",
             401,
             {"WWW-Authenticate": f'Basic realm="{self.realm}"'},
         )
 
-    def dispatch_request(self, request):
+    def dispatch_request(self: Self, request: Request) -> Response | HTTPException:
+        """Dispatch requests to handlers."""
         adapter = self.url_map.bind_to_environ(request.environ)
         try:
             endpoint, values = adapter.match()
@@ -101,25 +135,25 @@ def dispatch_request(self, request):
                 {"Content-Type": "application/json"},
             )
 
-    def on_webhook(self, request):
+    def on_webhook(self: Self, request: Request) -> Response:
         """Receive a CloudEvent and put it into the event queue."""
         logger.warning("Received a webhook")
         if (
             request.headers.get("Content-Type")
             not in _RABE_CLOUD_EVENTS_SUPPORTED_MEDIA_TYPES
         ):
-            raise UnsupportedMediaType()
+            raise UnsupportedMediaType
         try:
             event = from_http(request.headers, request.data)
         except CloudEventException as error:
-            raise BadRequest(description=f"{error}")
+            raise BadRequest(description=str(error)) from error
 
         try:
             crid = cridlib.parse(event["id"])
             logger.debug("Detected CRID: %s", crid)
         except cridlib.CRIDError as error:
             raise BadRequest(
-                description=f"CRID '{event['id']}' is not a RaBe CRID"
+                description=f"CRID '{event['id']}' is not a RaBe CRID",
             ) from error
 
         logger.info("Received event: %s", event)