Console and file output
Attach two handlers to one named logger. A simple custom handler signals that dispatch has reached both outputs; then flush, read the file, detach, and close.
"""Run the console and file example against the documented femtologging source."""
from pathlib import Path
from tempfile import TemporaryDirectory
from threading import Event
from femtologging import (
FemtoFileHandler,
FemtoStreamHandler,
get_logger,
)
delivered = Event()
class _DeliverySignal:
"""Acknowledge that dispatch reached the handler set."""
def handle(self, logger: str, level: str, message: str) -> None:
"""Signal delivery without altering the record."""
delivered.set()
signal = _DeliverySignal()
logger = get_logger("survey.output")
logger.set_level("INFO")
logger.set_propagate(False)
console = FemtoStreamHandler.stdout()
with TemporaryDirectory() as directory:
path = Path(directory) / "survey.log"
file_handler = FemtoFileHandler(str(path))
for handler in (console, file_handler):
logger.add_handler(handler)
logger.add_handler(signal)
try:
logger.info("sample prepared")
if not delivered.wait(timeout=2):
message = "Record did not reach the handlers"
raise RuntimeError(message)
if not logger.flush_handlers():
message = "Logging did not flush"
raise RuntimeError(message)
if "sample prepared" not in path.read_text():
message = "File output is missing the expected record"
raise RuntimeError(message)
finally:
logger.remove_handler(signal)
for handler in (console, file_handler):
logger.remove_handler(handler)
handler.close()
Outputsurvey.output [INFO] sample prepared
Per-call request context
A callable formatter serializes the message and metadata.key_values. A per-call field also suits concurrent asyncio operations. The Event acknowledges formatter delivery before the logger flush.
"""Run the request context example against the documented femtologging source."""
import json
import typing as typ
from threading import Event
from femtologging import (
StreamHandlerBuilder,
get_logger,
)
delivered = Event()
def _json_record(record: dict[str, object]) -> str:
"""Format the message and fields, acknowledging formatter delivery."""
result = json.dumps(
{
"message": record["message"],
"fields": typ.cast("dict[str, object]", record["metadata"])["key_values"],
},
sort_keys=True,
)
delivered.set()
return result
logger = get_logger("service.request")
logger.set_level("INFO")
logger.set_propagate(False)
handler = StreamHandlerBuilder.stdout().with_formatter(_json_record).build()
logger.add_handler(handler)
try:
# Per-call fields also suit concurrent asyncio tasks.
logger.info("request accepted", extra={"request_id": 42})
if not delivered.wait(timeout=2):
message = "Record did not reach the formatter"
raise RuntimeError(message)
if not logger.flush_handlers():
message = "Logging did not flush"
raise RuntimeError(message)
finally:
logger.remove_handler(handler)
handler.close()
Output{"fields": {"request_id": "42"}, "message": "request accepted"}
The current native mapping normalizes this integer to a string. Consumers should validate their actual output schema.
Reuse a stdlib handler
The adapter translates to a LogRecord for a configured stdlib stream handler. The observed handler confirms delivery before checking output and closing the adapter.
"""Run the stdlib handler example against the documented femtologging source."""
import io
import logging
from threading import Event
from femtologging import (
StdlibHandlerAdapter,
get_logger,
)
delivered = Event()
class _ObservedHandler(logging.StreamHandler):
"""Acknowledge delivery to the in-memory output."""
def emit(self, record: logging.LogRecord) -> None:
"""Write the record and acknowledge its arrival."""
super().emit(record)
delivered.set()
output = io.StringIO()
stdlib_handler = _ObservedHandler(output)
stdlib_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
adapter = StdlibHandlerAdapter(stdlib_handler)
logger = get_logger("survey.adapter")
logger.set_level("INFO")
logger.set_propagate(False)
logger.add_handler(adapter)
try:
logger.info("interface inspection complete")
if not delivered.wait(timeout=2):
message = "Record did not reach the adapter"
raise RuntimeError(message)
if not logger.flush_handlers():
message = "Logging did not flush"
raise RuntimeError(message)
if output.getvalue() != "INFO: interface inspection complete\n":
message = "Adapter output does not match the expected record"
raise RuntimeError(message)
print(output.getvalue(), end="")
finally:
logger.remove_handler(adapter)
adapter.close()
OutputINFO: interface inspection complete