Logger

Logger

Logger(file: str | Path | None = None)

An asynchronous logger supporting contextual logging with configurable output.

Provides a queue-based logging system that can write to either a file or stdout. Supports contextual logging where entries can be grouped under specific contexts, and includes support for both message and error logging.

Can be used either as an async context manager or directly:

• As a context manager:

  async with Logger("app.log") as logger:
      await logger.log("message")
      # Logger automatically closes
  

• Direct usage:

  logger = Logger("app.log")
  await logger.log("message")
  await logger.aclose()  # Must be explicitly closed
  

Parameters:

Name	Type	Description	Default
file	str | Path | None	Path to the log file. If None, logs will be written to stdout	None

Initialize a new Logger instance.

Parameters:

Name	Type	Description	Default
file	str | Path | None	Path to the log file. If None, logs will be written to stdout.	None

Source code in freeact/logger.py

def __init__(self, file: str | Path | None = None):
    """Initialize a new Logger instance.

    Args:
        file: Path to the log file. If None, logs will be written to stdout.
    """
    self.writer = FileWriter(file) if file else StdoutWriter()
    self.var = ContextVar[List[str]]("context", default=[])
    self.queue = asyncio.Queue()  # type: ignore
    self.runner = asyncio.create_task(self._run())

_get_caller_module_name

_get_caller_module_name()

Get the name of the module that called the logger.

Returns:

Type	Description
	The name of the calling module.

Source code in freeact/logger.py

def _get_caller_module_name(self):
    """Get the name of the module that called the logger.

    Returns:
        The name of the calling module.
    """
    caller_frame = inspect.stack()[2]
    caller_module = inspect.getmodule(caller_frame[0])
    return caller_module.__name__  # type: ignore

_run async

_run()

Internal consumer that processes the log entry queue.

Continuously pulls entries from the queue and writes them using the configured writer.

Source code in freeact/logger.py

async def _run(self):
    """Internal consumer that processes the log entry queue.

    Continuously pulls entries from the queue and writes them using the configured writer.
    """
    while True:
        entry = await self.queue.get()
        await self.writer.write(entry)
        self.queue.task_done()

aclose async

aclose()

Close the logger and process remaining entries.

This method must be called explicitly when using the Logger directly (not as a context manager) to ensure all queued log entries are processed before shutting down.

When using the Logger as a context manager, this method is called automatically.

Source code in freeact/logger.py

async def aclose(self):
    """Close the logger and process remaining entries.

    This method must be called explicitly when using the `Logger` directly (not as
    a context manager) to ensure all queued log entries are processed before
    shutting down.

    When using the `Logger` as a context manager, this method is called automatically.
    """
    # first process all queued entries
    await self.queue.join()
    # then cancel the queue consumer
    self.runner.cancel()

context async

context(frame: str)

A context manager that adds an additional context frame to the current context.

Example

async with logger.context("User Login"):
    await logger.log("Attempting login")  # Logs with "User Login" context
    await db.query(...)
    await logger.log("Login successful")  # Logs with "User Login" context

    # Contexts can be nested
    async with logger.context("Profile"):
        # Logs with "User Login / Profile" context
        await logger.log("Loading user profile")

Parameters:

Name	Type	Description	Default
frame	str	The context frame name to add	required

Yields:

Type	Description
	The Logger instance.

Raises:

Type	Description
Exception	Re-raises any exception that occurs within the context after logging it.

Source code in freeact/logger.py

@asynccontextmanager
async def context(self, frame: str):
    """A context manager that adds an additional context frame to the current context.

    Example:
        ```python
        async with logger.context("User Login"):
            await logger.log("Attempting login")  # Logs with "User Login" context
            await db.query(...)
            await logger.log("Login successful")  # Logs with "User Login" context

            # Contexts can be nested
            async with logger.context("Profile"):
                # Logs with "User Login / Profile" context
                await logger.log("Loading user profile")
        ```

    Args:
        frame: The context frame name to add

    Yields:
        The Logger instance.

    Raises:
        Exception: Re-raises any exception that occurs within the context after logging it.
    """
    context = self.var.get().copy()
    context.append(frame)
    token = self.var.set(context)

    try:
        yield self
    except Exception as e:
        await self.log_error(e)
        raise
    finally:
        self.var.reset(token)

log async

log(message: str, metadata: dict[str, Any] | None = None)

Log a message with optional metadata.

Parameters:

Name	Type	Description	Default
message	str	The message to log	required
metadata	dict[str, Any] | None	Optional dictionary of additional data to include	None

Source code in freeact/logger.py

async def log(self, message: str, metadata: dict[str, Any] | None = None):
    """Log a message with optional metadata.

    Args:
        message: The message to log
        metadata: Optional dictionary of additional data to include
    """
    entry = MessageEntry(
        context=self.var.get(),
        message=message,
        caller=self._get_caller_module_name(),
        metadata=metadata,
    )
    await self.queue.put(entry)

log_error async

log_error(e: Exception)

Log an exception with its full traceback.

Parameters:

Name	Type	Description	Default
e	Exception	The exception to log.	required

Source code in freeact/logger.py

async def log_error(self, e: Exception):
    """Log an exception with its full traceback.

    Args:
        e: The exception to log.
    """
    entry = ErrorEntry(
        context=self.var.get(),
        error=e,
    )
    await self.queue.put(entry)