boehnenelton2024

BEJSON 104 Python Library

A comprehensive Python library for working with BEJSON version 104 files. Created for handling structured data interchange in Python.

Full Library Source


### IMPORTS AND SETUP ###
----
import json
import os
import datetime
import copy
from typing import Callable, List, Dict, Any

### EXCEPTION CLASS ###
----
class BEJSON104Error(Exception):
    """Custom exception for BEJSON 104 errors."""
    pass

### MAIN CLASS ###
----
class BEJSON104:
    """A class to manage BEJSON 104 configuration files with enhanced features."""
    
    ### INITIALIZATION ###
    ----
    def __init__(self, default_path: str = "~/.bejson_configs/config.json"):
        self.data = {
            "Format": "BEJSON",
            "Format_Version": "104",
            "Format_Creator": "Elton Boehnen",
            "Records_Type": [],
            "Fields": [],
            "Values": []
        }
        self.default_path = os.path.expanduser(default_path)
        self.cache = None  # Cache for loaded data
        self.non_null_fields = set()  # Custom non-null constraints
    
    ### VALIDATION ###
    ----
    def validate(self, strict: bool = True):
        """Validate BEJSON 104 structure and data types.
        
        Args:
            strict (bool): If False, skip non-critical validations for performance.
        """
        required_keys = {"Format", "Format_Version", "Format_Creator", "Records_Type", "Fields", "Values"}
        if strict and not all(key in self.data for key in required_keys):
            raise BEJSON104Error(f"Missing required keys: {required_keys - set(self.data)}")
        if strict and len(self.data) > len(required_keys):
            raise BEJSON104Error("Custom top-level keys not allowed in 104")
        
        if self.data["Format"] != "BEJSON":
            raise BEJSON104Error("Format must be 'BEJSON'")
        if self.data["Format_Version"] != "104":
            raise BEJSON104Error("Format_Version must be '104'")
        if self.data["Format_Creator"] != "Elton Boehnen":
            raise BEJSON104Error("Format_Creator must be 'Elton Boehnen'")
        
        if not isinstance(self.data["Records_Type"], list) or len(self.data["Records_Type"]) != 1 or not isinstance(self.data["Records_Type"][0], str):
            raise BEJSON104Error("Records_Type must be a list with one string")
        
        if not isinstance(self.data["Fields"], list) or not isinstance(self.data["Values"], list):
            raise BEJSON104Error("Fields and Values must be lists")
        
        valid_types = {"string", "integer", "number", "boolean", "array", "object"}
        for field in self.data["Fields"]:
            if not isinstance(field, dict) or "name" not in field or "type" not in field:
                raise BEJSON104Error("Each Field must have 'name' and 'type'")
            if field["type"] not in valid_types:
                raise BEJSON104Error(f"Invalid type {field['type']} in Fields")
        
        for record in self.data["Values"]:
            if not isinstance(record, list) or len(record) != len(self.data["Fields"]):
                raise BEJSON104Error("Each Values record must match Fields length")
            for value, field in zip(record, self.data["Fields"]):
                if field["name"] in self.non_null_fields and value is None:
                    raise BEJSON104Error(f"Field {field['name']} cannot be null")
                if field["type"] == "integer" and (not isinstance(value, int) or isinstance(value, bool)):
                    raise BEJSON104Error(f"Field {field['name']} must be an integer")
                elif field["type"] == "number" and not isinstance(value, (int, float)) or isinstance(value, bool):
                    raise BEJSON104Error(f"Field {field['name']} must be a number")
                elif field["type"] == "string" and not isinstance(value, str):
                    raise BEJSON104Error(f"Field {field['name']} must be a string")
                elif field["type"] == "boolean" and not isinstance(value, bool):
                    raise BEJSON104Error(f"Field {field['name']} must be a boolean")
                elif field["type"] == "array" and not isinstance(value, list):
                    raise BEJSON104Error(f"Field {field['name']} must be an array")
                elif field["type"] == "object" and not isinstance(value, dict):
                    raise BEJSON104Error(f"Field {field['name']} must be an object")
    
    ### CREATE CONFIGURATION ###
    ----
    def create(self, record_type: str, fields: List[Dict[str, Any]], template: str = None):
        """Create a new BEJSON 104 configuration, optionally from a template.
        
        Args:
            record_type (str): The single record type (e.g., 'Settings').
            fields (list): List of dicts with 'name', 'type', and optional 'default' or 'non_null'.
            template (str, optional): Template name ('app_settings', 'user_prefs').
        """
        if template:
            templates = {
                "app_settings": [
                    {"name": "app_name", "type": "string", "default": "DefaultApp", "non_null": True},
                    {"name": "version", "type": "number", "default": 1.0},
                    {"name": "debug_mode", "type": "boolean", "default": False}
                ],
                "user_prefs": [
                    {"name": "user_id", "type": "integer", "default": 0, "non_null": True},
                    {"name": "username", "type": "string", "default": "guest"},
                    {"name": "preferences", "type": "object", "default": {}}
                ]
            }
            if template not in templates:
                raise BEJSON104Error(f"Unknown template: {template}")
            fields = templates[template]
        
        if not isinstance(record_type, str):
            raise BEJSON104Error("Record type must be a string")
        if not isinstance(fields, list) or not all(isinstance(f, dict) and "name" in f and "type" in f for f in fields):
            raise BEJSON104Error("Fields must be a list of dicts with 'name' and 'type'")
        
        self.data["Records_Type"] = [record_type]
        self.data["Fields"] = fields
        self.data["Values"] = []
        self.cache = copy.deepcopy(self.data)
        self.non_null_fields = {f["name"] for f in fields if f.get("non_null", False)}
        self.validate()
    
    ### ADD RECORD ###
    ----
    def add_record(self, values: List[Any] = None):
        """Add a record to Values, using defaults for unspecified values.
        
        Args:
            values (list, optional): Record values matching Fields; None uses defaults.
        """
        if values is None:
            values = [f.get("default", None) for f in self.data["Fields"]]
        elif not isinstance(values, list) or len(values) != len(self.data["Fields"]):
            raise BEJSON104Error("Values record must match Fields length")
        
        final_values = []
        for i, (value, field) in enumerate(zip(values, self.data["Fields"])):
            final_values.append(value if value is not None else field.get("default", None))
        
        self.data["Values"].append(final_values)
        self.cache = copy.deepcopy(self.data)
        self.validate()
    
    ### ADD RECORDS BATCH ###
    ----
    def add_records_batch(self, records: List[List[Any]]):
        """Add multiple records to Values.
        
        Args:
            records (list): List of record value lists.
        """
        for values in records:
            self.add_record(values)
    
    ### DELETE RECORD ###
    ----
    def delete_record(self, record_index: int = None, condition: Callable[[List[Any]], bool] = None):
        """Delete a record by index or condition.
        
        Args:
            record_index (int, optional): Index of the record to delete.
            condition (callable, optional): Function returning True for records to delete.
        """
        if record_index is not None:
            if not 0 <= record_index < len(self.data["Values"]):
                raise BEJSON104Error("Invalid record index")
            self.data["Values"].pop(record_index)
        elif condition:
            self.data["Values"] = [r for r in self.data["Values"] if not condition(r)]
        else:
            raise BEJSON104Error("Must provide index or condition")
        self.cache = copy.deepcopy(self.data)
        self.validate()
    
    ### SAVE CONFIGURATION ###
    ----
    def save(self, filename: str = None):
        """Save BEJSON 104 configuration to a file with automatic backup.
        
        Args:
            filename (str, optional): Path to save the JSON file; uses default_path if None.
        """
        filename = filename or self.default_path
        os.makedirs(os.path.dirname(filename), exist_ok=True)
        
        if os.path.exists(filename):
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_filename = f"{filename}.{timestamp}.bak"
            with open(filename, "r") as f:
                with open(backup_filename, "w") as bf:
                    bf.write(f.read())
        
        self.validate()
        with open(filename, "w") as f:
            json.dump(self.data, f, indent=2)
        self.cache = copy.deepcopy(self.data)
    
    ### LOAD CONFIGURATION ###
    ----
    def load(self, filename: str = None):
        """Load and validate a BEJSON 104 configuration from a file.
        
        Args:
            filename (str, optional): Path to the JSON file; uses default_path if None.
        """
        filename = filename or self.default_path
        if not os.path.exists(filename):
            raise BEJSON104Error(f"File {filename} does not exist")
        with open(filename, "r") as f:
            self.data = json.load(f)
        self.cache = copy.deepcopy(self.data)
        self.non_null_fields = {f["name"] for f in self.data["Fields"] if f.get("non_null", False)}
        self.validate()
    
    ### SYNC CACHE ###
    ----
    def sync_cache(self):
        """Sync cache with current data and save to default file."""
        self.cache = copy.deepcopy(self.data)
        self.save()
    
    ### SELECT DATA ###
    ----
    def select(self):
        """Return the current BEJSON 104 data from cache or data."""
        return self.cache if self.cache else self.data
    
    ### SEARCH RECORDS ###
    ----
    def search_records(self, field_name: str, value: Any) -> List[List[Any]]:
        """Search records by field value.
        
        Args:
            field_name (str): Name of the field to search.
            value: Value to match.
        
        Returns:
            list: List of matching records.
        """
        field_index = None
        for i, field in enumerate(self.data["Fields"]):
            if field["name"] == field_name:
                field_index = i
                break
        if field_index is None:
            raise BEJSON104Error(f"Field {field_name} not found")
        
        return [record for record in self.data["Values"] if record[field_index] == value]
    
    ### FILTER RECORDS ###
    ----
    def filter_records(self, condition: Callable[[List[Any]], bool]) -> List[List[Any]]:
        """Filter records based on a condition.
        
        Args:
            condition (callable): Function returning True for records to keep.
        
        Returns:
            list: List of matching records.
        """
        return [record for record in self.data["Values"] if condition(record)]
    
    ### EDIT FIELD ###
    ----
    def edit_field(self, field_name: str, new_name: str = None, new_type: str = None, new_default: Any = None, non_null: bool = None):
        """Edit a field's name, type, default value, or non-null constraint.
        
        Args:
            field_name (str): Name of the field to edit.
            new_name (str, optional): New field name.
            new_type (str, optional): New field type.
            new_default (Any, optional): New default value.
            non_null (bool, optional): Set non-null constraint.
        """
        for field in self.data["Fields"]:
            if field["name"] == field_name:
                if new_name:
                    if field["name"] in self.non_null_fields:
                        self.non_null_fields.remove(field["name"])
                        self.non_null_fields.add(new_name)
                    field["name"] = new_name
                if new_type:
                    if new_type not in {"string", "integer", "number", "boolean", "array", "object"}:
                        raise BEJSON104Error(f"Invalid type {new_type}")
                    field["type"] = new_type
                if new_default is not None:
                    field["default"] = new_default
                if non_null is not None:
                    if non_null:
                        self.non_null_fields.add(field["name"])
                    elif field["name"] in self.non_null_fields:
                        self.non_null_fields.remove(field["name"])
                self.cache = copy.deepcopy(self.data)
                self.validate()
                return
        raise BEJSON104Error(f"Field {field_name} not found")
    
    ### EDIT VALUE ###
    ----
    def edit_value(self, record_index: int, field_name: str, new_value: Any):
        """Edit a value in a specific record.
        
        Args:
            record_index (int): Index of the record in Values.
            field_name (str): Name of the field to edit.
            new_value: New value for the field.
        """
        if not 0 <= record_index < len(self.data["Values"]):
            raise BEJSON104Error("Invalid record index")
        
        field_index = None
        for i, field in enumerate(self.data["Fields"]):
            if field["name"] == field_name:
                field_index = i
                break
        if field_index is None:
            raise BEJSON104Error(f"Field {field_name} not found")
        
        field_type = self.data["Fields"][field_index]["type"]
        if field_name in self.non_null_fields and new_value is None:
            raise BEJSON104Error(f"Field {field_name} cannot be null")
        if field_type == "integer" and (not isinstance(new_value, int) or isinstance(new_value, bool)):
            raise BEJSON104Error(f"Value for {field_name} must be an integer")
        elif field_type == "number" and not isinstance(new_value, (int, float)) or isinstance(new_value, bool):
            raise BEJSON104Error(f"Value for {field_name} must be a number")
        elif field_type == "string" and not isinstance(new_value, str):
            raise BEJSON104Error(f"Value for {field_name} must be a string")
        elif field_type == "boolean" and not isinstance(new_value, bool):
            raise BEJSON104Error(f"Value for {field_name} must be a boolean")
        elif field_type == "array" and not isinstance(new_value, list):
            raise BEJSON104Error(f"Value for {field_name} must be an array")
        elif field_type == "object" and not isinstance(new_value, dict):
            raise BEJSON104Error(f"Value for {field_name} must be an object")
        
        self.data["Values"][record_index][field_index] = new_value
        self.cache = copy.deepcopy(self.data)
        self.validate()

    ### EXAMPLE USAGE ###
    ----
    if __name__ == "__main__":
        bejson = BEJSON104()
        
        # Create using template
        bejson.create("Settings", [], template="app_settings")
        
        # Add records with defaults and batch
        bejson.add_record()  # Uses defaults
        bejson.add_records_batch([
            ["TestApp", 2.0, True],
            ["ProdApp", 1.5, False]
        ])
        
        # Set non-null constraint
        bejson.edit_field("app_name", non_null=True)
        
        # Save to default path
        bejson.save()
        
        # Load and search
        bejson.load()
        print("Search app_name='TestApp':", bejson.search_records("app_name", "TestApp"))
        
        # Filter records
        filtered = bejson.filter_records(lambda r: r[1] > 1.0)
        print("Filtered records (version > 1.0):", filtered)
        
        # Edit field and value
        bejson.edit_field("debug_mode", new_name="is_debug", new_default=True)
        bejson.edit_value(0, "version", 1.1)
        
        # Delete record
        bejson.delete_record(record_index=1)
        
        # Sync cache and save
        bejson.sync_cache()
        print("Final data:", bejson.select())