### Install Observabilipy with Package Manager

Source: https://github.com/philhem/observabilipy/blob/main/README.md

Installs the observabilipy package and its dependencies using the `uv` package manager. Optional extra dependencies for framework adapters like FastAPI and Django can be installed using the `--extra` flag.

```bash
git clone https://github.com/PhilHem/observabilipy.git
cd observabilipy
uv sync
```

```bash
uv sync --extra fastapi
uv sync --extra django
```

--------------------------------

### Install and Run Project Commands with uv

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

These commands use the 'uv' package manager and runner for installing dependencies, running tests, performing type checking, and linting the project. Ensure 'uv' is installed and configured for the project.

```bash
# Install dependencies
uv sync

# Run all tests
uv run pytest

# Run single test
uv run pytest tests/unit/test_models.py::test_log_entry -v

# Type checking
uv run mypy observability/

# Linting
uv run ruff check observability/
uv run ruff format observability/
```

--------------------------------

### Installing and Running Pre-commit Hooks

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

Demonstrates how to install and manually run pre-commit hooks for the project. These hooks automate linting, formatting checks, type checking, and testing locally, ensuring code quality and consistency before pushing changes.

```bash
uv run pre-commit install

# To run all hooks manually on all files:
uv run pre-commit run --all-files
```

--------------------------------

### FastAPI with In-Memory Storage for Development

Source: https://context7.com/philhem/observabilipy/llms.txt

A minimal FastAPI application using in-memory storage for logs and metrics. This setup is ideal for development and testing purposes as it requires no external database setup and data is ephemeral. The application includes a root endpoint that logs an access event and records a metric sample.

```python
import time
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.in_memory import (
    InMemoryLogStorage,
    InMemoryMetricsStorage
)
from observability.core.models import LogEntry, MetricSample

# Initialize
log_storage = InMemoryLogStorage()
metrics_storage = InMemoryMetricsStorage()
app = FastAPI()
app.include_router(create_observability_router(log_storage, metrics_storage))

@app.get("/")
async def root():
    await log_storage.write(
        LogEntry(time.time(), "INFO", "Root accessed", {{"path": "/"}})
    )
    await metrics_storage.write(
        MetricSample("requests_total", time.time(), 1.0, {{"endpoint": "/"}})
    )
    return {{"message": "OK"}}

# Run: uvicorn main:app --reload
```

--------------------------------

### Manual Runtime Lifecycle Management

Source: https://context7.com/philhem/observabilipy/llms.txt

Demonstrates manual control over the EmbeddedRuntime lifecycle, including starting, running a single iteration, and stopping the runtime. This is useful for scenarios where automatic integration is not desired. It ensures that cleanup operations are triggered correctly.

```python
async def manual_lifecycle():
    await runtime.start()
    try:
        # Do work
        await log_storage.write(LogEntry(time.time(), "INFO", "Working", {{}}))

        # Trigger immediate cleanup
        await runtime.run_once()

    finally:
        await runtime.stop()

asyncio.run(main())
```

--------------------------------

### GET /metrics

Source: https://context7.com/philhem/observabilipy/llms.txt

Returns all metrics in Prometheus text exposition format. This endpoint is useful for scraping by Prometheus or compatible systems.

```APIDOC
## GET /metrics

### Description
Returns all metrics in Prometheus text exposition format. This endpoint is useful for scraping by Prometheus or compatible systems.

### Method
GET

### Endpoint
/metrics

### Parameters

#### Query Parameters
- **since** (number) - Optional - Fetch metrics updated since this Unix timestamp.

### Request Example
```bash
curl http://localhost:8000/metrics
```

### Response
#### Success Response (200)
- **Content-Type**: text/plain; version=0.0.4
- **Body**: Prometheus metrics in text exposition format.

#### Response Example
```text
# HELP http_requests_total Total number of HTTP requests.
# TYPE http_requests_total counter
http_requests_total{method="GET",path="/api/users",status="200"} 1247.0 1701234567890
http_requests_total{method="POST",path="/api/users",status="201"} 89.0 1701234567891
# HELP http_request_duration_seconds Duration of HTTP requests in seconds.
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{le="0.05",method="GET",path="/api/users"} 100.0 1701234567892
http_request_duration_seconds_count 1247.0 1701234567892
http_request_duration_seconds_sum 45.123 1701234567892
# HELP process_memory_bytes Current process memory usage in bytes.
# TYPE process_memory_bytes gauge
process_memory_bytes 52428800.0 1701234567893
```

### Configuration Example for Prometheus
```yaml
prometheus.yml:
scrape_configs:
  - job_name: 'myapp'
    static_configs:
      - targets: ['localhost:8000']
    metrics_path: '/metrics'
    scrape_interval: 15s
