Continue working on project

source/__init__.py

@@ -0,0 +1,3 @@
+from .levels import levels
+from .logger import logger
+from .handler import handler

source/handler.py

@@ -1,15 +1,118 @@
+import os
+import sys
 import pathlib
+import asyncio
 
 class handler:
-    def __init__(self, target: pathlib.Path) -> None:
-        if target.is_file():
-            self.__create(target)
+    """
+    That is used to appending data into log files, stdout and stderr.
+    It requires file which is used to store logs.
 
-        self.__target = target
+    Methods
+    -------
+    async _to_file(content: str) : None
+        That add content, as new line, to the file.
+
+    async _to_stdout(content: str) : None
+        That add content, as new line, to the stdout.
+
+    async _to_stderr(content: str) : None
+        That add content, as new line, to the stderr.
+    """
+
+    def __init__(self, target: pathlib.Path | None) -> None:
+        """
+        That create required locks, and log file when it is not exists.
 
-    def __create(self, target: pathlib.Path) -> None
-        try:
+        Parameters
+        ----------
+        target : pathlib.Path | None
+            File which would be used to store logs. If it is set to None, then
+            only stdout and stderr could be used.
+        """
+
+        self.__file_lock = asyncio.Lock()
+        self.__stderr_lock = asyncio.Lock()
+        self.__stdout_lock = asyncio.Lock()
+
+        self.__target = target
+        self.__pipe = None
+        
+        if target is not None and not target.exists():
             target.touch()
 
-        except:
-            RuntimeError("Can not create log file: " + str(target) + ".")
+    async def _to_file(self, content: str) -> None:
+        """
+        That adding given content as new line in the file.
+
+        Parameters
+        ----------
+        content : str
+            Content which would be added as new line in the file.
+        """
+
+        if self.__target is None:
+            return
+
+        async with self.__file_lock:
+            await asyncio.to_thread(self.__add_file, content)
+
+    def __add_file(self, content: str) -> None:
+        """
+        That adding given content as new line to the file. It is synchronized
+        function, running in new thread. When file is not open, that open it.
+        
+        Parameters
+        ----------
+        content : str
+            Content to append into file as new line.
+        """
+
+        if self.__pipe is None or self.__pipe.closed:
+            self.__pipe = self.__target.open("a")
+
+        self.__pipe.write(content + os.linesep)
+
+    async def _to_stdout(self, content: str) -> None:
+        """
+        That add content to stdout fifo as new line.
+
+        Parameters
+        ----------
+        content : str
+            New content which would be add to stdout.
+        """
+
+        content = content + os.linesep
+
+        async with self.__stdout_lock:
+            await asyncio.to_thread(sys.stdout.write, content)
+
+    async def _to_stderr(self, content: str) -> None:
+        """
+        That add content to stderr fifo as new line.
+
+        Parameters
+        ----------
+        content : str
+            New content, which would be add to stderr.
+        """
+
+        content = content + os.linesep
+
+        async with self.__stderr_lock:
+            await asyncio.to_thread(sys.stderr.write, content)
+
+    def __del__(self) -> None:
+        """
+        That is object destructor, which close log file, when it is not 
+        closed yet.
+        """
+
+        if self.__pipe is None:
+            return
+
+        if self.__pipe.closed:
+            return 
+
+        self.__pipe.close()

source/levels.py

@@ -0,0 +1,10 @@
+import enum
+
+class levels(enum.Enum):
+    info = 0
+    warning = 1
+    error = 2
+    critical = 3
+
+
+

source/logger.py

