ObjJson is a utility class for handling JSON objects with a focus on SQL-safe storage and serialization. It provides optimized JSON operations using the orjson library for high-performance JSON parsing and generation.
Module: factory.core/ObjJson.py
Inherits from: Objects.Object
Test file: resource.test/pytests/factory.core/test_ObjJson.py
Version: 1.0
This module enables developers to:
orjson library for fast JSON operations__init__(DB: Objects.DatabaseConnection = 0) -> NoneInitializes the ObjJson instance.
Parameters:
DB (Objects.DatabaseConnection, optional): Database connection object (default: 0)safe_json(data: Union[Dict[Any, Any], List[Any], str, None]) -> strConverts a JSON-like object into a SQL-safe JSON string.
Parameters:
data: The data to convert (dict, list, string representing JSON, or None)Returns:
Behavior:
"NULL""NULL" if invalid"NULL" and logs errorExamples:
obj = ObjJson()
# Dictionary
obj.safe_json({"key": "value"})
# Returns: '{"key":"value"}'
# List
obj.safe_json(["item1", 2, {"nested": "value"}])
# Returns: '["item1",2,{"nested":"value"}]'
# String with single quotes
obj.safe_json({"message": "User's input"})
# Returns: '{"message":"User''s input"}'
# None
obj.safe_json(None)
# Returns: NULL
# Invalid JSON string
obj.safe_json('{"invalid": }')
# Returns: NULL
load_json_string(json_string: str) -> Union[Dict[Any, Any], List[Any]]Loads a JSON string into a Python object using orjson.
Parameters:
json_string (str): The JSON string to loadReturns:
Raises:
orjson.JSONDecodeError: If the string is not valid JSONExample:
obj = ObjJson()
json_str = '{"name": "test", "value": 123}'
result = obj.load_json_string(json_str)
# Returns: {'name': 'test', 'value': 123}
dump_json_to_string(data: Union[Dict[Any, Any], List[Any]]) -> strDumps a Python object to a JSON string using orjson.
Parameters:
data: The data to dump to a JSON string (dict or list)Returns:
Example:
obj = ObjJson()
data = {"key": "value", "number": 42}
result = obj.dump_json_to_string(data)
# Returns: '{"key":"value","number":42}'
from ObjJson import ObjJson
# Create instance
json_handler = ObjJson()
# Prepare data for SQL insertion
data = {"user_id": 123, "action": "login", "timestamp": "2026-02-04"}
sql_safe_json = json_handler.safe_json(data)
# Use in SQL query
sql = f"INSERT INTO logs (data) VALUES ({sql_safe_json})"
from ObjJson import ObjJson
json_handler = ObjJson()
# User input may contain single quotes
user_data = {
"comment": "It's working great!",
"query": "SELECT * FROM 'users'"
}
# Safe for SQL insertion
safe_data = json_handler.safe_json(user_data)
# Result: '{"comment":"It''s working great!","query":"SELECT * FROM ''users''"}'
from ObjJson import ObjJson
json_handler = ObjJson()
# Handle invalid JSON
invalid_json = '{"name": "test", "id":}' # Missing value
result = json_handler.safe_json(invalid_json)
# Result: NULL (and error is logged)
# Handle unserializable objects
class CustomObject:
pass
data = {"object": CustomObject()}
result = json_handler.safe_json(data)
# Result: NULL (and error is logged)
from ObjJson import ObjJson
json_handler = ObjJson()
# Complex nested structure
data = {
"level1": {
"level2": [
"item1",
{"level3": 3}
]
}
}
safe_json = json_handler.safe_json(data)
# Result: '{"level1":{"level2":["item1",{"level3":3}]}}'
The module has comprehensive test coverage for all methods and edge cases:
| Test Case | Description | Status |
|---|---|---|
test_safe_json_valid_dict |
Valid dictionary input | ✓ |
test_safe_json_valid_list |
Valid list input | ✓ |
test_safe_json_string_input_valid_json |
String containing valid JSON | ✓ |
test_safe_json_string_input_invalid_json |
String with invalid JSON | ✓ |
test_safe_json_none_input |
None value handling | ✓ |
test_safe_json_empty_dict |
Empty dictionary | ✓ |
test_safe_json_empty_list |
Empty list | ✓ |
test_safe_json_nested_data |
Nested dictionaries and lists | ✓ |
test_safe_json_with_quotes |
Data containing single quotes | ✓ |
test_safe_json_unserializable_object |
Non-serializable objects | ✓ |
Test file location: resource.test/pytests/factory.core/test_ObjJson.py
Run tests:
pytest resource.test/pytests/factory.core/test_ObjJson.py -v
orjson - High-performance JSON libraryObjects - Base object classThe module uses orjson instead of the standard json library for several advantages:
SQL uses single quotes for string literals. JSON data may contain single quotes that need escaping:
# Input
{"message": "It's a test"}
# Output (single quotes doubled for SQL)
'{"message":"It''s a test"}'
For SQL compatibility, None values and invalid data return the literal string "NULL" (without quotes):
# Input
None
# Output (no quotes around NULL)
NULL
# SQL usage
INSERT INTO table (json_col) VALUES (NULL) # Valid SQL
safe_json() method properly escapes single quotesload_json_string() and dump_json_to_string() instead of safe_json()from ObjJson import ObjJson
from ObjData import ObjData
json_handler = ObjJson()
db = ObjData()
data = {"config": {"setting1": "value1", "setting2": 123}}
safe_json = json_handler.safe_json(data)
sql = f"INSERT INTO configs (name, data) VALUES ('myconfig', {safe_json})"
db.sql_update(sql)
from ObjJson import ObjJson
json_handler = ObjJson()
# Receive JSON from API
user_json = request.json
# Convert to SQL-safe format
safe_json = json_handler.safe_json(user_json)
if safe_json == "NULL":
return {"error": "Invalid JSON data"}
else:
# Store in database
sql = f"INSERT INTO user_data (json_data) VALUES ({safe_json})"
ObjData.py - Database operations that use JSON storageObjApi.py - API operations that handle JSON payloadsObjWorkflow.py - Workflow configurations stored as JSONorjson which requires UTF-8 encodingerror() method from Objects.Object