```
```

--------------------------------

### Access Prometheus Metrics Endpoint

Source: https://context7.com/philhem/observabilipy/llms.txt

Provides instructions on how to scrape Prometheus metrics from the application. The `/metrics` endpoint exposes metrics in Prometheus text exposition format, suitable for scraping by Prometheus servers or compatible systems. The example shows how to use `curl` and provides a configuration snippet for `prometheus.yml`.

```bash
# Scrape metrics endpoint
curl http://localhost:8000/metrics

# Example response (text/plain; version=0.0.4):
# http_requests_total{method="GET",path="/api/users",status="200"} 1247.0 1701234567890
# http_requests_total{method="POST",path="/api/users",status="201"} 89.0 1701234567891
# http_request_duration_seconds{method="GET",path="/api/users"} 0.045 1701234567892
# process_memory_bytes 52428800.0 1701234567893

# Configure Prometheus to scrape:
# prometheus.yml:
# scrape_configs:
#   - job_name: 'myapp'
#     static_configs:
#       - targets: ['localhost:8000']
#     metrics_path: '/metrics'
#     scrape_interval: 15s
```

--------------------------------

### FastAPI Lifespan Integration for EmbeddedRuntime

Source: https://context7.com/philhem/observabilipy/llms.txt

Integrates Observabilipy's EmbeddedRuntime with FastAPI's lifespan events. This ensures the runtime starts automatically when the FastAPI application starts and stops gracefully when it shuts down. It uses SQLite for persistent storage of logs and metrics.

```python
import time
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.sqlite import SQLiteLogStorage, SQLiteMetricsStorage
from observability.core.models import RetentionPolicy
from observability.runtime.embedded import EmbeddedRuntime

# Setup storage and runtime
log_storage = SQLiteLogStorage("logs.db")
metrics_storage = SQLiteMetricsStorage("metrics.db")

runtime = EmbeddedRuntime(
    log_storage=log_storage,
    log_retention=RetentionPolicy(max_age_seconds=3600),
    metrics_storage=metrics_storage,
    metrics_retention=RetentionPolicy(max_age_seconds=3600),
    cleanup_interval_seconds=60
)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    """Manage runtime lifecycle."""
    await runtime.start()  # Start cleanup on app startup
    yield
    await runtime.stop()   # Stop cleanup on app shutdown

# Create app with lifespan
app = FastAPI(title="Production App", lifespan=lifespan)
app.include_router(create_observability_router(log_storage, metrics_storage))

# Run with: uvicorn main:app
# Background cleanup runs every 60 seconds, deleting data older than 1 hour
```

--------------------------------

### GET /logs

Source: https://context7.com/philhem/observabilipy/llms.txt

Returns structured logs in newline-delimited JSON format. This endpoint is suitable for ingestion by log collectors like Grafana Alloy.

```APIDOC
## GET /logs

### Description
Returns structured logs in newline-delimited JSON format. This endpoint is suitable for ingestion by log collectors like Grafana Alloy or for incremental synchronization.

### Method
GET

### Endpoint
/logs

### Parameters

#### Query Parameters
- **since** (number) - Optional - Fetch log entries with a timestamp strictly greater than this value.

### Request Example
```bash
# Fetch all logs
curl http://localhost:8000/logs

# Fetch logs since a specific timestamp (e.g., for incremental sync)
curl "http://localhost:8000/logs?since=1701234567.5"
```

### Response
#### Success Response (200)
- **Content-Type**: application/x-ndjson
- **Body**: Newline-delimited JSON objects, each representing a log entry.

#### Response Example
```json
{"timestamp":1701234567.123,"level":"INFO","message":"User login","attributes":{"user_id":123}}
{"timestamp":1701234568.456,"level":"ERROR","message":"DB connection lost","attributes":{"error":"timeout"}}
```

### Configuration Example for Grafana Alloy
```yaml
alloy config.alloy:
loki.source.api "logs" {
  http {
    url = "http://localhost:8000/logs"
    poll_interval = "10s"
  }
  forward_to = [loki.write.default.receiver]
}
```
```

--------------------------------

### FastAPI with Persistent SQLite Storage

Source: https://context7.com/philhem/observabilipy/llms.txt

This snippet demonstrates a FastAPI application configured with persistent SQLite storage for both logs and metrics. It includes an example API endpoint that writes log entries and metric samples to these SQLite databases. Data persists across application restarts, and the databases are named 'logs.db' and 'metrics.db'.

