Loki
the_data_packet.utils.loki
¶
Log upload utility for sending logs to Grafana Loki.
This module provides classes and functions for uploading structured logs from JSONL files to Grafana Loki for centralized log aggregation and analysis.
Classes:
| Name | Description |
|---|---|
LokiUploader |
Main class for uploading logs to Grafana Loki |
LogUploadError |
Custom exception for upload failures |
JsonEncoder |
JSON encoder with datetime support |
Functions:
| Name | Description |
|---|---|
upload_logs_to_loki |
Convenience function for one-time log uploads |
Example
from the_data_packet.utils.log_upload import LokiUploader
uploader = LokiUploader( ... url="https://logs-prod.grafana.net/loki/api/v1/push", ... user="your_user_id", ... api_key="your_api_key" ... )
uploader.upload_from_file("/path/to/logs.jsonl")
logger = logging.getLogger(__name__)
module-attribute
¶
LogUploadError
¶
Exception raised for log upload errors.
Attributes:
| Name | Type | Description |
|---|---|---|
message |
Explanation of the error |
|
response |
Optional HTTP response object if applicable |
message = message
instance-attribute
¶
response = response
instance-attribute
¶
__init__(message: str, response: Optional[requests.Response] = None)
¶
Initialize LogUploadError.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
Error message describing what went wrong |
required |
response
|
Optional[Response]
|
Optional HTTP response object for debugging |
None
|
JsonEncoder
¶
Custom JSON encoder for datetime objects and other Python types.
This encoder extends the default JSON encoder to handle datetime objects by converting them to ISO format strings.
default(o: Any) -> Any
¶
Encode special Python objects to JSON-serializable formats.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
obj
|
Object to encode |
required |
Returns:
| Type | Description |
|---|---|
Any
|
JSON-serializable representation of the object |
Raises:
| Type | Description |
|---|---|
TypeError
|
If object type is not serializable |
LokiUploader
¶
A class for uploading logs to Grafana Loki.
This class provides methods to upload structured logs from JSONL files to Grafana Loki for centralized log management and analysis.
Attributes:
| Name | Type | Description |
|---|---|---|
url |
The Loki push API endpoint URL |
|
user |
Username for authentication |
|
api_key |
API key for authentication |
|
service_name |
Default service name for log streams |
|
environment |
Default environment for log streams |
|
timeout |
Default timeout for HTTP requests in seconds |
Example
uploader = LokiUploader( ... url="https://logs-prod.grafana.net/loki/api/v1/push", ... user="your_user_id", ... api_key="your_api_key" ... ) uploader.upload_from_file("/path/to/logs.jsonl")
url = url
instance-attribute
¶
user = user
instance-attribute
¶
api_key = api_key
instance-attribute
¶
service_name = service_name
instance-attribute
¶
environment = environment
instance-attribute
¶
timeout = timeout
instance-attribute
¶
__init__(url: str, user: str, api_key: str, service_name: str = 'the_data_packet', environment: str = 'production', timeout: int = 30)
¶
Initialize the LokiUploader.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
Loki push API endpoint URL |
required |
user
|
str
|
Username for authentication |
required |
api_key
|
str
|
API key for authentication |
required |
service_name
|
str
|
Default service name for log streams |
'the_data_packet'
|
environment
|
str
|
Default environment for log streams |
'production'
|
timeout
|
int
|
Default timeout for HTTP requests in seconds |
30
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If any required parameter is empty or None |
upload_from_file(file_path: Path, service_name: Optional[str] = None, environment: Optional[str] = None, timeout: Optional[int] = None) -> int
¶
Upload logs from a JSONL file to Loki.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path
|
Path
|
Path to the JSONL log file |
required |
service_name
|
Optional[str]
|
Override default service name |
None
|
environment
|
Optional[str]
|
Override default environment |
None
|
timeout
|
Optional[int]
|
Override default timeout |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of log entries successfully uploaded |
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
If the log file doesn't exist |
ValueError
|
If the file contains invalid JSON |
LogUploadError
|
If the upload fails |
upload_logs(logs: List[Dict[str, Any]], service_name: Optional[str] = None, environment: Optional[str] = None, timeout: Optional[int] = None) -> int
¶
Upload logs directly from a list of dictionaries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
logs
|
List[Dict[str, Any]]
|
List of log entries as dictionaries |
required |
service_name
|
Optional[str]
|
Override default service name |
None
|
environment
|
Optional[str]
|
Override default environment |
None
|
timeout
|
Optional[int]
|
Override default timeout |
None
|
Returns:
| Type | Description |
|---|---|
int
|
Number of log entries successfully uploaded |
Raises:
| Type | Description |
|---|---|
ValueError
|
If logs list is empty or contains invalid data |
LogUploadError
|
If the upload fails |
upload_logs_to_loki(file_path: Path, url: str, user: str, api_key: str, service_name: str = 'the_data_packet', environment: str = 'production', timeout: int = 30) -> int
¶
Convenience function for uploading logs to Loki.
This is a simple wrapper around the LokiUploader class for one-time log uploads without needing to instantiate the class.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path
|
Path
|
Path to the JSONL log file |
required |
url
|
str
|
Loki push API endpoint URL |
required |
user
|
str
|
Username for authentication |
required |
api_key
|
str
|
API key for authentication |
required |
service_name
|
str
|
Service name label for the logs |
'the_data_packet'
|
environment
|
str
|
Environment label for the logs |
'production'
|
timeout
|
int
|
Request timeout in seconds |
30
|
Returns:
| Type | Description |
|---|---|
int
|
Number of log entries successfully uploaded |
Raises:
| Type | Description |
|---|---|
LogUploadError
|
If log upload fails |
FileNotFoundError
|
If the log file doesn't exist |
ValueError
|
If the file contains invalid JSON or missing parameters |
Example
count = upload_logs_to_loki( ... file_path="/path/to/logs.jsonl", ... url="https://logs-prod.grafana.net/loki/api/v1/push", ... user="your_user_id", ... api_key="your_api_key" ... ) print(f"Uploaded {count} log entries")