✨ use toml to parse the recent_run file

pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
     "httpx>=0.28.1",
     "ms-active-directory>=1.14.1",
     "pyyaml>=6.0.2",
+    "tomli-w>=1.2.0",
     "validators>=0.34.0",
 ]
 

src/adgroupsync/__main__.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from ldap3 import SIMPLE
 from ms_active_directory import ADDomain, ADGroup, ADUser, ADObject
 
+from .exceptions import ScriptAlreadyRunningError
 from .conf import (
     AD_DOMAIN,
     AD_USER_NAME,
@@ -265,6 +266,7 @@ def main():
 
         # Get the recent run timestamp
         file_path = Path().home() / '.recent_run'
+
         with RecentRun(file_path, tz=AD_TIMEZONE) as recent_run:
             if recent_run.datetime is None:
                 logger.info('No recent run found')
@@ -280,6 +282,11 @@ def main():
             started_at = recent_run.started_at.strftime('%Y-%m-%d %H:%M:%S %Z')
             logger.info(f"Setting previous run to: {started_at}")
 
+    # Exit if the script is already running
+    except ScriptAlreadyRunningError:
+        logger.info('Script is already running. Exiting...')
+        exit(0)
+
     except Exception as e:
         logger.error(f"An error occurred: {e}", exc_info=True)
         exit(1)

src/adgroupsync/exceptions.py

@@ -0,0 +1,5 @@
+class ScriptAlreadyRunningError(Exception):
+    """
+    A custom exception to raise when the script is already running
+    """
+    pass

src/adgroupsync/models.py

@@ -1,15 +1,19 @@
+import datetime
 import json
 import logging
+import tomllib
 from collections import deque
 from datetime import datetime as dt, timezone
 from pathlib import Path
 
 import pytz
+import tomli_w
 from civifang import api
 from httpx import post
 from ms_active_directory import ADUser, ADGroup
 
 from .enums import Priority
+from .exceptions import ScriptAlreadyRunningError
 
 logger = logging.getLogger(__package__)
 
@@ -28,7 +32,8 @@ class RecentRun:
         self._datetime = None
         self._timezone = tz
         self._file_path = file_path
-        self._is_running = False
+        self._is_running = None
+        self._already_running = None
         self._started_at = None
 
         # Create the file if it does not exist
@@ -36,6 +41,10 @@ class RecentRun:
 
         self._read_data_from_file()
 
+        # If the script was already running, throw an exception
+        if self._already_running:
+            raise ScriptAlreadyRunningError('The script is already running.')
+
     def _sync_file(
             self,
             recent_run: dt | None = None,
@@ -47,69 +56,38 @@ class RecentRun:
         :param is_running:
         :return:
         """
-        # Convert the is_running boolean to a string
+        rr = recent_run if recent_run else self._datetime
-        is_running = 'true' if is_running else 'false' \
+        is_running = is_running if is_running is not None else self._already_running
-            if is_running is not None else None
+        new_data = {
+            'recent-run': rr,
+            'is-running': is_running,
+        }
 
-        # Read the file and update the values if they are different
+        # Write the data to the file
-        with open(self._file_path, 'r+') as f:
+        with open(self._file_path, 'wb') as f:
-            # Read the data from the file
+            tomli_w.dump(new_data, f)
-            data = f.readlines()
-            old_recent_run, old_is_running = self._read_data(data)
-
-            # Update the values if they were provided
-            timestamp = recent_run.timestamp() if recent_run else old_recent_run
-            is_running = is_running or old_is_running
-            new_data = [
-                f"recent-run:{timestamp}",
-                '\n',
-                f"is-running:{is_running}",
-            ]
-
-            # Write the new data to the file
-            f.seek(0)
-            f.truncate()
-            f.writelines(new_data)
-
-    @staticmethod
-    def _read_data(data: list):
-        """
-        Read data
-        :param data:
-        :return: Tuple of recent_run and is_running ('true'/'false')
-        """
-        time = None
-        is_running = None
-        for line in data:
-            line = line.strip()
-            if line.startswith('recent-run:'):
-                time = line.split(':', 1)[1].strip()
-            elif line.startswith('is-running:'):
-                is_running = line.split(':', 1)[1].strip()
-
-        return float(time), is_running
 
     def _read_data_from_file(self):
         """
         Read the recent run time from the file
         :return:
         """
-        with open(self._file_path, 'r') as f:
+        with open(self._file_path, 'rb') as f:
-            data = f.readlines()
+            data = tomllib.load(f)
-            recent_run, is_running = self._read_data(data)
+            self._already_running = data.get('is-running', False)
-
+            recent_run = data.get('recent-run')
-            # Read running status
-            self._is_running = is_running == 'true'
 
             # Set the datetime to the recent run time
             if not recent_run:
                 return
-            try:
+            if isinstance(recent_run, datetime.datetime):
-                self._datetime = dt.fromtimestamp(float(recent_run)) \
+                self._datetime = recent_run
+            elif isinstance(recent_run, float):
+                self._datetime = dt.fromtimestamp(recent_run) \
                     .astimezone(self._timezone)
-            except ValueError as e:
+            else:
                 raise ValueError(
-                    f"Invalid timestamp '{recent_run}' in {self._file_path}: {e}")
+                    f"Invalid recent_run '{recent_run}' in {self._file_path}.")
 
     @property
     def datetime(self) -> dt | None:
@@ -177,14 +155,14 @@ class RecentRun:
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        datetime = None
+        recent_run = None
         self._is_running = False
 
-        # If an exception occurred, do not update the recent run timestamp
+        # If no exception occurred, set the recent run time to the current time
         if exc_type is None:
-            self.datetime = datetime = self._started_at
+            self.datetime = recent_run = self._started_at
 
-        self._sync_file(datetime, is_running=self._is_running)
+        self._sync_file(recent_run=recent_run, is_running=self._is_running)
 
     def __gt__(self, other: dt | str | float):
         return self.datetime > self._to_datetime(other)

uv.lock

@@ -4,13 +4,14 @@ requires-python = ">=3.12"
 
 [[package]]
 name = "adgroupsync"
-version = "0.1.0"
+version = "1.0.0"
 source = { editable = "." }
 dependencies = [
     { name = "civifang" },
     { name = "httpx" },
     { name = "ms-active-directory" },
     { name = "pyyaml" },
+    { name = "tomli-w" },
     { name = "validators" },
 ]
 
@@ -26,6 +27,7 @@ requires-dist = [
     { name = "httpx", specifier = ">=0.28.1" },
     { name = "ms-active-directory", specifier = ">=1.14.1" },
     { name = "pyyaml", specifier = ">=6.0.2" },
+    { name = "tomli-w", specifier = ">=1.2.0" },
     { name = "validators", specifier = ">=0.34.0" },
 ]
 
@@ -447,6 +449,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235 },
 ]
 
+[[package]]
+name = "tomli-w"
+version = "1.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/19/75/241269d1da26b624c0d5e110e8149093c759b7a286138f4efd61a60e75fe/tomli_w-1.2.0.tar.gz", hash = "sha256:2dd14fac5a47c27be9cd4c976af5a12d87fb1f0b4512f81d69cce3b35ae25021", size = 7184 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c7/18/c86eb8e0202e32dd3df50d43d7ff9854f8e0603945ff398974c1d91ac1ef/tomli_w-1.2.0-py3-none-any.whl", hash = "sha256:188306098d013b691fcadc011abd66727d3c414c571bb01b1a174ba8c983cf90", size = 6675 },
+]
+
 [[package]]
 name = "tomlkit"
 version = "0.13.2"