added appconfigmonitor

directory_monitor.py

@@ -14,17 +14,17 @@ logger = logging.getLogger(__name__)
 class DirectoryMonitor:
     dir: Path
     glob: str = '*'
-    times: Dict[Path | anyio.Path, datetime] = field(default_factory=dict, repr=False)
-    changed: List[Path] = field(init=False, repr=False)
-    new: List[Path] = field(init=False, repr=False)
-    deleted: List[Path] = field(init=False, repr=False)
+    times: Dict[Path | anyio.Path, float] = field(default_factory=dict, repr=False)
+    changed: List[Path | anyio.Path] = field(init=False, repr=False)
+    new: List[Path | anyio.Path] = field(init=False, repr=False)
+    deleted: List[Path | anyio.Path] = field(init=False, repr=False)
     last_checked: datetime = field(default_factory=datetime.now)
 
     def _reset_counts(self):
         self.last_checked = datetime.now()
         self.new, self.changed, self.deleted = [], [], []
 
-    def _process_file(self, file: Path | anyio.Path, modified_time: float):
+    def _update_file(self, file: Path | anyio.Path, modified_time: float):
         if file in self.times:
             if modified_time > self.times[file]:
                 logger.info(f'Updated file: {file}')
@@ -35,6 +35,11 @@ class DirectoryMonitor:
 
         self.times[file] = modified_time
 
+    def _delete_file(self, file: Path | anyio.Path):
+        logger.warning(f'Detected deleted file: {file}')
+        del self.times[file]
+        self.deleted.append(file)
+
     def rglob(self) -> Iterable[Path]:
         for file in self.dir.rglob(self.glob):
             if file.is_dir():
@@ -56,28 +61,23 @@ class DirectoryMonitor:
         self._reset_counts()
 
         for file in self.rglob():
-            self._process_file(file, file.stat().st_mtime)
+            self._update_file(file, file.stat().st_mtime)
 
         for file in list(self.times.keys()):
             if not file.exists():
-                logger.warning(f'Detected deleted file: {file}')
-                del self.times[file]
-                self.deleted.append(file)
+                self._delete_file(file)
 
     async def update_async(self):
-        logger.debug('Update async')
+        logger.debug('Async update')
         self._reset_counts()
 
         async for file in self.async_rglob():
             modified_time: float = (await file.stat()).st_mtime
-            self._process_file(file, modified_time)
+            self._update_file(file, modified_time)
 
         for file in list(self.times.keys()):
-            exists = await file.exists()
-            if not exists:
-                logger.warning(f'Detected deleted file: {file}')
-                del self.times[file]
-                self.deleted.append(file)
+            if not (await file.exists()):
+                self._delete_file(file)
 
     async def monitor(self):
         while True:
@@ -87,16 +87,49 @@ class DirectoryMonitor:
                 # if a KeyboardInterrupt happens during the update, pass it through
                 raise
             except Exception as e:
-                # if anything unexpected happens, log it and break the loop
+                # if anything unexpected happens, log it and break the loop cleanly
                 logger.exception(e)
                 break
             else:
                 await asyncio.sleep(1.0)
 
 
+@dataclass
+class AppConfigMonitor:
+    dir: str | Path
+    poll_sleep_time: float = 0.5
+    config_monitor: DirectoryMonitor = field(init=False)
+    module_monitor: DirectoryMonitor = field(init=False)
+
+    def __post_init__(self):
+        self.dir = Path(self.dir)
+        assert self.dir.exists()
+        self.config_monitor = DirectoryMonitor(self.dir, '*.yaml')
+        self.module_monitor = DirectoryMonitor(self.dir, '*.py')
+
+    async def monitor(self):
+        while True:
+            try:
+                await self.config_monitor.update_async()
+                await self.module_monitor.update_async()
+            except KeyboardInterrupt:
+                # if a KeyboardInterrupt happens during the update, pass it through
+                raise
+            except Exception as e:
+                # if anything unexpected happens, log it and break the loop cleanly
+                logger.exception(e)
+                break
+            else:
+                await asyncio.sleep(self.poll_sleep_time)
+
+
 if __name__ == '__main__':
     import logging.config
 
+    from rich.console import Console
+
+    console = Console()
+
     logging.config.dictConfig(
         {
             'version': 1,
@@ -107,15 +140,25 @@ if __name__ == '__main__':
                     '()': 'rich.logging.RichHandler',
                     'omit_repeated_times': False,
                     'highlighter': None,
+                    'console': console,
                 }
             },
             'loggers': {__name__: {'level': 'DEBUG', 'handlers': ['rich']}},
         }
     )
 
-    dm = DirectoryMonitor(Path('../conf/apps'), '*.py')
+    # dm = DirectoryMonitor(Path('../conf/apps'), '*.py')
+
+    # try:
+    #     with console.status(f'Monitoring {dm.dir}'):
+    #         asyncio.run(dm.monitor())
+    # except KeyboardInterrupt:
+    #     logger.error('KeyboardInterrupt')
+
+    acm = AppConfigMonitor('../conf/apps')
 
     try:
-        asyncio.run(dm.monitor())
+        with console.status(f'Monitoring {acm.dir}'):
+            asyncio.run(acm.monitor())
     except KeyboardInterrupt:
         logger.error('KeyboardInterrupt')