amq-adapter-python/amqp/adapter/pydantic_serializer.py

"""
This module provides serialization and deserialization functions for Pydantic models that extend BaseModel
"""

import uuid
import json
from datetime import datetime
from typing import Dict, Any
from pydantic import BaseModel
from urllib.parse import urlparse, parse_qs

from amqp.adapter.logging_utils import logging_debug


def serialize_object(obj: BaseModel) -> str:
    """
    Serializes an R2RSerializable object to a string in the format
    "<ClassName>:{key1:value1, key2:value2,...}".  Handles UUIDs and datetimes,
    and nested Pydantic models.

    Args:
        obj: The BaseModel object to serialize.

    Returns:
        A string representation of the object.
    """
    class_name = obj.__class__.__name__
    obj_dict = {}
    for key, value in obj.model_dump().items():
        if isinstance(value, uuid.UUID):
            obj_dict[key] = str(value)
        elif isinstance(value, datetime):
            obj_dict[key] = value.isoformat()
        elif isinstance(value, dict):  # Ensure dictionaries are handled
            obj_dict[key] = value
        elif isinstance(value, BaseModel):
            obj_dict[key] = serialize_object(value)  # Recursively serialize
        elif isinstance(value, list):
            serialized_list = []
            for item in value:
                if isinstance(item, BaseModel):
                    serialized_list.append(serialize_object(item))
                else:
                    serialized_list.append(item)
            obj_dict[key] = serialized_list
        else:
            obj_dict[key] = value

    dict_string = ",".join(f"{k}:{v}" for k, v in obj_dict.items())
    return f"{class_name}:{{{dict_string}}}"


def deserialize_object(data: str) -> BaseModel:
    """
    Deserializes a string in the format "<ClassName>:{key1:value1, key2:value2,...}"
    back into an BaseModel object.  Handles UUIDs, datetimes, and
    nested Pydantic models.

    Args:
        data: The string to deserialize.

    Returns:
        An R2RSerializable object.
    """
    class_name, dict_str = data.split(":", 1)
    dict_str = dict_str.strip("{}")
    pairs = dict_str.split(",") if dict_str else []

    obj_dict: Dict[str, Any] = {}
    for pair in pairs:
        if not pair or ":" not in pair:
            continue
        k, v = pair.split(":", 1)
        v = v.strip()  # important
        obj_dict[k] = _convert_value(v)

    logging_debug(f"Deserialized object: {class_name}")
    if "[" in class_name:
        class_name = class_name.split("[")[0]
    logging_debug(f"trying object: {class_name}")
    cls = globals().get(class_name)
    if cls is None:
        raise ValueError(f"Unknown class: {class_name}")

    return cls(**obj_dict)


def _convert_value(v: str) -> Any:
    """
    Helper function to convert a string to its appropriate Python type.
    """
    if v == "None":
        return None
    elif _is_uuid_string(v):
        return uuid.UUID(v)
    elif _is_datetime_string(v):
        return datetime.fromisoformat(v)
    elif _is_int_string(v):
        return int(v)
    elif (
        v.startswith("{")
        and v.endswith("}")
        and ":" in v
        and "{" not in v[1:-1]
        and "}" not in v[1:-1]
    ):  # nested object
        # check if it is a json dict
        try:
            json.loads(v)
            return json.loads(v)
        except json.JSONDecodeError:
            return deserialize_object(v)  # recursive call
    elif v.startswith("[") and v.endswith("]"):  # list
        v = v[1:-1]  # remove brackets
        items = v.split(",") if v else []
        deserialized_list = []
        for item in items:
            item = item.strip()
            deserialized_list.append(_convert_value(item))  # recursive call
        return deserialized_list
    else:
        return v  # Keep as string


def _is_uuid_string(s: str) -> bool:
    """Helper to check if a string is a valid UUID."""
    try:
        uuid.UUID(s)
        return True
    except ValueError:
        return False


def _is_datetime_string(s: str) -> bool:
    """Helper to check if a string is a valid datetime in ISO format"""
    try:
        datetime.fromisoformat(s)
        return True
    except ValueError:
        return False


def _is_int_string(s: str) -> bool:
    """Helper to check if a string is a valid int."""
    try:
        int(s)
        return True
    except ValueError:
        return False


def parse_url_query(url):
    """
    Parses a HTTP GET URL and returns (context_path, query_params_dict)

    Args:
        url: The URL string to parse (e.g., "http://example.com/path?foo=1&bar=2&foo=3")

    Returns:
        tuple: (context_path, query_params_dict)
        - context_path: The path part of the URL (e.g., "/path")
        - query_params_dict: Dictionary of query parameters where values are always lists
                           (e.g., {"foo": ["1", "3"], "bar": ["2"]})
    """
    parsed = urlparse(url)
    query_params = parse_qs(parsed.query)

    # Convert values from lists to single values when there's only one value
    # If you prefer to always keep values as lists, remove this comprehension
    simplified_params = {k: v[0] if len(v) == 1 else v for k, v in query_params.items()}

    return (parsed.path, simplified_params)