```python
import time
from fastapi import FastAPI, HTTPException
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.sqlite import SQLiteLogStorage, SQLiteMetricsStorage
from observability.core.models import LogEntry, MetricSample

log_storage = SQLiteLogStorage("logs.db")
metrics_storage = SQLiteMetricsStorage("metrics.db")
app = FastAPI()
app.include_router(create_observability_router(log_storage, metrics_storage))

@app.get("/api/items/{item_id}")
async def get_item(item_id: int):
    try:
        # Simulate business logic
        if item_id < 1:
            raise ValueError("Invalid item ID")

        await log_storage.write(
            LogEntry(time.time(), "INFO", "Item retrieved", {"item_id": item_id})
        )
        await metrics_storage.write(
            MetricSample("api_calls_total", time.time(), 1.0,
                        {"endpoint": "/api/items", "status": "success"})
        )
        return {"item_id": item_id, "name": f"Item {item_id}"}

    except ValueError as e:
        await log_storage.write(
            LogEntry(time.time(), "ERROR", str(e), {"item_id": item_id})
        )
        await metrics_storage.write(
            MetricSample("api_errors_total", time.time(), 1.0,
                        {"endpoint": "/api/items", "error": "validation"})
        )
        raise HTTPException(status_code=400, detail=str(e))

# Run: uvicorn main:app
# Data persists in logs.db and metrics.db files

```

--------------------------------

### Access NDJSON Logs Endpoint

Source: https://context7.com/philhem/observabilipy/llms.txt

Details how to fetch structured logs from the `/logs` endpoint in newline-delimited JSON (NDJSON) format. This format is compatible with log collectors like Grafana Alloy. Examples demonstrate fetching all logs, fetching logs since a specific timestamp for incremental synchronization, and a configuration snippet for Grafana Alloy.

```bash
# Fetch all logs
curl http://localhost:8000/logs

# Example response (application/x-ndjson):
# {"timestamp":1701234567.123,"level":"INFO","message":"User login","attributes":{"user_id":123}}
# {"timestamp":1701234568.456,"level":"ERROR","message":"DB connection lost","attributes":{"error":"timeout"}}

# Fetch logs since specific timestamp (incremental sync)
curl "http://localhost:8000/logs?since=1701234567.5"

# Only returns entries with timestamp > 1701234567.5
# {"timestamp":1701234568.456,"level":"ERROR","message":"DB connection lost","attributes":{"error":"timeout"}}

# Configure Grafana Alloy to collect:
# alloy config.alloy:
# loki.source.api "logs" {
#   http {
#     url = "http://localhost:8000/logs"
#     poll_interval = "10s"
#   }
#   forward_to = [loki.write.default.receiver]
# }
```

--------------------------------

### Record Metrics and Logs with Observabilipy

Source: https://github.com/philhem/observabilipy/blob/main/README.md

Demonstrates how to record log entries and metric samples using observabilipy's storage backends. It shows the structure of `LogEntry` and `MetricSample` objects, including timestamps, levels, messages, attributes, names, values, and labels. This functionality requires an initialized storage backend.

```python
import time
from observability.core.models import LogEntry, MetricSample

# Record a log entry
await log_storage.write(
    LogEntry(
        timestamp=time.time(),
        level="INFO",
        message="User logged in",
        attributes={"user_id": 123, "ip": "192.168.1.1"},
    )
)

# Record a metric sample
await metrics_storage.write(
    MetricSample(
        name="http_requests_total",
        timestamp=time.time(),
        value=1.0,
        labels={"method": "GET", "path": "/api/users"},
    )
)
```

--------------------------------

### Write Logs and Metrics

Source: https://context7.com/philhem/observabilipy/llms.txt

Demonstrates writing log entries and metric samples to storage. The storage automatically creates the necessary schema on the first write operation. Inputs include LogEntry and MetricSample objects with timestamps, levels, messages, attributes, names, values, and labels.

```python
await log_storage.write(
    LogEntry(
        timestamp=time.time(),
        level="ERROR",
        message="Payment processing failed",
        attributes={
            "transaction_id": "txn_12345",
            "amount": 99.99,
            "currency": "USD",
            "error": "insufficient_funds"
        }
    )
)

await metrics_storage.write(
    MetricSample(
        name="payment_failures_total",
        timestamp=time.time(),
        value=1.0,
        labels={"reason": "insufficient_funds", "gateway": "stripe"}
    )
)
```

--------------------------------

### Use InMemoryLogStorage and InMemoryMetricsStorage in Python

Source: https://context7.com/philhem/observabilipy/llms.txt

