Logly

Appearance

Integrations API ​

All integrations are available as submodules of logly.integrations. Most use importlib.util.find_spec to check for optional dependencies - no installation required if you don't use them.

python
from logly.integrations import fastapi, django, flask

FastAPI ​

LoglyMiddleware ​

python
from logly.integrations.fastapi import LoglyMiddleware

app = FastAPI()
app.add_middleware(LoglyMiddleware, level="INFO")

Parameters:

ParameterTypeDefaultDescription
appFastAPIFastAPI application
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string
backtraceboolFalseInclude backtrace on exceptions
diagnoseboolFalseInclude variable values on exceptions

Django ​

LoglyHandler ​

python
# settings.py
LOGGING = {
    "handlers": {
        "logly": {
            "()": "logly.integrations.django.LoglyHandler",
            "level": "INFO",
        },
    },
    "root": {
        "handlers": ["logly"],
        "level": "INFO",
    },
}

Parameters:

ParameterTypeDefaultDescription
levelstr | int"INFO"Minimum log level
formatstr | NoneNoneCustom format string

LoglyMiddleware ​

python
MIDDLEWARE = [
    "logly.integrations.django.LoglyMiddleware",
    # ... other middleware
]

Parameters:

ParameterTypeDefaultDescription
get_responseCallable | NoneNoneDjango response callable

Flask ​

LoglyHandler ​

python
from flask import Flask
from logly.integrations.flask import LoglyHandler

app = Flask(__name__)
handler = LoglyHandler()
handler.init_app(app)

Methods:

init_app(app, **kwargs) ​

Initialize the handler with a Flask app.

ParameterTypeDefaultDescription
appFlaskFlask application
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

Starlette ​

LoglyMiddleware ​

python
from starlette.applications import Starlette
from logly.integrations.starlette import LoglyMiddleware

app = Starlette()
app.add_middleware(LoglyMiddleware, level="INFO")

Parameters:

ParameterTypeDefaultDescription
appStarletteStarlette application
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

Stdlib Logging ​

InterceptHandler ​

Bridge stdlib logging to Logly.

python
import logging
from logly.integrations.stdlib import InterceptHandler

logging.basicConfig(handlers=[InterceptHandler()], level=logging.INFO)

Parameters:

ParameterTypeDefaultDescription
levelintlogging.NOTSETMinimum log level

Structlog ​

logly_processor ​

python
import structlog

structlog.configure(
    processors=[
        logly.integrations.structlog.logly_processor(
            logger_name="mylogger",
            wrapper_class=structlog.BoundLogger,
        ),
    ],
)

Parameters:

ParameterTypeDefaultDescription
logger_namestr | NoneNoneLogger name key
wrapper_classtype | NoneNoneStructlog wrapper class
levelstr"INFO"Minimum log level

LoglyRenderer ​

python
structlog.configure(
    processors=[
        logly.integrations.structlog.LoglyRenderer(level="INFO"),
    ],
)

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

Rich Console ​

LoglyRichSink ​

python
from logly import logger
from logly.integrations.rich import LoglyRichSink

logger.add(LoglyRichSink(), level="INFO")

Parameters:

ParameterTypeDefaultDescription
fileIO | NoneNoneOutput file (default: stderr)

RichHandler ​

python
from logly.integrations.rich import RichHandler

handler = RichHandler(level="INFO", show_path=False)

Parameters:

ParameterTypeDefaultDescription
levelintlogging.NOTSETMinimum log level
show_pathboolTrueShow file path
show_line_noboolFalseShow line number
show_timeboolTrueShow timestamp
rich_tracebacksboolTrueUse Rich tracebacks

Gunicorn ​

LoglyWorker ​

python
# gunicorn.conf.py
from logly.integrations.gunicorn import LoglyWorker

worker_class = LoglyWorker

setup_gunicorn_logging ​

python
# gunicorn.conf.py
from logly.integrations.gunicorn import setup_gunicorn_logging

setup_gunicorn_logging(level="INFO")

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

Uvicorn ​

setup_uvicorn_logging ​

python
# uvicorn config
from logly.integrations.uvicorn import setup_uvicorn_logging

