Modern high-performance serialization utilities for Python
High-performance JSON serialization and deserialization with comprehensive file I/O support, gzip compression, JSONL format handling, and cross-platform compatibility.
Core JSON serialization functions for converting between Python objects and JSON strings with optimized performance via ujson.
def json_dumps(data, indent=0, sort_keys=False):
"""
Serialize an object to a JSON string.
Parameters:
- data: JSON-serializable Python object
- indent (int): Number of spaces for indentation (0 for compact)
- sort_keys (bool): Sort dictionary keys (falls back to standard json)
Returns:
str: JSON string representation
"""
def json_loads(data):
"""
Deserialize JSON string or bytes to Python object.
Parameters:
- data (str | bytes): JSON data to deserialize
Returns:
Any: Deserialized Python object
"""Read and write JSON data to files with automatic encoding handling and support for standard input/output streams.
def read_json(location):
"""
Load JSON from file or standard input.
Parameters:
- location (str | Path | "-"): File path or "-" for stdin
Returns:
dict | list: Loaded JSON content
Raises:
ValueError: If file doesn't exist or contains invalid JSON
"""
def write_json(location, data, indent=2):
"""
Write JSON data to file or standard output.
Parameters:
- location (str | Path | "-"): File path or "-" for stdout
- data: JSON-serializable data to write
- indent (int): Number of spaces for indentation
Returns:
None
"""Handle compressed JSON files with automatic gzip compression and decompression.
def read_gzip_json(location):
"""
Load JSON from gzipped file.
Parameters:
- location (str | Path): Path to .json.gz file
Returns:
dict | list: Loaded JSON content
Raises:
ValueError: If file doesn't exist or contains invalid JSON
"""
def write_gzip_json(location, data, indent=2):
"""
Write JSON data to gzipped file.
Parameters:
- location (str | Path): Path for .json.gz file
- data: JSON-serializable data to write
- indent (int): Number of spaces for indentation
Returns:
None
"""Handle newline-delimited JSON format for streaming large datasets and log files.
def read_jsonl(location, skip=False):
"""
Read JSONL file line by line, yielding each JSON object.
Parameters:
- location (str | Path | "-"): File path or "-" for stdin
- skip (bool): Skip invalid lines instead of raising error
Yields:
Any: Deserialized JSON object from each line
Raises:
ValueError: If line contains invalid JSON and skip=False
"""
def write_jsonl(location, lines, append=False, append_new_line=True):
"""
Write iterable of objects as JSONL format.
Parameters:
- location (str | Path | "-"): File path or "-" for stdout
- lines (Iterable): JSON-serializable objects to write
- append (bool): Append to existing file instead of overwriting
- append_new_line (bool): Add newline before appending
Returns:
None
"""Utility function to check if objects are JSON-serializable.
def is_json_serializable(obj):
"""
Check if a Python object can be serialized to JSON.
Parameters:
- obj: Object to test for JSON compatibility
Returns:
bool: True if object is JSON-serializable, False otherwise
"""import srsly
# Serialize to JSON string
data = {"users": [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]}
json_str = srsly.json_dumps(data, indent=2)
print(json_str)
# Parse JSON string
parsed = srsly.json_loads(json_str)
print(parsed["users"][0]["name"]) # "Alice"import srsly
try:
# Write to file
data = {"config": {"debug": True, "timeout": 30}}
srsly.write_json("config.json", data)
# Read from file
loaded_config = srsly.read_json("config.json")
print(f"Debug mode: {loaded_config['config']['debug']}")
except ValueError as e:
print(f"JSON error: {e}")import srsly
# Write log entries
log_entries = [
{"timestamp": "2023-01-01T12:00:00", "level": "INFO", "message": "Server started"},
{"timestamp": "2023-01-01T12:01:00", "level": "ERROR", "message": "Connection failed"}
]
srsly.write_jsonl("app.log", log_entries)
# Read and process log entries
for entry in srsly.read_jsonl("app.log"):
if entry["level"] == "ERROR":
print(f"Error at {entry['timestamp']}: {entry['message']}")import srsly
# Check if objects are serializable
test_objects = [
{"valid": "object"},
lambda x: x, # Functions are not serializable
[1, 2, 3],
set([1, 2, 3]) # Sets are not serializable
]
for obj in test_objects:
if srsly.is_json_serializable(obj):
json_str = srsly.json_dumps(obj)
print(f"Serialized: {json_str}")
else:
print(f"Cannot serialize: {type(obj)}")