Demonstrates the usage of in-memory storage adapters for logs and metrics. These are suitable for development and testing environments due to their speed and lack of persistence. Supports writing, reading, counting, and deleting log entries.

```Python
import time
from observability.adapters.storage.in_memory import (
    InMemoryLogStorage,
    InMemoryMetricsStorage
)
from observability.core.models import LogEntry, MetricSample

# Initialize storage
log_storage = InMemoryLogStorage()
metrics_storage = InMemoryMetricsStorage()

# Write log entries
await log_storage.write(
    LogEntry(
        timestamp=time.time(),
        level="INFO",
        message="Application started",
        attributes={"version": "1.0.0"}
    )
)

# Write metric samples
await metrics_storage.write(
    MetricSample(
        name="app_start_count",
        timestamp=time.time(),
        value=1.0,
        labels={"environment": "production"}
    )
)

# Read logs since timestamp
async for entry in log_storage.read(since=time.time() - 3600):
    print(f"{entry.level}: {entry.message}")

# Count total entries
total_logs = await log_storage.count()
print(f"Total logs: {total_logs}")

# Delete old entries
deleted = await log_storage.delete_before(time.time() - 86400)
print(f"Deleted {deleted} old logs")
```

--------------------------------

### Manage Project Dependencies with uv

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

Instructions for managing project dependencies using 'uv' commands. It emphasizes never editing 'pyproject.toml' directly and using 'uv add', 'uv remove', and 'uv lock' for dependency updates. This ensures consistency between the lock file and project requirements.

```bash
uv add requests              # Add dependency
uv add --dev pytest          # Add dev dependency
uv remove requests           # Remove dependency
uv lock                      # Update lock file
```

--------------------------------

### Create LogEntry Objects in Python

Source: https://context7.com/philhem/observabilipy/llms.txt

Demonstrates how to create structured log entries using the LogEntry dataclass. Supports standard fields like timestamp, level, message, and custom attributes for additional context. Useful for logging application events and errors.

```Python
import time
from observability.core.models import LogEntry

# Create a basic log entry
log = LogEntry(
    timestamp=time.time(),
    level="INFO",
    message="User authentication successful",
    attributes={"user_id": 123, "ip": "192.168.1.100"}
)

# Create an error log with additional context
error_log = LogEntry(
    timestamp=time.time(),
    level="ERROR",
    message="Database connection failed",
    attributes={
        "error_code": "CONN_TIMEOUT",
        "retry_count": 3,
        "database": "users_db"
    }
)
```

--------------------------------

### Configure Django URL patterns for observability

Source: https://context7.com/philhem/observabilipy/llms.txt

Defines URL patterns for a Django application, including a user detail view and observability-related endpoints. It utilizes `create_observability_urlpatterns` to integrate logging and metrics endpoints. The ASGI application is obtained using `get_asgi_application` and can be run with `uvicorn`.

```python
urlpatterns = [
    path("users/<int:user_id>/", user_detail),
    *create_observability_urlpatterns(log_storage, metrics_storage)
]

from django.core.asgi import get_asgi_application
application = get_asgi_application()

# Run with: uvicorn myapp:application
# Access:
#   http://localhost:8000/metrics
#   http://localhost:8000/logs?since=0
```

--------------------------------

### Use SQLiteLogStorage and SQLiteMetricsStorage in Python

Source: https://context7.com/philhem/observabilipy/llms.txt

Details the implementation of persistent storage adapters using SQLite for logs and metrics. These adapters are designed for production, offering concurrent access and atomic writes via SQLite's WAL mode. Initialization requires specifying database file paths.

```Python
import time
from observability.adapters.storage.sqlite import (
    SQLiteLogStorage,
    SQLiteMetricsStorage
)
from observability.core.models import LogEntry, MetricSample

# Initialize SQLite storage with file paths
log_storage = SQLiteLogStorage("logs.db")
metrics_storage = SQLiteMetricsStorage("metrics.db")

```

--------------------------------

### FastAPI with EmbeddedRuntime Lifespan

Source: https://context7.com/philhem/observabilipy/llms.txt

Demonstrates integrating EmbeddedRuntime with FastAPI's lifespan events for automatic startup and shutdown management of the observability runtime.

```APIDOC
```python
import time
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.sqlite import SQLiteLogStorage, SQLiteMetricsStorage
from observability.core.models import RetentionPolicy
from observability.runtime.embedded import EmbeddedRuntime

# Setup storage and runtime
log_storage = SQLiteLogStorage("logs.db")
metrics_storage = SQLiteMetricsStorage("metrics.db")