setup_uvicorn_logging(level="INFO")

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

get_log_config ​

python
from logly.integrations.uvicorn import get_log_config

config = get_log_config(level="INFO")
# Use with uvicorn: uvicorn.run(app, log_config=config)

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

Returns: dict - Uvicorn-compatible log config

Celery ​

setup_celery_logging ​

python
from logly.integrations.celery import setup_celery_logging

setup_celery_logging(level="INFO")

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level
formatstr | NoneNoneCustom format string

patch_task_logger ​

python
from logly.integrations.celery import patch_task_logger


@app.task
def my_task():
    patch_task_logger(app.task_logger, level="INFO")

Parameters:

ParameterTypeDefaultDescription
task_loggerLoggerCelery task logger
levelstr"INFO"Minimum log level

SQLAlchemy ​

setup_sqlalchemy_logging ​

python
from logly.integrations.sqlalchemy import setup_sqlalchemy_logging

setup_sqlalchemy_logging(level="WARNING", echo=False)

Parameters:

ParameterTypeDefaultDescription
levelstr"WARNING"Minimum log level
echoboolFalseEcho SQL statements

patch_engine ​

python
from logly.integrations.sqlalchemy import patch_engine

engine = create_engine("sqlite:///db.sqlite3")
patch_engine(engine, level="INFO")

Parameters:

ParameterTypeDefaultDescription
engineEngineSQLAlchemy engine
levelstr"INFO"Minimum log level

OpenTelemetry ​

OTelLogSink ​

python
from logly import logger
from logly.integrations.opentelemetry import OTelLogSink

logger.add(
    OTelLogSink(
        service_name="myapp",
        endpoint="http://localhost:4318",
        protocol="http",
    )
)

Parameters:

ParameterTypeDefaultDescription
service_namestr"logly"Service name
endpointstr"http://localhost:4318"OTLP endpoint
protocolstr"http"Protocol: "http" or "grpc"
headersdict | NoneNoneRequest headers

Prometheus ​

PrometheusLogSink ​

python
from logly import logger
from logly.integrations.prometheus import PrometheusLogSink

logger.add(PrometheusLogSink(namespace="logly"))

Parameters:

ParameterTypeDefaultDescription
namespacestr"logly"Prometheus namespace
Subsystemstr | NoneNonePrometheus subsystem

Elasticsearch ​

ElasticsearchSink ​

python
from logly import logger
from logly.integrations.elasticsearch import ElasticsearchSink

logger.add(
    ElasticsearchSink(
        endpoint="http://localhost:9200",
        index="logs",
    )
)

Parameters:

ParameterTypeDefaultDescription
endpointstrElasticsearch endpoint
indexstr"logly"Index name
timeoutint30Request timeout (seconds)
usernamestr | NoneNoneBasic auth username
passwordstr | NoneNoneBasic auth password

Sentry ​

SentrySink ​

python
from logly import logger
from logly.integrations.sentry import SentrySink

logger.add(
    SentrySink(
        dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
        environment="production",
        release="1.0.0",
        level="WARNING",
    )
)

Parameters:

ParameterTypeDefaultDescription
dsnstrSentry DSN
environmentstr | NoneNoneEnvironment name
releasestr | NoneNoneRelease version
levelstr"WARNING"Minimum log level

Redis ​

RedisHandler ​

python
from logly import logger
from logly.integrations.redis import RedisHandler

logger.add(
    RedisHandler(
        url="redis://localhost:6379",
        key="logs",
        mode="list",
    )
)

Parameters:

ParameterTypeDefaultDescription
urlstrRedis connection URL
keystr"logly:logs"Redis key
modestr"list"Storage mode: "list" or "stream"
timeoutint5Connection timeout (seconds)
max_stream_lenint10000Max stream length

Kafka ​

KafkaHandler ​

python
from logly import logger
from logly.integrations.kafka import KafkaHandler

logger.add(
    KafkaHandler(
        bootstrap_servers="localhost:9092",
        topic="logs",
    )
)

Parameters:

ParameterTypeDefaultDescription
bootstrap_serversstrKafka broker addresses
topicstr"logly"Kafka topic
client_idstr | NoneNoneClient ID
acksstr | int"all"Acknowledgment mode
timeoutint10Request timeout (seconds)

