cx-logger/source/handler.py

import os
import sys
import pathlib
import asyncio

class handler:
    """
    That is used to appending data into log files, stdout and stderr.
    It requires file which is used to store logs.

    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.

        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()

    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()