runtim<ctrl62>untime.stop()   # Stop cleanup on app shutdown

# Create app with lifespan
app = FastAPI(title="Production App", lifespan=lifespan)
app.include_router(create_observability_router(log_storage, metrics_storage))

# Run with: uvicorn main:app
# Background cleanup runs every 60 seconds, deleting data older than 1 hour
```
```

--------------------------------

### Create MetricSample Objects in Python

Source: https://context7.com/philhem/observabilipy/llms.txt

Illustrates the creation of single metric measurements using the MetricSample dataclass. Supports metric name, timestamp, value, and labels for dimensional data. Suitable for recording counters, gauges, and other time-series metrics.

```Python
import time
from observability.core.models import MetricSample

# Record an HTTP request counter metric
request_metric = MetricSample(
    name="http_requests_total",
    timestamp=time.time(),
    value=1.0,
    labels={"method": "GET", "path": "/api/users", "status": "200"}
)

# Record a response time gauge metric
latency_metric = MetricSample(
    name="http_request_duration_seconds",
    timestamp=time.time(),
    value=0.042,
    labels={"method": "POST", "path": "/api/login"}
)

# Record a memory usage metric
memory_metric = MetricSample(
    name="process_memory_bytes",
    timestamp=time.time(),
    value=52428800.0,  # 50MB
    labels={}
)
```

--------------------------------

### Scrape All Metrics

Source: https://context7.com/philhem/observabilipy/llms.txt

Demonstrates how to scrape all stored metrics. This function iterates through all collected MetricSample objects.

```python
all_metrics = []
async for sample in metrics_storage.scrape():
    all_metrics.append(sample)
```

--------------------------------

### Log user detail view and metrics

Source: https://context7.com/philhem/observabilipy/llms.txt

Asynchronously logs user profile view events and metrics. It requires an HttpRequest object and a user ID, returning an HttpResponse. It uses `log_storage` and `metrics_storage` for writing log entries and metric samples respectively.

```python
async def user_detail(request: HttpRequest, user_id: int) -> HttpResponse:
    await log_storage.write(
        LogEntry(
            timestamp=time.time(),
            level="INFO",
            message="User profile viewed",
            attributes={"user_id": user_id, "path": request.path}
        )
    )

    await metrics_storage.write(
        MetricSample(
            name="profile_views_total",
            timestamp=time.time(),
            value=1.0,
            labels={"user_id": str(user_id)}
        )
    )

    return HttpResponse(f"User {user_id} profile")
```

--------------------------------

### FastAPI with In-Memory Storage

Source: https://context7.com/philhem/observabilipy/llms.txt

A minimal FastAPI application using in-memory storage for logs and metrics, suitable for development and testing environments.

```APIDOC
```python
import time
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.in_memory import (
    InMemoryLogStorage,
    InMemoryMetricsStorage
)
from observability.core.models import LogEntry, MetricSample

# Initialize
log_storage = InMemoryLogStorage()
metrics_storage = InMemoryMetricsStorage()
app = FastAPI()
app.include_router(create_observability_router(log_storage, metrics_storage))

@app.get("/")
async def root():
    await log_storage.write(
        LogEntry(time.time(), "INFO", "Root accessed", {"path": "/"})
    )
    await metrics_storage.write(
        MetricSample("requests_total", time.time(), 1.0, {"endpoint": "/"})
    )
    return {"message": "OK"}

# Run: uvicorn main:app --reload
```
```

--------------------------------

### FastAPI - Create observability router

Source: https://context7.com/philhem/observabilipy/llms.txt

Creates a FastAPI router with `/metrics` and `/logs` endpoints, designed for Prometheus and Grafana Alloy. It integrates provided log and metrics storage instances. The router can be mounted at the root or with a specified prefix. Application endpoints can then use the storage to record logs and metrics.

```python
import time
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.in_memory import (
    InMemoryLogStorage,
    InMemoryMetricsStorage
)
from observability.core.models import LogEntry, MetricSample

# Initialize storage
log_storage = InMemoryLogStorage()
metrics_storage = InMemoryMetricsStorage()

# Create FastAPI app
app = FastAPI(title="My API")

# Mount observability endpoints at root level
app.include_router(create_observability_router(log_storage, metrics_storage))

# Or mount with prefix
app.include_router(
    create_observability_router(log_storage, metrics_storage),
    prefix="/observability"
)

# Application endpoints record logs and metrics
@app.post("/api/users")
async def create_user(username: str):
    start_time = time.time()

    # Record log
    await log_storage.write(
        LogEntry(
            timestamp=time.time(),
            level="INFO",
            message="User created",
            attributes={"username": username}
        )
    )

    # Record metrics
    await metrics_storage.write(
        MetricSample(
            name="user_registrations_total",
            timestamp=time.time(),
            value=1.0,
            labels={"source": "api"}
        )
    )

    duration = time.time() - start_time
    await metrics_storage.write(
        MetricSample(
            name="api_duration_seconds",
            timestamp=time.time(),
            value=duration,
            labels={"endpoint": "/api/users", "method": "POST"}
        )
    )

    return {"username": username}

# Run with: uvicorn main:app --reload
# Access:
#   http://localhost:8000/metrics - Prometheus text format
#   http://localhost:8000/logs - All logs as NDJSON
#   http://localhost:8000/logs?since=1234567890 - Filtered logs
```