MongoDB ​

MongoHandler ​

python
from logly import logger
from logly.integrations.mongodb import MongoHandler

logger.add(
    MongoHandler(
        uri="mongodb://localhost:27017",
        database="logs",
        collection="app_logs",
    )
)

Parameters:

ParameterTypeDefaultDescription
uristrMongoDB connection URI
databasestrDatabase name
collectionstr"logs"Collection name
timeoutint5Connection timeout (seconds)

PostgreSQL ​

PostgresHandler ​

python
from logly import logger
from logly.integrations.postgresql import PostgresHandler

logger.add(
    PostgresHandler(
        dsn="postgresql://user:pass@localhost:5432/logs",
        table="app_logs",
    )
)

Parameters:

ParameterTypeDefaultDescription
dsnstrPostgreSQL connection string
tablestr"logs"Table name
create_tableboolTrueAuto-create table

Discord ​

DiscordHandler ​

python
from logly import logger
from logly.integrations.discord import DiscordHandler

logger.add(
    DiscordHandler(
        webhook_url="https://discord.com/api/webhooks/...",
        username="Logly Bot",
    )
)

Parameters:

ParameterTypeDefaultDescription
webhook_urlstrDiscord webhook URL
timeoutint10Request timeout (seconds)
usernamestr | NoneNoneOverride username
avatar_urlstr | NoneNoneOverride avatar URL

Slack ​

SlackHandler ​

python
from logly import logger
from logly.integrations.slack import SlackHandler

logger.add(
    SlackHandler(
        webhook_url="https://hooks.slack.com/services/...",
        channel="#logs",
    )
)

Parameters:

ParameterTypeDefaultDescription
webhook_urlstrSlack webhook URL
channelstr | NoneNoneOverride channel
usernamestr"Logly"Bot username
icon_emojistr":robot_face:"Bot icon emoji
timeoutint10Request timeout (seconds)

Email ​

EmailHandler ​

python
from logly import logger
from logly.integrations.email import EmailHandler

logger.add(
    EmailHandler(
        smtp_host="smtp.gmail.com",
        smtp_port=587,
        from_addr="alerts@myapp.com",
        to_addrs=["admin@myapp.com"],
        username="alerts@myapp.com",
        password="app-password",
        use_tls=True,
    )
)

Parameters:

ParameterTypeDefaultDescription
smtp_hoststrSMTP server host
smtp_portintSMTP server port
from_addrstrSender email address
to_addrslist[str]Recipient email addresses
usernamestr | NoneNoneSMTP username
passwordstr | NoneNoneSMTP password
use_tlsboolTrueUse STARTTLS
use_sslboolFalseUse SSL/TLS
timeoutint30Connection timeout (seconds)
subject_prefixstr""Prefix for email subject

HTTP ​

HttpHandler ​

python
from logly import logger
from logly.integrations.http import HttpHandler

logger.add(
    HttpHandler(
        url="https://api.example.com/logs",
        method="POST",
        headers={"Authorization": "Bearer token"},
    )
)

Parameters:

ParameterTypeDefaultDescription
urlstrHTTP endpoint URL
methodstr"POST"HTTP method
headersdict | NoneNoneRequest headers
timeoutint10Request timeout (seconds)
formatstr | NoneNoneCustom format string

Loki ​

LokiSink ​

python
from logly import logger
from logly.integrations.loki import LokiSink

logger.add(
    LokiSink(
        endpoint="http://localhost:3100",
        labels={"app": "myapp", "env": "production"},
    )
)

Parameters:

ParameterTypeDefaultDescription
endpointstrLoki endpoint URL
labelsdict{}Default labels
timeoutint10Request timeout (seconds)
usernamestr | NoneNoneBasic auth username
passwordstr | NoneNoneBasic auth password

Propagate ​

PropagateHandler ​

python
import logging
from logly.integrations.propagate import PropagateHandler

logging.getLogger("myapp").addHandler(PropagateHandler())

Parameters:

ParameterTypeDefaultDescription
namestr"logly"Logger name
levelintlogging.NOTSETMinimum log level

