Add SUPPRESS_ACCESS_LOGS setting and logging configuration tests

.gitignore

@@ -58,3 +58,4 @@ test_fq.py
 .idea/*
 venv/*
 .vscode/
+.codex

.qlty/.gitignore

@@ -0,0 +1,7 @@
+*
+!configs
+!configs/**
+!hooks
+!hooks/**
+!qlty.toml
+!.gitignore

.qlty/configs/.hadolint.yaml

@@ -0,0 +1,2 @@
+ignored:
+  - DL3008

.qlty/qlty.toml

@@ -0,0 +1,95 @@
+# This file was automatically generated by `qlty init`.
+# You can modify it to suit your needs.
+# We recommend you to commit this file to your repository.
+#
+# This configuration is used by both Qlty CLI and Qlty Cloud.
+#
+#     Qlty CLI -- Code quality toolkit for developers
+#     Qlty Cloud -- Fully automated Code Health Platform
+#
+# Try Qlty Cloud: https://qlty.sh
+#
+# For a guide to configuration, visit https://qlty.sh/d/config
+# Or for a full reference, visit https://qlty.sh/d/qlty-toml
+config_version = "0"
+
+exclude_patterns = [
+  "*_min.*",
+  "*-min.*",
+  "*.min.*",
+  "**/.yarn/**",
+  "**/*.d.ts",
+  "**/assets/**",
+  "**/bower_components/**",
+  "**/build/**",
+  "**/cache/**",
+  "**/config/**",
+  "**/db/**",
+  "**/deps/**",
+  "**/dist/**",
+  "**/extern/**",
+  "**/external/**",
+  "**/generated/**",
+  "**/Godeps/**",
+  "**/gradlew/**",
+  "**/mvnw/**",
+  "**/node_modules/**",
+  "**/protos/**",
+  "**/seed/**",
+  "**/target/**",
+  "**/templates/**",
+  "**/testdata/**",
+  "**/vendor/**",
+]
+
+test_patterns = [
+  "**/test/**",
+  "**/spec/**",
+  "**/*.test.*",
+  "**/*.spec.*",
+  "**/*_test.*",
+  "**/*_spec.*",
+  "**/test_*.*",
+  "**/spec_*.*",
+]
+
+[smells]
+mode = "comment"
+
+[[source]]
+name = "default"
+default = true
+
+
+[[plugin]]
+name = "actionlint"
+
+[[plugin]]
+name = "bandit"
+
+[[plugin]]
+name = "hadolint"
+
+[[plugin]]
+name = "radarlint-iac"
+mode = "monitor"
+
+[[plugin]]
+name = "radarlint-python"
+mode = "comment"
+
+[[plugin]]
+name = "ripgrep"
+mode = "comment"
+
+[[plugin]]
+name = "ruff"
+drivers = [
+  "lint",
+]
+
+[[plugin]]
+name = "trufflehog"
+
+[[plugin]]
+name = "zizmor"

README.md

@@ -38,6 +38,7 @@ These application settings are validated at startup by `QueueServerSettings` wit
 | `DEFAULT_JOB_REQUEUE_LIMIT` | `-1` | Default retry limit. `-1` retries forever. |
 | `ENABLE_REQUEUE_SCRIPT` | `true` | Enables the background requeue loop. |
 | `LOG_LEVEL` | `INFO` | Application log level. |
+| `SUPPRESS_ACCESS_LOGS` | `true` | Suppresses all Uvicorn access logs. |
 | `REDIS_DB` | `0` | Redis database number. |
 | `REDIS_KEY_PREFIX` | `fq_server` | Prefix used for Redis keys. |
 | `REDIS_CONN_TYPE` | `tcp_sock` | Redis connection type: `tcp_sock` or `unix_sock`. |

asgi.py

@@ -1,26 +1,15 @@
 # Copyright (c) 2025 Flowdacity Team. See LICENSE.txt for details.
 # ASGI application entrypoint for Flowdacity Queue (FQ) Server
 
-import logging
-
 from fq_server import QueueServerSettings, setup_server
-
-
-def configure_logging(log_level: str) -> None:
-    level = getattr(logging, log_level)
-    root_logger = logging.getLogger()
-
-    if not root_logger.handlers:
-        logging.basicConfig(
-            level=level,
-            format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
-        )
-
-    logging.getLogger("fq_server").setLevel(level)
+from fq_server.logging import configure_logging
 
 
 settings = QueueServerSettings.from_env()
-configure_logging(settings.log_level)
+configure_logging(
+    settings.log_level,
+    suppress_access_logs=settings.suppress_access_logs,
+)
 server = setup_server(settings.to_fq_config())
 
 # ASGI app exposed for Uvicorn/Hypercorn

docs/configuration.rst

@@ -24,6 +24,9 @@ Queue settings
     Application log level. Supported values are ``DEBUG``, ``INFO``, ``WARNING``,
     ``ERROR``, and ``CRITICAL``.
 