--------------------------------

### Configure RetentionPolicy in Python

Source: https://context7.com/philhem/observabilipy/llms.txt

Shows how to define data retention rules using the RetentionPolicy class. Supports age-based (max_age_seconds) and count-based (max_count) constraints, which can be used independently or combined. Allows for disabling limits entirely.

```Python
from observability.core.models import RetentionPolicy

# Keep only last 1 hour of data
age_based = RetentionPolicy(max_age_seconds=3600)

# Keep only last 1000 entries
count_based = RetentionPolicy(max_count=1000)

# Keep last 1 hour OR 1000 entries (whichever hits first)
combined = RetentionPolicy(max_age_seconds=3600, max_count=1000)

# No limits (retain everything)
no_limits = RetentionPolicy()
```

--------------------------------

### Define Metrics Storage Port in Python

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

Defines the 'MetricsStoragePort' protocol, specifying the interface for metric storage adapters. It includes methods for writing metric samples and scraping metrics, maintaining a consistent API for metric storage.

```python
from typing import Protocol, Iterable
from .models import MetricSample

class MetricsStoragePort(Protocol):
    def write(self, sample: MetricSample) -> None: ...
    def scrape(self) -> Iterable[MetricSample]: ...
```

--------------------------------

### Update Lock File After Version Bump with uv

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

This command is crucial after bumping the version number in 'pyproject.toml'. Running 'uv lock' updates the project's lock file to reflect the new version, ensuring reproducible builds.

```bash
# After editing version in pyproject.toml
uv lock
```

--------------------------------

### Integrate Observabilipy with FastAPI

Source: https://github.com/philhem/observabilipy/blob/main/README.md

Integrates observabilipy's observability features into a FastAPI application. It creates an observability router using in-memory storage for logs and metrics, and includes this router in the FastAPI app. This allows for accessible `/metrics` and `/logs` endpoints.

```python
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.in_memory import (
    InMemoryLogStorage,
    InMemoryMetricsStorage,
)

app = FastAPI()
log_storage = InMemoryLogStorage()
metrics_storage = InMemoryMetricsStorage()

app.include_router(create_observability_router(log_storage, metrics_storage))
```

--------------------------------

### Manage runtime with embedded cleanup

Source: https://context7.com/philhem/observabilipy/llms.txt

Orchestrates the lifecycle and retention of logs and metrics using an `EmbeddedRuntime`. It supports configurable retention policies (max age and count) and an automatic cleanup interval. Dependencies include `asyncio`, `time`, `EmbeddedRuntime`, SQLite storage adapters, `LogEntry`, `MetricSample`, and `RetentionPolicy`. The runtime can be used as an asynchronous context manager.

```python
import asyncio
import time
from observability.runtime.embedded import EmbeddedRuntime
from observability.adapters.storage.sqlite import SQLiteLogStorage, SQLiteMetricsStorage
from observability.core.models import LogEntry, MetricSample, RetentionPolicy

# Create storage
log_storage = SQLiteLogStorage("app.db")
metrics_storage = SQLiteMetricsStorage("metrics.db")

# Define retention policies
log_retention = RetentionPolicy(
    max_age_seconds=86400,  # Keep 24 hours
    max_count=10000         # Or max 10k entries
)
metrics_retention = RetentionPolicy(
    max_age_seconds=604800  # Keep 7 days
)

# Create runtime with automatic cleanup
runtime = EmbeddedRuntime(
    log_storage=log_storage,
    log_retention=log_retention,
    metrics_storage=metrics_storage,
    metrics_retention=metrics_retention,
    cleanup_interval_seconds=300  # Run cleanup every 5 minutes
)

# Use as context manager (recommended)
async def main():
    async with runtime:
        # Runtime starts, cleanup runs in background

        # Write data
        await log_storage.write(
            LogEntry(time.time(), "INFO", "App running", {})
        )

        # Old data automatically cleaned up every 5 minutes
        await asyncio.sleep(10)

    # Runtime stops on exit
```

