Sink Management API Reference¶

from logly import logger

# Rotate at midnight, keep last 7 days
id = logger.add("app.log", rotation="daily", retention=7)

logger.info("Application started")
logger.error("Database connection failed", retry=3)

from logly import logger

# Rotate every hour, keep last 24 hours
id = logger.add("app.log", rotation="hourly", retention=24)

logger.info("Hourly checkpoint")

from logly import logger

# Rotate every minute, keep last 60 minutes
id = logger.add("test.log", rotation="minutely", retention=60)

logger.debug("Minutely test log")

from logly import logger

# Rotate when file reaches 10MB, keep 5 archives
id = logger.add("app.log", size_limit="10MB", retention=5)

for i in range(100000):
    logger.info(f"Log entry {i}", data="x" * 100)

from logly import logger

# Bytes
logger.add("tiny.log", size_limit="500B", retention=10)

# Kilobytes
logger.add("small.log", size_limit="5KB", retention=20)

# Megabytes (most common)
logger.add("app.log", size_limit="50MB", retention=5)

# Gigabytes (large applications)
logger.add("bigdata.log", size_limit="2GB", retention=3)

from logly import logger

# Both time AND size rotation
# Rotates whichever condition is met first
id = logger.add(
    "app.log",
    rotation="daily",       # Rotate at midnight
    size_limit="100MB",     # OR when file reaches 100MB
    retention=30            # Keep 30 most recent files
)

logger.info("Dual rotation strategy active")

from logly import logger

# Error-only sink
error_id = logger.add("errors.log", filter_min_level="ERROR")

# Info and above
app_id = logger.add("app.log", filter_min_level="INFO")

# Everything (trace and above)
debug_id = logger.add("debug.log", filter_min_level="TRACE")

# Now log at different levels
logger.trace("Trace message")      # Only to debug.log
logger.debug("Debug message")      # Only to debug.log
logger.info("Info message")        # To app.log and debug.log
logger.error("Error message")      # To all three sinks

from logly import logger

# Authentication module logs only
auth_id = logger.add("auth.log", filter_module="app.auth")

# Database module logs only
db_id = logger.add("database.log", filter_module="app.database")

# API module logs only
api_id = logger.add("api.log", filter_module="app.api")

# Module name is auto-detected from caller
logger.info("User login successful", user="alice")  # Goes to auth.log if called from app.auth

from logly import logger

# Only logs from payment processing
payment_id = logger.add("payments.log", filter_function="process_payment")

# Only logs from authentication handler
auth_id = logger.add("auth_handler.log", filter_function="authenticate_user")

# Function name is auto-detected
def process_payment(amount, user):
    logger.info(f"Processing payment", amount=amount, user=user)
    # This goes to payments.log

process_payment(100.50, "alice")

from logly import logger

# Error-level logs from auth module only
id = logger.add(
    "auth_errors.log",
    filter_min_level="ERROR",
    filter_module="app.auth"
)

# Critical logs from specific function
id2 = logger.add(
    "critical_payments.log",
    filter_min_level="CRITICAL",
    filter_function="process_payment"
)

from logly import logger

# Async enabled by default (recommended)
id = logger.add("app.log", async_write=True)

# High-throughput logging with zero blocking
for i in range(10000):
    logger.info(f"Processing item {i}", status="active")

# Main thread continues immediately
# Background thread handles all file I/O

from logly import logger

# High-volume application with custom async settings
id = logger.add(
    "high_volume.log",
    async_write=True,
    buffer_size=32768,          # 32KB buffer (4x default)
    flush_interval=500,         # Flush every 500ms (5x default)
    max_buffered_lines=5000     # 5000 lines max (5x default)
)

# Can handle massive log bursts
for i in range(100000):
    logger.info(f"High-volume log {i}")

from logly import logger

# Synchronous writing for critical logs
id = logger.add("critical.log", async_write=False)

logger.critical("Payment processed", amount=1000000, user="alice")
# Guaranteed written to disk before continuing

from logly import logger

# Enable JSON output
id = logger.add("data.log", json=True)

logger.info("User login", user="alice", ip="192.168.1.1", session="abc123")
logger.error("Database error", code=500, query="SELECT * FROM users")

from logly import logger

id = logger.add(
    "events.log",
    json=True,
    date_enabled=True,
    date_style="rfc3339"
)

logger.info("Event occurred", event="user_signup", tier="premium")

from logly import logger

# Perfect for streaming and log aggregation
id = logger.add("stream.log", json=True)

for event in events:
    logger.info("Event", type=event.type, data=event.data)

from logly import logger

# Simple format with time, level, and message
id = logger.add(
    "simple.log",
    format="{time} [{level}] {message}"
)

logger.info("Application started", version="1.0.0")
logger.error("Connection failed", service="database")

from logly import logger

# Include module, function, filename, and line number
id = logger.add(
    "debug.log",
    format="{time} [{level}] {message} - {module}:{function} ({filename}:{lineno})"
)

logger.debug("Processing request", request_id="abc123")

from logly import logger

# Just level and message
id = logger.add("minimal.log", format="{level}: {message}")

logger.info("User logged in", user="alice")
logger.error("Payment failed", amount=100.50)

from logly import logger

# Include specific extra fields in format
id = logger.add(
    "requests.log",
    format="{time} | {level} | {message} | user={user} request_id={request_id}"
)

logger.info("API request", user="alice", request_id="xyz789", endpoint="/api/users")

from logly import logger