Telemetry ​

TelemetrySink ​

python
from logly import logger
from logly.integrations.telemetry import TelemetrySink


def send_to_collector(record):
    # Send to your telemetry backend
    pass


logger.add(TelemetrySink(emit=send_to_collector))

Parameters:

ParameterTypeDefaultDescription
emitCallableEmission function
service_namestr"logly"Service name
environmentstr | NoneNoneEnvironment name

HttpJsonSink ​

python
from logly import logger
from logly.integrations.telemetry import HttpJsonSink

logger.add(
    HttpJsonSink(
        endpoint="https://api.example.com/telemetry",
        headers={"Authorization": "Bearer token"},
    )
)

Parameters:

ParameterTypeDefaultDescription
endpointstrHTTP endpoint URL
headersdict | NoneNoneRequest headers
timeoutint10Request timeout (seconds)

APScheduler ​

APSchedulerHandler ​

python
from logly.integrations.apscheduler import APSchedulerHandler

scheduler = APScheduler()
scheduler.add_job(my_job, "interval", seconds=60)
scheduler.add_listener(APSchedulerHandler(), EVENT_JOB_EXECUTED)

setup_apscheduler_logging ​

python
from logly.integrations.apscheduler import setup_apscheduler_logging

setup_apscheduler_logging(level="INFO")

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level

RQ ​

RQHandler ​

python
from logly.integrations.rq import RQHandler

# Attach to RQ worker

setup_rq_logging ​

python
from logly.integrations.rq import setup_rq_logging

setup_rq_logging(level="INFO")

Parameters:

ParameterTypeDefaultDescription
levelstr"INFO"Minimum log level

Click ​

click_echo ​

python
from logly.integrations.click import click_echo

click_echo("Processing...", nl=True, err=True)

Parameters:

ParameterTypeDefaultDescription
messagestrMessage to display
fileIO | NoneNoneOutput stream
nlboolTrueAdd newline
errboolFalseOutput to stderr
colorbool | NoneNoneEnable color

Typer ​

typer_echo ​

python
from logly.integrations.typer import typer_echo

typer_echo("Processing...", nl=True, err=True)

Parameters:

ParameterTypeDefaultDescription
messagestrMessage to display
fileIO | NoneNoneOutput stream
nlboolTrueAdd newline
errboolFalseOutput to stderr
colorbool | NoneNoneEnable color

RabbitMQ ​

RabbitMQHandler ​

python
from logly import logger
from logly.integrations.rabbitmq import RabbitMQHandler

logger.add(
    RabbitMQHandler(
        url="amqp://guest:guest@localhost:5672",
        queue="logs",
    )
)

Parameters:

ParameterTypeDefaultDescription
urlstrRabbitMQ connection URL
queuestr"logly"Queue name
exchangestr | NoneNoneExchange name
routing_keystr | NoneNoneRouting key
durableboolTrueDurable queue
timeoutint10Connection timeout (seconds)

Pydantic ​

PydanticLogHandler ​

python
import logging
from logly.integrations.pydantic import PydanticLogHandler

handler = PydanticLogHandler()
handler.setLevel(logging.INFO)

Routes Python logging records through Logly for Pydantic-based applications. No extra dependencies.

LoglyFormatter ​

python
import logging
from logly.integrations.pydantic import LoglyFormatter

handler = logging.StreamHandler()
handler.setFormatter(LoglyFormatter())
ParameterTypeDefaultDescription
logly_loggerLogger | NoneNoneLogly logger instance (uses global logger)

tqdm ​

TqdmSink ​

python
from logly import logger
from logly.integrations.tqdm import TqdmSink
from tqdm import tqdm

logger.remove()
logger.add(TqdmSink(), colorize=True)

for i in tqdm(range(100)):
    if i % 20 == 0:
        logger.info("Processing item {}", i)
ParameterTypeDefaultDescription
tqdm_instanceAnyNoneOptional tqdm class or instance

Datadog ​

DatadogSink ​

python
from logly import logger
from logly.integrations.datadog import DatadogSink