+``SUPPRESS_ACCESS_LOGS``
+    Suppresses all Uvicorn access logs.
+
 Redis settings
 --------------
 
@@ -69,6 +72,8 @@ Defaults
      - ``true``
    * - ``LOG_LEVEL``
      - ``INFO``
+   * - ``SUPPRESS_ACCESS_LOGS``
+     - ``true``
    * - ``REDIS_DB``
      - ``0``
    * - ``REDIS_KEY_PREFIX``

fq_server/logging.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2025 Flowdacity Development Team. See LICENSE.txt for details.
+
+import logging
+
+
+def configure_logging(
+    log_level: str,
+    suppress_access_logs: bool = True,
+) -> None:
+    level = getattr(logging, log_level)
+    root_logger = logging.getLogger()
+
+    if not root_logger.handlers:
+        logging.basicConfig(
+            level=level,
+            format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
+        )
+
+    logging.getLogger("fq_server").setLevel(level)
+    logging.getLogger("uvicorn.access").disabled = suppress_access_logs

fq_server/settings.py

@@ -50,6 +50,10 @@ class QueueServerSettings(BaseSettings):
         default=True, validation_alias="ENABLE_REQUEUE_SCRIPT"
     )
     log_level: LogLevelName = Field(default="INFO", validation_alias="LOG_LEVEL")
+    suppress_access_logs: bool = Field(
+        default=True,
+        validation_alias="SUPPRESS_ACCESS_LOGS",
+    )
 
     redis_db: int = Field(default=0, ge=0, validation_alias="REDIS_DB")
     redis_key_prefix: str = Field(
@@ -74,7 +78,12 @@ class QueueServerSettings(BaseSettings):
         validation_alias="REDIS_UNIX_SOCKET_PATH",
     )
 
-    @field_validator("enable_requeue_script", "redis_clustered", mode="before")
+    @field_validator(
+        "enable_requeue_script",
+        "suppress_access_logs",
+        "redis_clustered",
+        mode="before",
+    )
     @classmethod
     def validate_boolean_env(cls, value: bool | str) -> bool:
         if isinstance(value, bool):

tests/test_config_settings.py

@@ -66,6 +66,14 @@ def test_queue_server_settings_log_level_override(self):
         settings = QueueServerSettings.from_env({"LOG_LEVEL": "debug"})
         self.assertEqual(settings.log_level, "DEBUG")
 
+    def test_queue_server_settings_suppress_access_logs_default(self):
+        settings = QueueServerSettings.from_env({})
+        self.assertTrue(settings.suppress_access_logs)
+
+    def test_queue_server_settings_suppress_access_logs_override(self):
+        settings = QueueServerSettings.from_env({"SUPPRESS_ACCESS_LOGS": "false"})
+        self.assertFalse(settings.suppress_access_logs)
+
     def test_queue_server_settings_rejects_invalid_log_level(self):
         with self.assertRaisesRegex(ValidationError, "LOG_LEVEL"):
-            QueueServerSettings.from_env({"LOG_LEVEL": "verbose"})
+            QueueServerSettings.from_env({"LOG_LEVEL": "verbose"})

tests/test_logging_config.py

@@ -0,0 +1,42 @@
+# Copyright (c) 2025 Flowdacity Development Team. See LICENSE.txt for details.
+
+import logging
+import unittest
+
+from fq_server.logging import configure_logging
+
+
+class TestLoggingConfig(unittest.TestCase):
+    def setUp(self):
+        self.root_logger = logging.getLogger()
+        self.original_root_handlers = list(self.root_logger.handlers)
+        self.original_root_level = self.root_logger.level
+        self.fq_logger = logging.getLogger("fq_server")
+        self.original_fq_level = self.fq_logger.level
+        self.access_logger = logging.getLogger("uvicorn.access")
+        self.original_disabled = self.access_logger.disabled
+
+    def tearDown(self):
+        for handler in list(self.root_logger.handlers):
+            if handler not in self.original_root_handlers:
+                self.root_logger.removeHandler(handler)
+                handler.close()
+
+        self.root_logger.setLevel(self.original_root_level)
+        self.fq_logger.setLevel(self.original_fq_level)
+        self.access_logger.disabled = self.original_disabled
+
+    def test_configure_logging_disables_access_logger_when_suppressed(self):
+        self.access_logger.disabled = False
+        configure_logging("INFO", suppress_access_logs=True)
+        self.assertTrue(self.access_logger.disabled)
+
+    def test_configure_logging_enables_access_logger_when_not_suppressed(self):
+        self.access_logger.disabled = True
+        configure_logging("INFO", suppress_access_logs=False)
+        self.assertFalse(self.access_logger.disabled)
+
+    def test_configure_logging_sets_fq_logger_level(self):
+        self.fq_logger.setLevel(logging.NOTSET)
+        configure_logging("WARNING", suppress_access_logs=False)
+        self.assertEqual(self.fq_logger.level, logging.WARNING)