replaces print statements with structured logs

aim/cli/digifeeds.py

@@ -6,6 +6,7 @@
 from typing_extensions import Annotated
 from aim.digifeeds.database import models, main
 from aim.digifeeds import functions
+from aim.services import S
 
 import json
 import sys
@@ -31,14 +32,28 @@ def add_to_digifeeds_set(
     Args:
         barcode (str): Barcode of item
     """
-    print(f'Adding barcode "{barcode}" to database')
+    S.logger.info(
+        "add_to_digifeeds_set_start",
+        message="Start adding item to digifeeds set",
+        barcode=barcode,
+    )
     item = functions.add_to_digifeeds_set(barcode)
     if item.has_status("not_found_in_alma"):
-        print("Item not found in alma.")
+        S.logger.info(
+            "not_found_in_alma", message="Item not found in alma.", barcode=barcode
+        )
     if item.has_status("added_to_digifeeds_set"):
-        print("Item added to digifeeds set")
+        S.logger.info(
+            "added_to_digifeeds_set",
+            message="Item added to digifeeds set",
+            barcode=barcode,
+        )
     else:
-        print("Item NOT added to digifeeds set")
+        S.logger.error(
+            "not_added_to_digifeeds_set",
+            message="Item NOT added to digifeeds set",
+            barcode=barcode,
+        )
 
 
 @app.command()
@@ -57,9 +72,9 @@ def check_zephir(
     print(f"Checking Zephir for {barcode}")
     item = functions.check_zephir(barcode)
     if item:
-        print(f"{barcode} is in Zephir")
+        S.logger.info("in_zephir", message="Item is in zephir", barcode=barcode)
     else:
-        print(f"{barcode} is NOT in Zephir")
+        S.logger.info("not_in_zephir", message="Item is NOT in zephir", barcode=barcode)
 
 
 @app.command()
@@ -91,9 +106,21 @@ def move_to_pickup(
     pickup from google. When it's finished, the volume is moved to the processed
     folder in the bucket and prefixed with the date and time.
     """
-    print(f'Moving barcode "{barcode}" from the s3 bucket to the google drive')
+    S.logger.info(
+        "move_to_pickup_start",
+        message="Start moving item from s3 bucket to pickup google drive",
+        barcode=barcode,
+    )
     item = functions.move_to_pickup(barcode)
     if item is None:
-        print("Item has not been in zephir long enough")
+        S.logger.info(
+            "not_in_zephir_long_enough",
+            message="Item has not been in zephir long enough",
+            barcode=barcode,
+        )
     else:
-        print("Item has been successfully moved to pickup")
+        S.logger.info(
+            "move_to_pickup_success",
+            message="Item has been successfully moved to pickup",
+            barcode=barcode,
+        )

aim/services.py

@@ -1,13 +1,45 @@
 from typing import NamedTuple
 import os
 import sqlalchemy as sa
+import structlog
+import sys
+
+# Configuring the Logger
+shared_processors = [
+    # Processors that have nothing to do with output,
+    # e.g., add timestamps or log level names.
+    structlog.processors.add_log_level,
+]
+
+if sys.stderr.isatty():
+    # Pretty printing when we run in a terminal session.
+    # Automatically prints pretty tracebacks when "rich" is installed
+
+    processors = shared_processors + [
+        structlog.processors.TimeStamper(fmt="iso"),
+        structlog.dev.ConsoleRenderer(),
+    ]
+
+else:
+    # Print JSON when we run, e.g., in a Docker container.
+    # Also print structured tracebacks.
+
+    processors = shared_processors + [
+        structlog.processors.dict_tracebacks,
+        structlog.processors.JSONRenderer(),
+    ]
+
+structlog.configure(processors)
 
 
 class Services(NamedTuple):
     """
     Global Configuration Services
     """
 
+    #: The structured logger
+    logger: structlog.stdlib.BoundLogger
+
     #: The Digifeeds MySQL database
     mysql_database: sa.engine.URL
 
@@ -54,6 +86,7 @@ class Services(NamedTuple):
 
 
 S = Services(
+    logger=structlog.get_logger(),
     mysql_database=sa.engine.URL.create(
         drivername="mysql+mysqldb",
         username=os.environ["MARIADB_USER"],

pyproject.toml

@@ -20,6 +20,7 @@ alembic = "^1.13.2"
 typer = "^0.12.5"
 boto3 = "^1.35.28"
 rclone-python = "^0.1.12"
+structlog = "^24.4.0"
 
 
 [tool.poetry.group.dev.dependencies]

tests/cli/test_digifeeds.py

@@ -54,8 +54,8 @@ def test_add_to_db_where_item_is_not_in_digifeeds_set(item_data):
     assert add_to_digifeeds_set_stub.call_count == 1
     assert add_item_status.call_count == 1
     assert result.exit_code == 0
-    assert 'Adding barcode "some_barcode" to database' in result.stdout
-    assert "Item added to digifeeds set" in result.stdout
+    assert "add_to_digifeeds_set_start" in result.stdout
+    assert "added_to_digifeeds_set" in result.stdout
 
 
 def test_add_to_db_where_item_is_not_in_alma(item_data, mocker):
@@ -64,8 +64,8 @@ def test_add_to_db_where_item_is_not_in_alma(item_data, mocker):
     mocker.patch.object(functions, "add_to_digifeeds_set", return_value=item)
 
     result = runner.invoke(app, ["digifeeds", "add-to-digifeeds-set", "some_barcode"])
-    assert "Item not found in alma" in result.stdout
-    assert "Item NOT added to digifeeds set" in result.stdout
+    assert "not_found_in_alma" in result.stdout
+    assert "not_added_to_digifeeds_set" in result.stdout
 
 
 def test_load_statuses(mocker):
@@ -101,7 +101,7 @@ def test_check_zephir_for_item_when_item_is_in_zephir(item_data):
     assert get_item.call_count == 1
     assert add_item_status.call_count == 1
     assert result.exit_code == 0
-    assert "some_barcode is in Zephir" in result.stdout
+    assert "in_zephir" in result.stdout
 
 
 @responses.activate
@@ -114,7 +114,7 @@ def test_check_zephir_for_item_when_item_is_not_in_zephir(item_data):
     assert get_item.call_count == 1
     assert add_item_status.call_count == 0
     assert result.exit_code == 0
-    assert "some_barcode is NOT in Zephir" in result.stdout
+    assert "not_in_zephir" in result.stdout
 
 
 def test_move_to_pickup_success(mocker, item_data):
@@ -126,7 +126,7 @@ def test_move_to_pickup_success(mocker, item_data):
     result = runner.invoke(app, ["digifeeds", "move-to-pickup", "some_barcode"])
 
     move_volume_to_pickup_mock.assert_called_once()
-    assert "Item has been successfully moved to pickup" in result.stdout
+    assert "move_to_pickup_success" in result.stdout
     assert result.exit_code == 0
 
 
@@ -138,5 +138,5 @@ def test_move_to_pickup_where_not_in_zephir(mocker):
     result = runner.invoke(app, ["digifeeds", "move-to-pickup", "some_barcode"])
 
     move_volume_to_pickup_mock.assert_called_once()
-    assert "Item has not been in zephir long enough" in result.stdout
+    assert "not_in_zephir_long_enough" in result.stdout
     assert result.exit_code == 0