--------------------------------

### Django - Create observability URL patterns

Source: https://context7.com/philhem/observabilipy/llms.txt

Factory function that creates Django URL patterns with asynchronous views for `/metrics` and `/logs` endpoints. It utilizes specified log and metrics storage, such as SQLite storage, to handle observability data within a Django application. Configuration for Django settings is included.

```python
import time
import django
from django.conf import settings
from django.http import HttpRequest, HttpResponse

# Configure Django
if not settings.configured:
    settings.configure(
        DEBUG=True,
        ROOT_URLCONF=__name__,
        ALLOWED_HOSTS=["*"],
        SECRET_KEY="secret"
    )
    django.setup()

from django.urls import path
from observability.adapters.frameworks.django import create_observability_urlpatterns
from observability.adapters.storage.sqlite import SQLiteLogStorage, SQLiteMetricsStorage
from observability.core.models import LogEntry, MetricSample

# Initialize persistent storage
log_storage = SQLiteLogStorage("app_logs.db")
metrics_storage = SQLiteMetricsStorage("app_metrics.db")


```

--------------------------------

### Query Logs with Filtering

Source: https://context7.com/philhem/observabilipy/llms.txt

Shows how to query log entries with a time-based filter. It iterates through logs stored since a specific timestamp and filters them based on the log level. The output is a list of recent error entries.

```python
recent_errors = []
async for entry in log_storage.read(since=time.time() - 3600):
    if entry.level == "ERROR":
        recent_errors.append(entry)

print(f"Found {len(recent_errors)} errors in last hour")
```

--------------------------------

### Define Log Storage Port in Python

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

Defines the 'LogStoragePort' protocol, outlining the interface for log storage adapters. It specifies methods for writing and reading log entries, ensuring a consistent contract for different storage implementations.

```python
from typing import Protocol, Iterable
from .models import LogEntry

class LogStoragePort(Protocol):
    def write(self, entry: LogEntry) -> None: ...
    def read(self, since: float = 0) -> Iterable[LogEntry]: ...
```

--------------------------------

### FastAPI with Automatic Data Retention

Source: https://context7.com/philhem/observabilipy/llms.txt

This snippet configures a FastAPI application with persistent SQLite storage and an embedded runtime for automatic data retention. It sets specific retention policies for logs (7 days) and metrics (30 days), with cleanup operations running hourly. This ensures efficient management of storage space in production environments.

```python
import time
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from observability.adapters.frameworks.fastapi import create_observability_router
from observability.adapters.storage.sqlite import SQLiteLogStorage, SQLiteMetricsStorage
from observability.core.models import LogEntry, MetricSample, RetentionPolicy
from observability.runtime.embedded import EmbeddedRuntime

# Storage with persistence
log_storage = SQLiteLogStorage("production_logs.db")
metrics_storage = SQLiteMetricsStorage("production_metrics.db")

# Retention: keep 7 days of logs, 30 days of metrics
runtime = EmbeddedRuntime(
    log_storage=log_storage,
    log_retention=RetentionPolicy(max_age_seconds=604800),  # 7 days
    metrics_storage=metrics_storage,
    metrics_retention=RetentionPolicy(max_age_seconds=2592000),  # 30 days
    cleanup_interval_seconds=3600  # Run cleanup hourly
)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    await runtime.start()
    yield
    await runtime.stop()

app = FastAPI(lifespan=lifespan)
app.include_router(create_observability_router(log_storage, metrics_storage))

@app.post("/api/process")
async def process_task(task_id: str):
    start = time.time()

    await log_storage.write(
        LogEntry(time.time(), "INFO", "Task started", {"task_id": task_id})
    )

    # Simulate processing
    import asyncio
    await asyncio.sleep(0.1)

    duration = time.time() - start
    await log_storage.write(
        LogEntry(time.time(), "INFO", "Task completed",
                {"task_id": task_id, "duration": duration})
    )
    await metrics_storage.write(
        MetricSample("task_duration_seconds", time.time(), duration,
                    {"task_id": task_id})
    )
    await metrics_storage.write(
        MetricSample("tasks_completed_total", time.time(), 1.0, {})
    )

    return {"task_id": task_id, "duration": duration}

# Run: uvicorn main:app
# Automatic cleanup removes logs older than 7 days, metrics older than 30 days

```

--------------------------------

### RingBufferLogStorage/RingBufferMetricsStorage - Fixed-size circular buffer

