logfire
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseLogfire
Logfire
Structured observability for Python using Pydantic Logfire - fast setup, powerful features, OpenTelemetry-compatible.
基于Pydantic Logfire的Python结构化可观测性工具——设置快速、功能强大,且兼容OpenTelemetry。
Quick Start
快速开始
bash
uv pip install logfirepython
import logfire
logfire.configure(service_name="my-api", service_version="1.0.0")
logfire.info("Application started")bash
uv pip install logfirepython
import logfire
logfire.configure(service_name="my-api", service_version="1.0.0")
logfire.info("Application started")Core Patterns
核心模式
1. Service Configuration
1. 服务配置
Always set service metadata at startup:
python
import logfire
logfire.configure(
service_name="backend",
service_version="1.0.0",
environment="production",
console=False, # Disable console output in production
send_to_logfire=True, # Send to Logfire platform
)启动时务必设置服务元数据:
python
import logfire
logfire.configure(
service_name="backend",
service_version="1.0.0",
environment="production",
console=False, # 生产环境中禁用控制台输出
send_to_logfire=True, # 发送至Logfire平台
)2. Framework Instrumentation
2. 框架插桩
Instrument frameworks before creating clients/apps:
python
import logfire
from fastapi import FastAPI在创建客户端/应用之前对框架进行插桩:
python
import logfire
from fastapi import FastAPIConfigure FIRST
先配置
logfire.configure(service_name="backend")
logfire.configure(service_name="backend")
Then instrument
然后插桩
logfire.instrument_fastapi()
logfire.instrument_httpx()
logfire.instrument_sqlalchemy()
logfire.instrument_fastapi()
logfire.instrument_httpx()
logfire.instrument_sqlalchemy()
Then create app
再创建应用
app = FastAPI()
undefinedapp = FastAPI()
undefined3. Log Levels and Structured Logging
3. 日志级别与结构化日志
python
undefinedpython
undefinedAll log levels (trace → fatal)
所有日志级别(trace → fatal)
logfire.trace("Detailed trace", step=1)
logfire.debug("Debug context", variable=locals())
logfire.info("User action", action="login", success=True)
logfire.notice("Important event", event_type="milestone")
logfire.warn("Potential issue", threshold_exceeded=True)
logfire.error("Operation failed", error_code=500)
logfire.fatal("Critical failure", component="database")
logfire.trace("Detailed trace", step=1)
logfire.debug("Debug context", variable=locals())
logfire.info("User action", action="login", success=True)
logfire.notice("Important event", event_type="milestone")
logfire.warn("Potential issue", threshold_exceeded=True)
logfire.error("Operation failed", error_code=500)
logfire.fatal("Critical failure", component="database")
Python 3.11+ f-string magic (auto-extracts variables)
Python 3.11+ f-string魔法(自动提取变量)
user_id = 123
status = "active"
logfire.info(f"User {user_id} status: {status}")
user_id = 123
status = "active"
logfire.info(f"User {user_id} status: {status}")
Equivalent to: logfire.info("User {user_id}...", user_id=user_id, status=status)
等价于:logfire.info("User {user_id}...", user_id=user_id, status=status)
Exception logging with automatic traceback
异常日志,自动包含回溯信息
try:
risky_operation()
except Exception:
logfire.exception("Operation failed", context="extra_info")
undefinedtry:
risky_operation()
except Exception:
logfire.exception("Operation failed", context="extra_info")
undefined4. Manual Spans
4. 手动追踪跨度(Spans)
python
undefinedpython
undefinedSpans for tracing operations
用于追踪操作的跨度
with logfire.span("Process order {order_id}", order_id="ORD-123"):
logfire.info("Validating cart")
# ... processing logic
logfire.info("Order complete")
with logfire.span("Process order {order_id}", order_id="ORD-123"):
logfire.info("Validating cart")
# ... 处理逻辑
logfire.info("Order complete")
Dynamic span attributes
动态跨度属性
with logfire.span("Database query") as span:
results = execute_query()
span.set_attribute("result_count", len(results))
span.message = f"Query returned {len(results)} results"
undefinedwith logfire.span("Database query") as span:
results = execute_query()
span.set_attribute("result_count", len(results))
span.message = f"Query returned {len(results)} results"
undefined5. Custom Metrics
5. 自定义指标
python
undefinedpython
undefinedCounter - monotonically increasing
计数器 - 单调递增
request_counter = logfire.metric_counter("http.requests", unit="1")
request_counter.add(1, {"endpoint": "/api/users", "method": "GET"})
request_counter = logfire.metric_counter("http.requests", unit="1")
request_counter.add(1, {"endpoint": "/api/users", "method": "GET"})
Gauge - current value
仪表盘 - 当前值
temperature = logfire.metric_gauge("temperature", unit="°C")
temperature.set(23.5)
temperature = logfire.metric_gauge("temperature", unit="°C")
temperature.set(23.5)
Histogram - distribution of values
直方图 - 值的分布
latency = logfire.metric_histogram("request.duration", unit="ms")
latency.record(45.2, {"endpoint": "/api/data"})
undefinedlatency = logfire.metric_histogram("request.duration", unit="ms")
latency.record(45.2, {"endpoint": "/api/data"})
undefined6. LLM Observability
6. 大语言模型(LLM)可观测性
python
import logfire
from pydantic_ai import Agent
logfire.configure()
logfire.instrument_pydantic_ai() # Traces all agent interactions
agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")
result = agent.run_sync("Hello!")python
import logfire
from pydantic_ai import Agent
logfire.configure()
logfire.instrument_pydantic_ai() # 追踪所有Agent交互
agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")
result = agent.run_sync("Hello!")7. Suppress Noisy Instrumentation
7. 抑制冗余插桩
python
undefinedpython
undefinedSuppress entire scope (e.g., noisy library)
抑制整个范围(例如,冗余的库)
logfire.suppress_scopes("google.cloud.bigquery.opentelemetry_tracing")
logfire.suppress_scopes("google.cloud.bigquery.opentelemetry_tracing")
Suppress specific code block
抑制特定代码块
with logfire.suppress_instrumentation():
client.get("https://internal-healthcheck.local") # Not traced
undefinedwith logfire.suppress_instrumentation():
client.get("https://internal-healthcheck.local") # 不被追踪
undefined8. Sensitive Data Scrubbing
8. 敏感数据清理
python
import logfirepython
import logfireAdd custom patterns to scrub
添加自定义清理规则
logfire.configure(
scrubbing=logfire.ScrubbingOptions(
extra_patterns=["api_key", "secret", "token"]
)
)
logfire.configure(
scrubbing=logfire.ScrubbingOptions(
extra_patterns=["api_key", "secret", "token"]
)
)
Custom callback for fine-grained control
自定义回调实现细粒度控制
def scrubbing_callback(match: logfire.ScrubMatch):
if match.path == ("attributes", "safe_field"):
return match.value # Don't scrub this field
return None # Use default scrubbing
logfire.configure(
scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)
undefineddef scrubbing_callback(match: logfire.ScrubMatch):
if match.path == ("attributes", "safe_field"):
return match.value # 不清理该字段
return None # 使用默认清理规则
logfire.configure(
scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)
undefined9. Sampling for High-Traffic Services
9. 高流量服务的采样配置
python
import logfirepython
import logfireSample 50% of traces
采样50%的追踪数据
logfire.configure(sampling=logfire.SamplingOptions(head=0.5))
logfire.configure(sampling=logfire.SamplingOptions(head=0.5))
Disable metrics to reduce volume
禁用指标以减少数据量
logfire.configure(metrics=False)
undefinedlogfire.configure(metrics=False)
undefined10. Testing
10. 测试
python
import logfire
from logfire.testing import CaptureLogfire
def test_user_creation(capfire: CaptureLogfire):
create_user("Alice", "alice@example.com")
spans = capfire.exporter.exported_spans
assert len(spans) >= 1
assert spans[0].attributes["user_name"] == "Alice"
capfire.exporter.clear() # Clean up for next testpython
import logfire
from logfire.testing import CaptureLogfire
def test_user_creation(capfire: CaptureLogfire):
create_user("Alice", "alice@example.com")
spans = capfire.exporter.exported_spans
assert len(spans) >= 1
assert spans[0].attributes["user_name"] == "Alice"
capfire.exporter.clear() # 为下一个测试清理数据Available Integrations
可用集成
| Category | Integration | Method |
|---|---|---|
| Web | FastAPI | |
| Starlette | | |
| Django | | |
| Flask | | |
| AIOHTTP Server | | |
| ASGI | | |
| WSGI | | |
| HTTP | HTTPX | |
| Requests | | |
| AIOHTTP Client | | |
| Database | SQLAlchemy | |
| Asyncpg | | |
| Psycopg | | |
| Redis | | |
| PyMongo | | |
| LLM | Pydantic AI | |
| OpenAI | | |
| Anthropic | | |
| MCP | | |
| Tasks | Celery | |
| AWS Lambda | | |
| Logging | Standard logging | |
| Structlog | | |
| Loguru | | |
| ||
| Other | Pydantic | |
| System Metrics | |
| 分类 | 集成项 | 方法 |
|---|---|---|
| Web | FastAPI | |
| Starlette | | |
| Django | | |
| Flask | | |
| AIOHTTP Server | | |
| ASGI | | |
| WSGI | | |
| HTTP | HTTPX | |
| Requests | | |
| AIOHTTP Client | | |
| 数据库 | SQLAlchemy | |
| Asyncpg | | |
| Psycopg | | |
| Redis | | |
| PyMongo | | |
| LLM | Pydantic AI | |
| OpenAI | | |
| Anthropic | | |
| MCP | | |
| 任务 | Celery | |
| AWS Lambda | | |
| 日志 | 标准日志 | |
| Structlog | | |
| Loguru | | |
| ||
| 其他 | Pydantic | |
| 系统指标 | |
Common Pitfalls
常见陷阱
| Issue | Symptom | Fix |
|---|---|---|
| Missing service name | Spans hard to find in UI | Set |
| Late instrumentation | No spans captured | Call |
| High-cardinality attrs | Storage explosion | Use IDs, not full payloads as attributes |
| Console noise | Logs pollute stdout | Set |
| 问题 | 症状 | 解决方法 |
|---|---|---|
| 缺少服务名称 | 在UI中难以找到跨度 | 在 |
| 插桩过晚 | 未捕获到跨度 | 在创建客户端前调用 |
| 高基数属性 | 存储资源耗尽 | 使用ID而非完整负载作为属性 |
| 控制台冗余日志 | 标准输出被日志污染 | 生产环境中设置 |
References
参考资料
- Configuration Options - All parameters
configure() - Integrations Guide - Framework-specific setup
- Metrics Guide - Counter, gauge, histogram, system metrics
- Advanced Patterns - Sampling, scrubbing, suppression, testing
- Pitfalls & Troubleshooting - Common issues and solutions
- Official Docs
- 配置选项 - 所有参数
configure() - 集成指南 - 框架专属设置
- 指标指南 - 计数器、仪表盘、直方图、系统指标
- 高级模式 - 采样、清理、抑制、测试
- 官方文档