Building Custom Log Sinks ¶

Custom sinks need to implement the Sink protocol.

File Sink ¶

from datetime import datetime

class FileSink(Sink):
    def __init__(self, filepath: str):
        self.filepath = filepath
        self.file = None

    def open(self) -> None:
        self.file = open(self.filepath, "a")

    def close(self) -> None:
        self.file.close()

    async def write(self, records: list[Record]) -> None:
        for record in records:
            line = f"[{record.level}] {record.message}\n"
            self.file.write(line)

# Register while creating app
app = Stario(log_sinks=[FileSink("/var/log/app.log")])