Source: https://context7.com/philhem/observabilipy/llms.txt

Implements fixed-size circular buffers for logs and metrics, ensuring predictable memory usage by automatically evicting the oldest entries when the buffer is full. This is useful in constrained environments. The `max_size` parameter defines the buffer capacity.

```python
from observability.adapters.storage.ring_buffer import (
    RingBufferLogStorage,
    RingBufferMetricsStorage
)
from observability.core.models import LogEntry, MetricSample
import time

# Create ring buffers with size limits
log_storage = RingBufferLogStorage(max_size=1000)
metrics_storage = RingBufferMetricsStorage(max_size=5000)

# Write entries - oldest automatically evicted when full
for i in range(1500):
    await log_storage.write(
        LogEntry(
            timestamp=time.time(),
            level="DEBUG",
            message=f"Event {i}",
            attributes={"sequence": i}
        )
    )

# Only last 1000 entries retained
count = await log_storage.count()
print(f"Buffer contains {count} entries (max: 1000)")

# Buffer maintains chronological order
async for entry in log_storage.read():
    print(f"Sequence: {entry.attributes['sequence']}")
```

--------------------------------

### Pytest Marks for Architectural Components

Source: https://github.com/philhem/observabilipy/blob/main/CLAUDE.md

Defines pytest marks used to categorize tests by architectural component. These marks are crucial for organizing tests and running specific subsets during development and in CI pipelines, facilitating failure isolation.

```python
import pytest

@pytest.mark.core        # Core models, ports, services
@pytest.mark.encoding    # NDJSON, Prometheus encoders
@pytest.mark.storage     # Storage adapters (sqlite, in_memory, ring_buffer)
@pytest.mark.fastapi     # FastAPI framework adapter
@pytest.mark.django      # Django framework adapter
@pytest.mark.e2e         # End-to-end tests

# Example usage:
# pytest -m core
# pytest -m "storage and not sqlite"
# pytest -m fastapi
```

--------------------------------

### Encode metrics to Prometheus format

Source: https://context7.com/philhem/observabilipy/llms.txt

Encodes MetricSample objects into the Prometheus text exposition format. This function handles label escaping and timestamp conversion for compatibility with Prometheus. It requires `encode_metrics` from `observability.core.encoding.prometheus`, `InMemoryMetricsStorage`, `MetricSample`, and `time`. The output includes metric names, labels, values, and timestamps.

```python
from observability.core.encoding.prometheus import encode_metrics
from observability.adapters.storage.in_memory import InMemoryMetricsStorage
from observability.core.models import MetricSample
import time

# Create storage with sample metrics
storage = InMemoryMetricsStorage()
await storage.write(
    MetricSample("http_requests_total", time.time(), 127.0, {"method": "GET", "status": "200"})
)
await storage.write(
    MetricSample("http_requests_total", time.time(), 23.0, {"method": "POST", "status": "201"})
)
await storage.write(
    MetricSample("memory_usage_bytes", time.time(), 52428800.0, {})
)

# Encode to Prometheus format
prom_output = await encode_metrics(storage.scrape())
print(prom_output)

# Labels are sorted and values are escaped
await storage.write(
    MetricSample("test", time.time(), 1.0, {"label": 'value with "quotes"\nand newlines'})
)
output = await encode_metrics(storage.scrape())
assert 'label="value with \"quotes\"\nand newlines"' in output
```

--------------------------------

### Encode logs to NDJSON format

Source: https://context7.com/philhem/observabilipy/llms.txt

Encodes a list of LogEntry objects into newline-delimited JSON (NDJSON) format. This format is suitable for log aggregators like Grafana Alloy. It can handle empty storage and filter logs by timestamp. Dependencies include `encode_logs` from `observability.core.encoding.ndjson`, `InMemoryLogStorage`, `LogEntry`, and `time`.

```python
from observability.core.encoding.ndjson import encode_logs
from observability.adapters.storage.in_memory import InMemoryLogStorage
from observability.core.models import LogEntry
import time

# Create storage with sample logs
storage = InMemoryLogStorage()
await storage.write(LogEntry(time.time(), "INFO", "App started", {"version": "1.0"}))
await storage.write(LogEntry(time.time(), "ERROR", "DB error", {"code": 500}))

# Encode to NDJSON
ndjson_output = await encode_logs(storage.read())
print(ndjson_output)

# Can also encode filtered logs
recent_logs = await encode_logs(storage.read(since=time.time() - 3600))

# Empty result when no entries
empty_storage = InMemoryLogStorage()
empty_output = await encode_logs(empty_storage.read())
assert empty_output == ""
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.