chore: improve printing and logging (#26)

dsp_permissions_scripts/template.py

@@ -16,7 +16,7 @@
 
 def modify_oaps(oaps: list[Oap]) -> list[Oap]:
     for oap in oaps:
-        oap.scope.D.append(BuiltinGroup.PROJECT_MEMBER)
+        oap.scope.CR.append(BuiltinGroup.SYSTEM_ADMIN)
     return oaps
 
 

dsp_permissions_scripts/utils/doap_get.py

@@ -5,11 +5,14 @@
 
 from dsp_permissions_scripts.models.permission import Doap, DoapTarget, DoapTargetType
 from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.utils.get_logger import get_logger, get_timestamp
 from dsp_permissions_scripts.utils.project import get_project_iri_by_shortcode
 from dsp_permissions_scripts.utils.scope_serialization import (
     create_scope_from_admin_route_object,
 )
 
+logger = get_logger(__name__)
+
 
 def __filter_doaps_by_target(
     doaps: list[Doap],
@@ -79,6 +82,7 @@ def get_doaps_of_project(
     Optionally, select only the DOAPs that are related to either a group, or a resource class, or a property.
     By default, all DOAPs are returned, regardless of their target (target=all).
     """
+    logger.info(f"******* Getting DOAPs of project {shortcode} on server {host} *******")
     project_iri = get_project_iri_by_shortcode(
         shortcode=shortcode,
         host=host,
@@ -92,6 +96,7 @@ def get_doaps_of_project(
         doaps=doaps,
         target=target,
     )
+    logger.info(f"Found {len(doaps)} DOAPs, {len(filtered_doaps)} of which are related to {target}.")
     return filtered_doaps
 
 
@@ -104,7 +109,10 @@ def print_doaps_of_project(
     heading = f"Project {shortcode} on server {host} has {len(doaps)} DOAPs"
     if target != DoapTargetType.ALL:
         heading += f" which are related to a {target}"
-    print(f"\n{heading}\n{'=' * len(heading)}\n")
+    print(f"\n{get_timestamp()}: {heading}\n{'=' * (len(heading) + len(get_timestamp()) + 2)}\n")
+    logger.info(f"******* Printing DOAPs of project {shortcode} on server {host} *******")
+    logger.info(heading)
     for d in doaps:
-        print(d.model_dump_json(indent=2, exclude_none=True))
-        print()
+        representation = d.model_dump_json(indent=2, exclude_none=True)
+        print(representation + "\n")
+        logger.info(representation)

dsp_permissions_scripts/utils/doap_set.py

@@ -1,3 +1,4 @@
+from typing import Literal
 from urllib.parse import quote_plus
 
 import requests
@@ -6,10 +7,12 @@
 from dsp_permissions_scripts.models.scope import PermissionScope
 from dsp_permissions_scripts.utils.authentication import get_protocol
 from dsp_permissions_scripts.utils.doap_get import create_doap_from_admin_route_response
+from dsp_permissions_scripts.utils.get_logger import get_logger, get_timestamp
 from dsp_permissions_scripts.utils.scope_serialization import (
     create_admin_route_object_from_scope,
 )
 
+logger = get_logger(__name__)
 
 def __update_doap_scope(
     doap_iri: str,
@@ -31,6 +34,19 @@ def __update_doap_scope(
     return new_doap
 
 
+def __log_and_print_doap_update(
+    doap: Doap,
+    state: Literal["before", "after"],
+) -> None:
+    """
+    Logs and prints the DOAP before or after the update.
+    """
+    heading = f"DOAP {state}:"
+    body = doap.model_dump_json(indent=2)
+    print(f"{heading}\n{'-' * len(heading)}\n{body}\n")
+    logger.info(f"{heading}\n{body}")
+
+
 def apply_updated_doaps_on_server(
     doaps: list[Doap],
     host: str,
@@ -44,18 +60,16 @@ def apply_updated_doaps_on_server(
         host: the DSP server where the project is located
         token: the access token
     """
-    heading = f"Update {len(doaps)} DOAPs on {host}..."
+    logger.info(f"******* Updating {len(doaps)} DOAPs on {host} *******")
+    heading = f"{get_timestamp()}: Updating {len(doaps)} DOAPs on {host}..."
     print(f"\n{heading}\n{'=' * len(heading)}\n")
     for d in doaps:
-        print("Old DOAP:\n=========")
-        print(d.model_dump_json(indent=2))
+        __log_and_print_doap_update(doap=d, state="before")
         new_doap = __update_doap_scope(
             doap_iri=d.doap_iri,
             scope=d.scope,
             host=host,
             token=token,
         )
-        print("\nNew DOAP:\n=========")
-        print(new_doap.model_dump_json(indent=2))
-        print()
-    print("All DOAPs have been updated.")
+        __log_and_print_doap_update(doap=new_doap, state="after")
+    print(f"{get_timestamp()}: All DOAPs have been updated.")

dsp_permissions_scripts/utils/get_logger.py

@@ -0,0 +1,33 @@
+import logging
+from datetime import datetime
+
+
+def get_logger(name: str) -> logging.Logger:
+    """
+    Create a logger instance,
+    set its level to INFO,
+    and configure it to write to a file in the user's home directory.
+
+    Args:
+        name: name of the logger
+        filesize_mb: maximum size per log file in MB, defaults to 5
+        backupcount: number of log files to keep, defaults to 4
+
+    Returns:
+        the logger instance
+    """
+    _logger = logging.getLogger(name)
+    _logger.setLevel(logging.INFO)
+    formatter = logging.Formatter(fmt="{asctime} {filename: <25} {levelname: <8} {message}", style="{")
+    formatter.default_time_format = "%Y-%m-%d %H:%M:%S"
+    handler = logging.FileHandler(
+        filename="logging.log",
+        mode="a",
+    )
+    handler.setFormatter(formatter)
+    _logger.addHandler(handler)
+    return _logger
+
+
+def get_timestamp() -> str:
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

dsp_permissions_scripts/utils/oap.py

@@ -8,22 +8,31 @@
 from dsp_permissions_scripts.models.scope import PermissionScope
 from dsp_permissions_scripts.models.value import ValueUpdate
 from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.utils.get_logger import get_logger, get_timestamp
 from dsp_permissions_scripts.utils.scope_serialization import create_string_from_scope
 
+logger = get_logger(__name__)
 
 def apply_updated_oaps_on_server(
     resource_oaps: list[Oap],
     host: str,
     token: str,
 ) -> None:
     """Applies object access permissions on a DSP server."""
-    for resource_oap in resource_oaps:
+    logger.info("******* Applying updated object access permissions on server *******")
+    print(f"{get_timestamp()}: ******* Applying updated object access permissions on server *******")
+    for index, resource_oap in enumerate(resource_oaps):
+        msg = f"Updating permissions of resource {index + 1}/{len(resource_oaps)}: {resource_oap.object_iri}..."
+        logger.info("=====")
+        logger.info(msg)
+        print(f"{get_timestamp()}: {msg}")
         update_permissions_for_resources_and_values(
             resource_iris=[resource_oap.object_iri],
             scope=resource_oap.scope,
             host=host,
             token=token,
         )
+        logger.info(f"Updated permissions of resource {resource_oap.object_iri} and its values.")
 
 
 def update_permissions_for_resources_and_values(
@@ -48,7 +57,6 @@ def __update_permissions_for_resource_and_values(
     """
     Updates the permissions for the given resource and its values.
     """
-    print(f"Updating permissions for {resource_iri}...")
     resource = __get_resource(resource_iri, host, token)
     lmd = __get_lmd(resource)
     type_ = __get_type(resource)
@@ -57,7 +65,6 @@ def __update_permissions_for_resource_and_values(
     update_permissions_for_resource(resource_iri, lmd, type_, context, scope, host, token)
     for v in values:
         __update_permissions_for_value(resource_iri, v, type_, context, scope, host, token)
-    print("Done. \n")
 
 
 def update_permissions_for_resource(
@@ -85,7 +92,7 @@ def update_permissions_for_resource(
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.put(url, headers=headers, json=payload, timeout=5)
     assert response.status_code == 200
-    print(f"Updated permissions for {resource_iri}")
+    logger.info(f"Updated permissions of resource {resource_iri}")
 
 
 def __update_permissions_for_value(
@@ -100,7 +107,6 @@ def __update_permissions_for_value(
     """
     Updates the permissions for the given value.
     """
-    print(value.value_iri)
     payload = {
         "@id": resource_iri,
         "@type": resource_type,
@@ -116,22 +122,19 @@ def __update_permissions_for_value(
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.put(url, headers=headers, json=payload, timeout=5)
     if response.status_code == 400 and response.text:
-        if (
-            "dsp.errors.BadRequestException: "
-            "The submitted permissions are the same as the current ones" in response.text
-        ):
-            print(f"Permissions for {value.value_iri} are already up to date")
-            return
-    if response.status_code != 200:
-        print(response.status_code)
-        print(response.text)
-        print(resource_iri, value.value_iri)
-        print(json.dumps(payload, indent=4))
-        print("!!!!!")
-        print()
-        return
-        # raise Exception(f"Error updating permissions for {value.value_iri}")
-    print(f"Updated permissions for {value.value_iri}")
+        already = "dsp.errors.BadRequestException: The submitted permissions are the same as the current ones"
+        if already in response.text:
+            msg = f"Permissions of resource {resource_iri}, value {value.value_iri} are already up to date"
+            logger.warning(msg)
+    elif response.status_code != 200:
+        logger.error(
+            f"Error while updating permissions of resource {resource_iri}, value {value.value_iri}. "
+            f"Response status code: {response.status_code}. "
+            f"Response text: {response.text}. "
+            f"Payload: {json.dumps(payload, indent=4)}"
+        )
+    else:
+        logger.info(f"Updated permissions of resource {resource_iri}, value {value.value_iri}")
 
 
 def __get_value_iris(resource: dict[str, Any]) -> list[ValueUpdate]:

dsp_permissions_scripts/utils/project.py

@@ -4,8 +4,11 @@
 
 from dsp_permissions_scripts.models.permission import Oap
 from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.utils.get_logger import get_logger, get_timestamp
 from dsp_permissions_scripts.utils.scope_serialization import create_scope_from_string
 
+logger = get_logger(__name__)
+
 
 def get_project_iri_by_shortcode(shortcode: str, host: str) -> str:
     """
@@ -24,6 +27,8 @@ def get_all_resource_oaps_of_project(
     host: str,
     token: str,
 ) -> list[Oap]:
+    logger.info(f"******* Getting all resource OAPs of project {shortcode} *******")
+    print(f"{get_timestamp()}: ******* Getting all resource OAPs of project {shortcode} *******")
     project_iri = get_project_iri_by_shortcode(
         shortcode=shortcode,
         host=host,
@@ -42,6 +47,8 @@ def get_all_resource_oaps_of_project(
             token=token,
         )
         all_resource_oaps.extend(resource_oaps)
+    logger.info(f"Retrieved a TOTAL of {len(all_resource_oaps)} resource OAPs of project {shortcode}.")
+    print(f"{get_timestamp()}: Retrieved a TOTAL of {len(all_resource_oaps)} resource OAPs of project {shortcode}.")
     return all_resource_oaps
 
 
@@ -50,6 +57,7 @@ def __get_all_resource_class_iris_of_project(
     host: str,
     token: str,
 ) -> list[str]:
+    logger.info(f"Getting all resource class IRIs of project {project_iri}...")
     project_onto_iris = __get_onto_iris_of_project(
         project_iri=project_iri,
         host=host,
@@ -63,6 +71,7 @@ def __get_all_resource_class_iris_of_project(
             token=token,
         )
         all_class_iris.extend(class_iris)
+        logger.info(f"Found {len(class_iris)} resource classes in onto {onto_iri}.")
     return all_class_iris
 
 
@@ -109,12 +118,15 @@ def __get_all_resource_oaps_of_resclass(
     project_iri: str,
     token: str,
 ) -> list[Oap]:
+    print(f"{get_timestamp()}: Getting all resource OAPs of class {resclass_iri}...")
+    logger.info(f"Getting all resource OAPs of class {resclass_iri}...")
     protocol = get_protocol(host)
     headers = {"X-Knora-Accept-Project": project_iri, "Authorization": f"Bearer {token}"}
     resources: list[Oap] = []
     page = 0
     more = True
     while more:
+        logger.info(f"Getting page {page}...")
         more, iris = __get_next_page(
             protocol=protocol,
             host=host,
@@ -124,6 +136,8 @@ def __get_all_resource_oaps_of_resclass(
         )
         resources.extend(iris)
         page += 1
+    print(f"{get_timestamp()}: Retrieved {len(resources)} resource OAPs of class {resclass_iri}.")
+    logger.info(f"Retrieved {len(resources)} resource OAPs of class {resclass_iri}.")
     return resources
 
 

pyproject.toml

@@ -37,6 +37,7 @@ disable = [
     "invalid-name",
     "trailing-whitespace",
     "too-few-public-methods",
+    "logging-fstring-interpolation",
 ]
 
 [tool.black]