logger.add(
    DatadogSink(
        api_key="your-api-key",
        service="myapp",
        environment="production",
    )
)
ParameterTypeDefaultDescription
api_keystrDatadog API key
hoststr | NoneNoneHost name
sourcestr | None"python"Log source
servicestr | NoneNoneService name
tagslist[str] | NoneNoneTags list
sitestr"datadoghq.com"Datadog site
timeoutfloat5.0Request timeout (seconds)

New Relic ​

NewRelicSink ​

python
from logly import logger
from logly.integrations.newrelic import NewRelicSink

logger.add(
    NewRelicSink(
        license_key="your-license-key",
        app_name="myapp",
    )
)
ParameterTypeDefaultDescription
license_keystr | NoneNoneNew Relic license key
app_namestr | NoneNoneApplication name

Seq ​

SeqSink ​

python
from logly import logger
from logly.integrations.seq import SeqSink

logger.add(
    SeqSink(
        server_url="http://localhost:5341",
        api_key="your-api-key",
    )
)
ParameterTypeDefaultDescription
server_urlstrSeq server URL
api_keystr | NoneNoneAPI key
event_templatedict[str, Any] | NoneNoneAdditional fields for every event
timeoutfloat5.0Request timeout (seconds)

AWS CloudWatch ​

CloudWatchSink ​

python
from logly import logger
from logly.integrations.aws_cloudwatch import CloudWatchSink

logger.add(
    CloudWatchSink(
        log_group="/myapp/logs",
        log_stream="production",
        region="us-east-1",
    )
)
ParameterTypeDefaultDescription
log_groupstrCloudWatch log group name
log_streamstrCloudWatch log stream name
regionstr | NoneNoneAWS region
aws_access_key_idstr | NoneNoneAWS access key
aws_secret_access_keystr | NoneNoneAWS secret key
batch_sizeint10000Events per batch
flush_intervalfloat5.0Flush interval (seconds)
create_groupboolTrueAuto-create log group
create_streamboolTrueAuto-create log stream

Google Cloud Logging ​

GoogleCloudLoggingSink ​

python
from logly import logger
from logly.integrations.google_cloud_logging import GoogleCloudLoggingSink

logger.add(
    GoogleCloudLoggingSink(
        project_id="my-project",
        log_name="myapp",
    )
)
ParameterTypeDefaultDescription
project_idstrGCP project ID
log_namestr"logly"Log name
resourceAnyNoneMonitored resource
credentialsAnyNoneGCP credentials

Azure Monitor ​

AzureMonitorSink ​

python
from logly import logger
from logly.integrations.azure_monitor import AzureMonitorSink

logger.add(
    AzureMonitorSink(
        connection_string="InstrumentationKey=...",
    )
)
ParameterTypeDefaultDescription
connection_stringstr | NoneNoneApplication Insights connection string
instrumentation_keystr | NoneNoneInstrumentation key (legacy)

Logstash ​

LogstashSink ​

python
from logly import logger
from logly.integrations.logstash import LogstashSink

logger.add(
    LogstashSink(
        host="localhost",
        port=5959,
        protocol="tcp",
    )
)
ParameterTypeDefaultDescription
hoststr"localhost"Logstash host
portint5959Logstash port
protocolstr"tcp"Protocol: "tcp" or "udp"
message_typestr"logstash"Event type
tagslist[str] | NoneNoneEvent tags
key_prefixstr""Key prefix for fields
timeoutfloat5.0Connection timeout (seconds)

Graylog ​

GraylogSink ​

python
from logly import logger
from logly.integrations.graylog import GraylogSink

logger.add(
    GraylogSink(
        host="localhost",
        port=12201,
        protocol="udp",
    )
)
ParameterTypeDefaultDescription
hoststr"localhost"Graylog host
portint12201Graylog port
protocolstr"udp"Protocol: "tcp" or "udp"
graylog_versionstr"1.1"GELF version: "1.0" or "1.1"
chunk_sizeint8192UDP chunk size
facilitystr | NoneNoneFacility name
hostnamestr | NoneNoneOverride hostname
timeoutfloat5.0Connection timeout (seconds)

Released under the MIT License.

Copyright © 2023-2026 Muhammad Fiaz