# Add sink
handler_id = logger.add("temp.log")
logger.info("Temporary message")

# Remove sink
success = logger.remove(handler_id)
print(f"Removed: {success}")  # Output: Removed: True

# Subsequent logs don't go to temp.log
logger.info("Not logged to temp.log")

from logly import logger

app_id = logger.add("app.log")
error_id = logger.add("errors.log")
debug_id = logger.add("debug.log")

# Keep app.log, remove others
logger.remove(error_id)
logger.remove(debug_id)

logger.info("Only in app.log")

from logly import logger

# Try removing non-existent sink
success = logger.remove(999)
if not success:
    logger.warning("Sink not found")

from logly import logger

logger.add("app.log")
logger.add("error.log")
logger.add("debug.log")

logger.info("Some logs")

# Remove all sinks
count = logger.remove_all()
print(f"Removed {count} sinks")  # Output: Removed 3 sinks

import pytest
from logly import logger

@pytest.fixture(autouse=True)
def cleanup_logs():
    yield
    # Clean up all sinks after each test
    logger.complete()       # Flush pending writes
    logger.delete_all()     # Delete files
    logger.remove_all()     # Remove sinks

from logly import logger

def reconfigure_logging(environment):
    # Remove all existing sinks
    logger.remove_all()

    # Add new configuration
    if environment == "production":
        logger.add("app.log", rotation="daily", retention=30)
        logger.add("errors.log", filter_min_level="ERROR")
    else:
        logger.add("console")
        logger.add("debug.log", filter_min_level="DEBUG")

reconfigure_logging("production")

from logly import logger

logger.add("app.log")
logger.add("error.log")

count = logger.sink_count()
print(f"Active sinks: {count}")  # Output: Active sinks: 2

from logly import logger

# Ensure at least one sink is configured
if logger.sink_count() == 0:
    logger.add("console")
    logger.warning("No sinks found, added console sink")

from logly import logger

def check_logging_health():
    count = logger.sink_count()
    if count == 0:
        logger.add("fallback.log")
        print("WARNING: No sinks configured, added fallback")
    elif count > 10:
        print(f"WARNING: Too many sinks ({count}), may impact performance")
    else:
        print(f"✓ Logging healthy: {count} sinks active")

check_logging_health()

from logly import logger

id1 = logger.add("app.log")
id2 = logger.add("error.log")
id3 = logger.add("debug.log")

ids = logger.list_sinks()
print(f"Sink IDs: {ids}")  # Output: Sink IDs: [1, 2, 3]

from logly import logger

logger.add("app.log", rotation="daily")
logger.add("error.log", filter_min_level="ERROR")
logger.add("debug.log", filter_min_level="DEBUG")

# Process each sink
for sink_id in logger.list_sinks():
    info = logger.sink_info(sink_id)
    print(f"Sink {sink_id}: {info['path']}")

from logly import logger

app_id = logger.add("app.log")
error_id = logger.add("errors.log")
debug_id = logger.add("debug.log")

# Remove all except app.log
for sink_id in logger.list_sinks():
    if sink_id != app_id:
        logger.remove(sink_id)

print(f"Remaining sinks: {logger.list_sinks()}")  # [1]

from logly import logger

id = logger.add("app.log", rotation="daily", retention=7)

info = logger.sink_info(id)
print(f"Path: {info['path']}")           # app.log
print(f"Rotation: {info['rotation']}")   # daily
print(f"Retention: {info['retention']}") # 7

from logly import logger

id = logger.add(
    "app.log",
    rotation="daily",
    size_limit="10MB",
    retention=7,
    filter_min_level="INFO",
    async_write=True,
    buffer_size=16384,
    json=True
)

info = logger.sink_info(id)
for key, value in info.items():
    print(f"{key}: {value}")

from logly import logger

info = logger.sink_info(999)  # Non-existent ID
if info is None:
    logger.error("Sink not found")
else:
    logger.info(f"Found sink: {info['path']}")

from logly import logger

logger.add("app.log", rotation="daily", retention=7)
logger.add("errors.log", filter_min_level="ERROR", json=True)
logger.add("debug.log", filter_min_level="DEBUG")

all_sinks = logger.all_sinks_info()
for sink in all_sinks:
    print(f"\nSink {sink['id']}:")
    print(f"  Path: {sink['path']}")
    print(f"  Rotation: {sink['rotation']}")
    print(f"  JSON: {sink['json']}")

from logly import logger

logger.add("app.log", rotation="daily", async_write=True)
logger.add("errors.log", filter_min_level="ERROR", size_limit="50MB")
logger.add("data.log", json=True, async_write=True)

print("=== Logging Configuration ===")
for sink in logger.all_sinks_info():
    print(f"\n📁 {sink['path']}")
    if sink['rotation']:
        print(f"   🔄 Rotation: {sink['rotation']}")
    if sink['size_limit']:
        print(f"   📏 Size limit: {sink['size_limit']}")
    if sink['filter_min_level']:
        print(f"   🎯 Min level: {sink['filter_min_level']}")
    if sink['json']:
        print(f"   📋 Format: JSON")
    if sink['async_write']:
        print(f"   ⚡ Async: enabled (buffer: {sink['buffer_size']} bytes)")

from logly import logger
import json

logger.add("app.log", rotation="daily", retention=7)
logger.add("errors.log", filter_min_level="ERROR")

# Export configuration
config = logger.all_sinks_info()
with open("logging_config.json", "w") as f:
    json.dump(config, f, indent=2)

print("Configuration exported to logging_config.json")