Asynchronous logging library for Python (3.9–3.14). The API mirrors the standard logging module — same method names, signatures, hierarchy and semantics — with the logging methods being coroutines. Records are created at the call site and put on a queue; a background consumer performs the handler I/O, so await logger.info(...) never waits for a file write or an HTTP request.
- logging-compatible API: same methods, levels, hierarchy, filters and
LogRecordsemantics as the standardloggingmodule - Background consumer: handler I/O happens off the calling coroutine's path; the consumer starts lazily and survives event loop changes
- Configurable delivery:
awaitresolves on enqueue (default) or after handlers processed the record (delivery="await") - Configurable backpressure: bounded queue with
block(default),drop_newordrop_oldoverflow policies - Stdlib bridge:
captureStdlib()routes third-party library logs (aiohttp, sqlalchemy, ...) through the same async handlers - Async Handlers: non-blocking I/O for streams, files (with size/time rotation), HTTP endpoints with extensible authentication, and Telegram chats
- Buffered Handlers: batch processing for high-volume logging
- Performance Metrics: built-in metrics for loggers, handlers and the queue
- Strict Type Checking: full mypy support with type hints
pip install aiologging# For file handlers
pip install aiologging[aiofiles]
# For HTTP handlers (pick one backend)
pip install aiologging[aiohttp]
pip install aiologging[httpx]
# For Protobuf support
pip install aiologging[protobuf]
# All dependencies
pip install aiologging[all]
# Development dependencies
pip install aiologging[dev]import asyncio
import aiologging
async def main():
aiologging.basicConfig(level=aiologging.INFO)
logger = aiologging.getLogger("app")
await logger.info("Application started")
await logger.warning("Something might be wrong")
await logger.error("An error occurred: %s", "details")
# once, before the loop goes away: drain the queue, close handlers
await aiologging.shutdown()
asyncio.run(main())Module-level convenience functions work like in standard logging:
await aiologging.warning("Logged on the root logger")await logger.info(...) synchronously creates the LogRecord (so caller info, %-formatting and exc_info behave exactly like standard logging) and puts it on a bounded queue. A background task drains the queue and dispatches records to the async handlers. By default the await resolves as soon as the record is enqueued — logging is nearly free for the calling coroutine.
Lifecycle:
- the consumer starts lazily on the first logged record and is rebuilt transparently if the event loop changes (e.g. one loop per test);
await aiologging.flush()waits until everything queued has been handled;await aiologging.shutdown()drains the queue and closes all handlers — call it once at application exit.
aiologging.basicConfig(
level=aiologging.INFO,
queue_size=10_000, # queue capacity
overflow="block", # "block" | "drop_new" | "drop_old"
delivery="enqueue", # "enqueue" | "await"
)
# per-logger override for critical logs: the await resolves only
# after the handlers have processed the record
audit = aiologging.getLogger("app.audit")
audit.delivery = "await"overflow="block"(default): when the queue is full,await logger.info(...)waits for free space — no records are lost.overflow="drop_new"/"drop_old": the call never waits; the incoming or the oldest record is discarded and counted in metrics.
Third-party libraries log via the standard logging module. Route their records through your async handlers:
aiologging.basicConfig(level=aiologging.INFO, capture_stdlib=True)
# or:
aiologging.captureStdlib()Bridged records are routed through the aiologging hierarchy under the same logger name, so per-name handlers and propagation work as usual. The bridge is thread-safe and buffers records emitted before the event loop starts.
Complete runnable examples live in the examples/ directory:
- examples/basic_usage.py — levels,
basicConfig,exc_info, delivery modes,flush/shutdown - examples/stdlib_capture.py — routing third-party (stdlib
logging) records through async handlers, including from threads and before the loop starts - examples/file_logging.py — file handler, size- and time-based rotation
- examples/http_logging.py — JSON batches over HTTP with a custom authenticator (includes a local test collector)
- examples/telegram_logging.py — sending records to a Telegram chat, including rate-limit handling (includes a local Bot API stand-in)
- examples/config_usage.py — configuring loggers from a dictionary
import sys
import aiologging
async def main():
logger = aiologging.getLogger("app")
logger.addHandler(aiologging.AsyncStreamHandler(sys.stdout))
await logger.info("This goes to stdout")
await aiologging.shutdown()import aiologging
async def main():
logger = aiologging.getLogger("app")
logger.addHandler(aiologging.AsyncFileHandler("app.log"))
await logger.info("This goes to app.log")
await aiologging.shutdown()import aiologging
async def main():
logger = aiologging.getLogger("app")
# Size-based rotation
logger.addHandler(aiologging.AsyncRotatingFileHandler(
"app.log",
max_bytes=1024*1024, # 1MB
backup_count=5,
))
# Time-based rotation
logger.addHandler(aiologging.AsyncTimedRotatingFileHandler(
"timed.log",
when="midnight",
backup_count=7,
))
await logger.info("This will be rotated")
await aiologging.shutdown()import aiologging
async def main():
logger = aiologging.getLogger("app")
logger.addHandler(aiologging.AsyncHttpHandler(
"https://api.example.com/logs",
headers={"Authorization": "Bearer token"},
))
await logger.info("This will be sent via HTTP")
await aiologging.shutdown()HTTP handlers work on top of either aiohttp or httpx. By default
aiohttp is used when installed, with a fallback to httpx. The backend
can also be selected explicitly:
http_handler = aiologging.AsyncHttpHandler(
"https://api.example.com/logs",
backend="httpx", # or "aiohttp"
)Sends log records to a Telegram chat via the Bot API. Buffered
records are combined into as few sendMessage calls as possible,
each within the 4096-character limit; a 429 Too Many Requests
response is retried after the delay the Bot API returns in
retry_after.
import aiologging
async def main():
logger = aiologging.getLogger("app")
logger.addHandler(aiologging.AsyncTelegramHandler(
token="123456:ABC-DEF...", # from @BotFather
chat_id="-1001234567890", # or "@channelname"
level=aiologging.ERROR,
parse_mode=None, # or "HTML" / "MarkdownV2"
))
await logger.error("This will be sent to Telegram")
await aiologging.shutdown()import aiologging
async def oauth_authenticator(session, request_data):
"""Custom OAuth authentication."""
token = await refresh_oauth_token()
return {"Authorization": f"Bearer {token}"}
http_handler = aiologging.AsyncHttpHandler(
"https://api.example.com/logs",
authenticator=oauth_authenticator,
)json_handler = aiologging.AsyncHttpJsonHandler(
"https://api.example.com/logs"
)text_handler = aiologging.AsyncHttpTextHandler(
"https://api.example.com/logs"
)proto_handler = aiologging.AsyncHttpProtoHandler(
"https://api.example.com/logs"
)universal_handler = aiologging.AsyncHttpHandler(
"https://api.example.com/logs",
format_type="application/json" # Optional: auto-detected if not specified
)import aiologging
class CustomFilter:
def filter(self, record):
return "important" in record.getMessage()
logger = aiologging.getLogger("app")
logger.addFilter(CustomFilter())Formatters live on handlers, exactly like in standard logging:
import logging
import aiologging
handler = aiologging.AsyncStreamHandler()
handler.setFormatter(logging.Formatter(
"%(asctime)s [%(levelname)s] %(name)s: %(message)s"
))
aiologging.getLogger("app").addHandler(handler)import aiologging
async def error_handler(record, exception):
"""Custom error handler for failed log operations."""
print(f"Failed to log {record.getMessage()}: {exception}")
handler = aiologging.AsyncStreamHandler()
handler.error_handler = error_handlerfrom aiologging.logger import _logger_manager
logger = aiologging.getLogger("app")
print(logger.get_metrics()) # per-logger counters
print(_logger_manager.get_metrics()) # queue length, dropped records
for handler in logger.handlers:
print(handler.get_metrics()) # per-handler countersimport aiologging
from aiologging.types import BatchConfig
http_handler = aiologging.AsyncHttpHandler(
"https://api.example.com/logs",
batch_config=BatchConfig(
batch_size=100,
flush_interval=5.0,
max_retries=3,
),
)import asyncio
import aiologging
config = {
"version": 1,
"loggers": {
"myapp": {
"level": "INFO",
"handlers": ["console", "file"]
}
},
"handlers": {
"console": {
"class": "stream",
"level": "INFO",
"stream": "stdout"
},
"file": {
"class": "file",
"level": "DEBUG",
"filename": "app.log",
"mode": "a"
}
}
}
aiologging.configure_from_dict(config)
async def main():
logger = aiologging.get_configured_logger("myapp")
await logger.info("This uses configured logger")
await aiologging.shutdown()
asyncio.run(main())aiologging.configure_from_file("logging_config.json")# before
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("app")
logger.info("Message")
# after
import aiologging
aiologging.basicConfig(level=aiologging.INFO)
logger = aiologging.getLogger("app")
await logger.info("Message") # + await
await aiologging.shutdown() # once, at application exitKey differences:
- Await the logging methods —
debug/info/.../critical,logandexceptionare coroutines; everything else (setLevel,addHandler,getChild, ...) stays synchronous. - Call
await aiologging.shutdown()at exit — the queue must be drained while the event loop is still running. - Handlers are async — use the
Async*handler classes (or bridge stdlib records withcaptureStdlib()).
Async (require await):
await logger.log(level, msg, *args, **kwargs)await logger.debug/info/warning/error/critical(msg, *args, **kwargs)await logger.exception(msg, *args, exc_info=True, **kwargs)
All accept the standard keyword arguments: exc_info, extra, stack_info, stacklevel.
Sync (identical to logging.Logger):
setLevel(level),getEffectiveLevel(),isEnabledFor(level)addHandler(h)/removeHandler(h)/hasHandlers()addFilter(f)/removeFilter(f)/filter(record)getChild(suffix)/getChildren()makeRecord(...),findCaller(stack_info, stacklevel)- attributes:
name,level,parent,propagate,handlers,filters,disabled
getLogger(name=None)— hierarchical loggers; no name returns the root loggerbasicConfig(level, format, datefmt, handlers, force, queue_size, overflow, delivery, capture_stdlib)await flush()— wait until every queued record has been handledawait shutdown()— drain the queue and close all handlersdisable(level)— likelogging.disablecaptureStdlib(capture=True, level=NOTSET)— bridge stdlib logging recordsawait debug/info/warning/error/exception/critical/log(...)— root-logger convenience coroutines
AsyncHandler- Base async handlerAsyncStreamHandler- Stream output handlerAsyncFileHandler- File output handler (requires aiofiles)AsyncRotatingFileHandler- Size-based rotation (requires aiofiles)AsyncTimedRotatingFileHandler- Time-based rotation (requires aiofiles)AsyncHttpHandler- Universal HTTP handler (requires aiohttp or httpx)AsyncHttpTextHandler- Plain text HTTP handler (requires aiohttp or httpx)AsyncHttpJsonHandler- JSON HTTP handler (requires aiohttp or httpx)AsyncHttpProtoHandler- Protobuf HTTP handler (requires aiohttp or httpx, plus protobuf)StdlibBridgeHandler- Synclogging.Handlerforwarding records into aiologging
BatchConfig- Batch processing configurationFileConfig- File handler configurationHttpConfig- HTTP handler configurationLoggerConfig- Logger configurationRotationConfig- Rotation configuration for file handlers
AiologgingError- Base exception for all aiologging errorsHandlerError- Base exception for handler errorsConfigurationError- Configuration-related errorsDependencyError- Missing optional dependenciesAuthenticationError- Authentication failuresNetworkError- Network-related errorsFileError- File operation errorsRotationError- File rotation errorsBatchError- Batch processing errorsFormatterError- Formatting errorsLoggerError- Logger operation errorsContextError- Context manager errors
- Delivery mode: the default
enqueuemakes logging nearly free for the caller; reservedelivery="await"for records that must be confirmed - Backpressure: pick the overflow policy consciously —
blocknever loses records,drop_*never stalls the application - Buffering: use buffered/batching handlers for high-volume logging
- Metrics Collection: enable metrics to monitor drops and errors
- Rate Limiting: use rate limiters to prevent log flooding
- Shutdown: always
await aiologging.shutdown()before exit so nothing queued is lost
# Install development dependencies
pip install aiologging[dev]
# Run tests
pytest
# Run tests with coverage
pytest --cov=aiologging
# Run type checking
mypy aiologging- Fork the repository
- Create a feature branch
- Add tests for new functionality
- Ensure all tests pass and type checking succeeds
- Submit a pull request
MIT License - see LICENSE file for details.
Breaking: the logging pipeline is now queue-based and the API strictly follows the standard logging module.
- Records are enqueued at the call site and dispatched by a background consumer —
await logger.info(...)no longer waits for handler I/O - Configurable
delivery("enqueue"/"await") andoverflow("block"/"drop_new"/"drop_old") viabasicConfigor per logger await aiologging.flush()/await aiologging.shutdown()lifecycle; the consumer starts lazily and survives event loop changes- Stdlib bridge:
captureStdlib()/basicConfig(capture_stdlib=True)routes standardloggingrecords through async handlers - Module-level convenience coroutines (
aiologging.info(...), ...) andaiologging.disable(level) - API aligned with
logging.Logger:getEffectiveLevel,getChild,getChildren,hasHandlers,makeRecord,findCaller,stacklevelsupport,parent/propagate/filtersattributes;getLogger()without arguments returns the root logger - Removed:
Logger.setFormatter/getFormatter(formatters live on handlers),getLevel(usegetEffectiveLevel),disable()/enable()methods (use thedisabledattribute oraiologging.disable),setParent/getParent,getRootLogger,getLoggerContext,log_async async with logger:now flushes instead of closing the shared logger instance- Handlers can be constructed outside a running event loop on Python 3.9
- Python 3.13 and 3.14 support
- Replace deprecated
asyncio.iscoroutinefunctionwithinspect.iscoroutinefunction - Remove dead Python 3.8 compatibility branches
- Initial release
- Full async logging API
- Stream, file, and HTTP handlers
- File rotation support
- Extensible authentication
- Optional dependencies
- Strict type checking