.py

import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from time import perf_counter
from typing import Any, Callable
from uuid import UUID, uuid4

import orjson
from adaptix import NameStyle as RNameStyle
from adaptix import Retort, name_mapping
from dataclass_factory import Factory, NameStyle, Schema


def o_default(value: Any) -> str | None:
    if isinstance(value, (UUID, Decimal)):
        return str(value)
    if isinstance(value, bytes):
        return value.decode("utf-8")
    if callable(value):
        return


def is_snake_case(name: str) -> bool:
    return name.lower() == name


def split_by_underscore(s: str) -> list[str]:
    return s.split("_")


def camel_lower_fixed(snake_name: Any) -> str | Any:
    if isinstance(snake_name, str) and is_snake_case(snake_name):
        names = split_by_underscore(snake_name)
        return f"{names[0].lower()}{''.join(x.title() for x in names[1:])}"
    return snake_name


def keys_to_camel(d: Any) -> Any:
    if isinstance(d, list):
        return [keys_to_camel(i) if isinstance(i, (dict, list)) else i for i in d]
    return {
        camel_lower_fixed(a): keys_to_camel(b) if isinstance(b, (dict, list)) else b
        for a, b in d.items()
    }


def dumps_and_load(obj: Any, to_camel: bool = False) -> dict[str, Any]:
    dumped = orjson.dumps(
        obj,
        default=o_default,
        option=orjson.OPT_NON_STR_KEYS | orjson.OPT_SERIALIZE_DATACLASS,
    )
    _result = json.loads(dumped)
    if to_camel:
        _result = keys_to_camel(_result)

    return _result


def get_dcf():
    return Factory(
        default_schema=Schema(
            name_style=NameStyle.camel_lower,
        )
    )


def get_retort():
    return Retort(recipe=[name_mapping(name_style=RNameStyle.CAMEL)])


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


@dataclass
class IdUuidMixin:
    id: UUID = field(default_factory=uuid4)


@dataclass
class CreatedUpdatedAtMixin:
    created_at: datetime = field(default_factory=utcnow)
    updated_at: datetime = field(default_factory=utcnow)


@dataclass
class CreatedUpdatedByMixin:
    created_by_id: UUID | None = None
    updated_by_id: UUID | None = None


@dataclass
class _CompanyInvitationCode:
    code: str
    position: str
    email: str
    company_id: UUID
    is_used: bool = False
    expires_at: datetime | None = None


@dataclass
class CompanyInvitationCode(
    IdUuidMixin, CreatedUpdatedAtMixin, CreatedUpdatedByMixin, _CompanyInvitationCode
):
    ...


@dataclass
class _Permission:
    name: str
    i10n: str
    codename: str
    description: str | None = None


@dataclass
class Permission(IdUuidMixin, CreatedUpdatedAtMixin, _Permission):
    ...


@dataclass
class _Role:
    name: str
    i10n: str
    description: str | None = None
    granted_after_registration: bool = False
    is_company_role: bool = False


@dataclass
class Role(IdUuidMixin, CreatedUpdatedAtMixin, _Role):
    ...


@dataclass
class RolePermissions(Role):
    permissions: list[Permission] = field(default_factory=list)


@dataclass
class CompanyInvitationCodeRoles(CompanyInvitationCode):
    roles: list[RolePermissions] = field(default_factory=list)


example = CompanyInvitationCodeRoles(
    code="example",
    position="example",
    email="[email protected]",
    company_id=uuid4(),
    is_used=False,
    expires_at=utcnow(),
    created_by_id=uuid4(),
    updated_by_id=uuid4(),
    created_at=utcnow(),
    updated_at=utcnow(),
    id=uuid4(),
    roles=[
        RolePermissions(
            id=uuid4(),
            name="example",
            i10n="example",
            description="example",
            is_company_role=True,
            created_at=utcnow(),
            updated_at=utcnow(),
            permissions=[
                Permission(
                    name=f"example-{i}",
                    i10n=f"example-{i}",
                    codename=f"example-{i}",
                    description=f"example-{i}",
                    created_at=utcnow(),
                    updated_at=utcnow(),
                    id=uuid4(),
                )
                for i in range(5)
            ],
        )
    ],
)


def eq_check(d1_name: str, d1: dict[str, Any], d2_name: str, d2: dict[str, Any]):
    print(f"\nEquality check: {d1_name} and {d2_name}")
    if d1 != d2:
        print(f"{d1_name} != {d2_name}")
    else:
        print(f"{d1_name} == {d2_name}")


def eq_test():
    custom_result = dumps_and_load(example, to_camel=True)
    dcf_result = get_dcf().dump(example)
    adaptix_result = get_retort().dump(example)
    eq_check("custom", custom_result, "dataclass_factory", dcf_result)
    eq_check("custom", custom_result, "adaptix", adaptix_result)
    print()


def _timeit(fn: Callable[..., Any], *args: Any):
    t0 = perf_counter()
    fn(*args)
    t1 = perf_counter()
    return t1 - t0


def perf_example():
    factory = get_dcf()
    retort = get_retort()
    print("First call custom serialize")
    execution_time = _timeit(dumps_and_load, example, True)
    print(f"{execution_time} seconds")
    print("Second call custom serialize")
    execution_time = _timeit(dumps_and_load, example, True)
    print(f"{execution_time} seconds")
    print("First call dataclass_factory")
    execution_time = _timeit(factory.dump, example)
    print(f"{execution_time} seconds")
    print("Second call dataclass_factory")
    execution_time = _timeit(factory.dump, example)
    print(f"{execution_time} seconds")
    print("First call adaptix")
    execution_time = _timeit(retort.dump, example)
    print(f"{execution_time} seconds")
    print("Second call adaptix")
    execution_time = _timeit(retort.dump, example)
    print(f"{execution_time} seconds")


if __name__ == "__main__":
    eq_test()
    perf_example()