coretex.logging.logger.LogHandler Class Reference

Inherits StreamHandler.

Public Member Functions
bool	flushLogs (self)
None	reset (self)

Detailed Description

    Custom StreamHandler which intercepts and stores all
    received logs from python std logging module until they
    are all uploaded to Coretex API

Definition at line 90 of file logger.py.

Member Function Documentation

flushLogs()

bool coretex.logging.logger.LogHandler.flushLogs

(

self

)

    Uploads all currently stored logs to Coretex backend

    Returns
    -------
    bool -> True if the upload is successful, False otherwise

Definition at line 149 of file logger.py.

    def flushLogs(self) -> bool:
        """
            Uploads all currently stored logs to Coretex backend
 
            Returns
            -------
            bool -> True if the upload is successful, False otherwise
        """
 
        if len(self.__pendingLogs) == 0:
            return True
 
        if self.currentExperimentId is None:
            self.__pendingLogs.clear()
            return True
 
        snapshot = copy.copy(self.__pendingLogs)
 
        response = networkManager.genericJSONRequest(
            endpoint = "model-queue/add-console-log",
            requestType = RequestType.post,
            parameters = {
                "model_queue_id": self.currentExperimentId,
                "logs": [log.encode() for log in snapshot]
            }
        )
 
        # Only clear logs if they were successfully uploaded to coretex
        if not response.hasFailed():
            self.__clearLogs(snapshot)
 
        return not response.hasFailed()
 

reset()

None coretex.logging.logger.LogHandler.reset

(

self

)

    Resets the internal state of the LogHandler
    Clears all pending logs
    Resets the upload worker thread

Definition at line 182 of file logger.py.

    def reset(self) -> None:
        """
            Resets the internal state of the LogHandler
            Clears all pending logs
            Resets the upload worker thread
        """
 
        self.currentExperimentId = None
        self.__pendingLogs.clear()
        self.__uploadWorker.reset()

---

The documentation for this class was generated from the following file:

• src/coretex/logging/logger.py