@@ -0,0 +1,81 @@
+import time
+import pathlib
+
+from .levels import levels
+from .handler import handler
+
+class logger(handler):
+    def __init__(self, log_file: pathlib.Path | None) -> None:
+        super().__init__(log_file)
+
+        self.__debug = False
+        self.__time_stamp_format = "%d-%m-%Y %H:%M:%S "
+
+        self.__minimum_stdout = levels.critical.value
+        self.__minimum_stderr = levels.warning.value
+        self.__minimum_file = levels.warning.value
+
+    async def info(self, *args, **kwargs) -> None:
+        await self.log(levels.info, *args, **kwargs)
+
+    async def warning(self, *args, **kwargs) -> None:
+        await self.log(levels.warning, *args, **kwargs)
+
+    async def error(self, *args, **kwargs) -> None:
+        await self.log(levels.error, *args, **kwargs)
+
+    async def critical(self, *args, **kwargs) -> None:
+        await self.log(levels.critical, *args, **kwargs)
+
+    async def log(self, level: levels, *args, **kwargs) -> None:
+        value = level.value
+        content = self.__generate(*args, **kwargs)
+
+        if value >= self.__minimum_stdout or self.debug:
+            await self._to_stdout(content)
+
+        if value >= self.__minimum_stderr or self.debug:
+            await self._to_stderr(content)
+
+        if value >= self.__minimum_file or self.debug:
+            await self._to_file(content)
+
+    def __generate(self, *args, **kwargs) -> str:
+        if len(kwargs) > 0:
+            return self.__generate_from_kwargs(*args, **kwargs)
+
+        if len(args) > 0:
+            return self.__generate_from_args(*args)
+        
+        raise RuntimeError("Can not log empty message.")
+
+    def __generate_from_kwargs(self, *args, **kwargs) -> str:
+        if len(args) != 1 or type(args[0]) is not str:
+            RuntimeError("Can not log with format, when format is not given.")
+        
+        return self.time_stamp + args[0].format(**kwargs)
+
+    def __generate_from_args(self, *args) -> str:
+        mapper = lambda count: count if type(count) is str else repr(count)
+        mapped = map(mapper, args)
+
+        return self.time_stamp + str(", ").join(mapped)
+
+    @property
+    def time_stamp(self) -> str:
+        return time.strftime(self.__time_stamp_format, time.localtime())
+
+    def set_time_stamp(self, target: str) -> None:
+            self.__time_stamp_format = target
+
+    @property
+    def debug(self) -> bool:
+        return self.__debug
+    
+    def start_debugging() -> None:
+        self.__debug = True
+
+    def stop_debugging() -> None:
+        self.__debug = False
+
+

tests/000-default.py

@@ -0,0 +1,10 @@
+import pathlib
+import sys
+
+tests_dir = pathlib.Path(__file__).parent
+project_dir = tests_dir.parent
+
+sys.path.append(str(project_dir))
+
+import source
+

tests/001-handler.py

@@ -0,0 +1,38 @@
+import pathlib
+import asyncio
+import sys
+import random
+
+tests_dir = pathlib.Path(__file__).parent
+project_dir = tests_dir.parent
+
+sys.path.append(str(project_dir))
+
+import source
+
+class logger(source.handler):
+    async def write(self, content: str) -> None:
+        await self._to_file(content)
+        await self._to_stdout(content)
+        await self._to_stderr(content)
+
+log = tests_dir / pathlib.Path("x.log")
+
+if log.is_file():
+    log.unlink()
+
+loging = logger(log)
+
+async def main():
+    await loging.write("That is so simple: " + str(random.randrange(1, 20)))
+
+async def core():
+    alls = list()
+
+    for count in range(2000):
+        alls.append(main())
+
+    await asyncio.gather(*alls)
+
+if __name__ == "__main__":
+    asyncio.run(core())

tests/002-logger.py

@@ -0,0 +1,17 @@
+import pathlib
+import sys
+import asyncio
+
+tests_dir = pathlib.Path(__file__).parent
+project_dir = tests_dir.parent
+
+sys.path.append(str(project_dir))
+
+import source
+
+async def main():
+    log = source.logger(tests_dir / pathlib.Path("002.log"))
+    await log.error("sample")
+
+if __name__ == "__main__":
+    asyncio.run(main())