Logs¶

The Logs module allows you to collect and send log entries to the Upassist Cloud API.

Features¶

Send log entries in batches
Structured log data with timestamps
Custom metadata support
Type-safe log entries with Pydantic

Examples¶

Basic Usage¶

Send log entries to Upassist Cloud

from upassist.entities.logs import Logs, LogItemSchema
from datetime import datetime

logs = Logs()
log_entries = [
    LogItemSchema(
        dt=datetime.utcnow(),
        host="server1",
        message="Application started",
        file="app.py",
        data={"version": "1.0.0"}
    ),
    LogItemSchema(
        dt=datetime.utcnow(),
        host="server1",
        message="Database connection established",
        file="db.py",
        data={"connection_id": "123"}
    )
]
response = logs.collect(log_entries)

Advanced Usage¶

Send logs with custom configuration

from upassist.entities.logs import Logs, LogItemSchema
from datetime import datetime
from upassist import AsyncAPIClient

# Create logs instance with custom API key
logs = Logs(api_key="your-api-key")

# Use async client for high-performance logging
logs = Logs(api_client_cls=AsyncAPIClient)

# Send logs with custom metadata
log_entries = [
    LogItemSchema(
        dt=datetime.utcnow(),
        host="prod-server-1",
        message="High CPU usage detected",
        file="monitor.py",
        data={
            "cpu_usage": 95.5,
            "memory_usage": 80.2,
            "process_count": 150,
            "environment": "production"
        }
    )
]
response = logs.collect(log_entries)

API Reference¶

`LogItemSchema` ¶

Bases: BaseModel

Schema for individual log items.

Attributes:

Name	Type	Description
`dt`	`datetime \| None`	Timestamp of the log entry
`host`	`str \| None`	Hostname or source of the log
`message`	`str \| None`	Log message content
`file`	`str \| None`	Source file or location
`data`	`dict[str, Any] \| None`	Additional structured data associated with the log

Source code in upassist/entities/logs/schemas.py

class LogItemSchema(BaseModel):
    """Schema for individual log items.

    Attributes:
        dt: Timestamp of the log entry
        host: Hostname or source of the log
        message: Log message content
        file: Source file or location
        data: Additional structured data associated with the log
    """

    dt: datetime | None = None
    host: str | None = None
    message: str | None = None
    file: str | None = None
    data: dict[str, Any] | None = None

`Logs` ¶

Bases: BaseEntity

Entity for managing log collection and submission.

This class provides functionality to collect and submit logs to the logging service.

Source code in upassist/entities/logs/logs.py

class Logs(BaseEntity):
    """Entity for managing log collection and submission.

    This class provides functionality to collect and submit logs to the logging service.
    """

    base_logs_api_url: str = "https://logs.upassist.cloud/collect"

    def collect(self, logs: list[LogItemSchema]) -> DetailResponse:
        """Submit a collection of logs to the logging service.

        Args:
            logs: List of log items to submit

        Returns:
            DetailResponse containing the submission result
        """
        response = self.api_client.post(
            self.base_logs_api_url,
            json=[log.model_dump(mode="json") for log in logs],
        )
        return DetailResponse.model_validate(response)

`collect(logs)` ¶

Submit a collection of logs to the logging service.

Parameters:

Name	Type	Description	Default
`logs`	`list[LogItemSchema]`	List of log items to submit	required

Returns:

Type	Description
`DetailResponse`	DetailResponse containing the submission result