2468 WHO DO WE APPRECIATE THAT'S BBNO$ HE'S ALWAYS UP TO SOEMTHIGN

Author: AWeirdDev

python/exacting/__init__.py

@@ -0,0 +1,44 @@
+from .core import Exact, exact
+from .types import ValidationError
+from .validators import (
+    AnnotatedV,
+    AnyV,
+    BoolV,
+    BytesV,
+    DataclassV,
+    DictV,
+    FloatV,
+    IntV,
+    ListV,
+    LiteralV,
+    LooseDictV,
+    LooseListV,
+    NoneV,
+    StrV,
+    UnionV,
+    Validator,
+)
+from .utils import unsafe
+
+__all__ = [
+    "Exact",
+    "exact",
+    "ValidationError",
+    "AnnotatedV",
+    "AnyV",
+    "BoolV",
+    "BytesV",
+    "DataclassV",
+    "DictV",
+    "FloatV",
+    "IntV",
+    "ListV",
+    "LiteralV",
+    "LooseDictV",
+    "LooseListV",
+    "NoneV",
+    "StrV",
+    "UnionV",
+    "Validator",
+    "unsafe"
+]

python/exacting/core.py

@@ -0,0 +1,106 @@
+import json
+from dataclasses import MISSING, asdict, dataclass
+
+from typing import Any, Dict, Type
+from typing_extensions import Self, dataclass_transform
+
+from .validators import DataclassV, Validator
+from .validator_map import get_dc_validator
+from .types import DataclassType
+from .result import Result
+from .utils import get_field_value, unsafe_mode
+
+from .exacting import py_to_bytes
+
+
+def get_exact_init(dc: DataclassType):
+    setattr(dc, "__validator__", get_dc_validator(dc))
+
+    def init(self, **kwargs):
+        for field in self.__dataclass_fields__.values():
+            value = get_field_value(kwargs.get(field.name, MISSING), field)
+            setattr(self, field.name, value)
+
+        validator: Validator = getattr(dc, "__validator__")
+        res: Result = validator.validate(self)
+        res.raise_for_err()
+
+        return None
+
+    return init
+
+
+def get_unsafe_init():
+    @classmethod
+    def __unsafe_init__(cls, **kwargs):
+        if not unsafe_mode.get():
+            raise RuntimeError("Scope is not in unsafe(), canceled operation")
+
+        item = cls.__new__(cls)
+        for field in cls.__dataclass_fields__.values():
+            setattr(
+                item,
+                field.name,
+                get_field_value(kwargs.get(field.name, MISSING), field),
+            )
+        return item
+
+    return __unsafe_init__
+
+
+@dataclass_transform(kw_only_default=True)
+def exact(cls: Type) -> DataclassType:
+    dc = dataclass(kw_only=True)(cls)
+    unsafe_init = get_unsafe_init()
+    setattr(dc, "__unsafe_init__", unsafe_init)
+
+    exact_init = get_exact_init(dc)
+    setattr(dc, "__init__", exact_init)
+
+    return dc
+
+
+class _Internals:
+    __validator__: DataclassV
+
+    @classmethod
+    def __unsafe_init__(cls, **kwargs) -> Self:
+        """Unsafely initialize the dataclass with no-brain filling.
+
+        Example:
+
+        ```python
+        from exacting import unsafe
+
+        with unsafe():
+            SomeDataclass.__unsafe_init__(**kwargs)
+        ```
+        """
+        raise NotImplementedError()
+
+
+@dataclass_transform(kw_only_default=True)
+class _Dc: ...
+
+
+class Exact(_Dc, _Internals):
+    def __init_subclass__(cls) -> None:
+        exact(cls)
+
+    def exact_as_dict(self) -> Dict[str, Any]:
+        """Get this model instance as dictionary."""
+        return asdict(self)
+
+    def exact_as_json(self) -> str:
+        """Get this model instance as JSON."""
+        return json.dumps(self.exact_as_dict())
+
+    def exact_as_bytes(self) -> bytes:
+        """Get this model instance as bytes with `rkyv`."""
+        return py_to_bytes(self.exact_as_dict())
+
+    @classmethod
+    def exact_from_dict(cls, d: Dict[str, Any]) -> Self:
+        res = cls.__validator__.validate(d, from_dict=True)
+        res.raise_for_err()
+        return res.unwrap()

python/exacting/result.py

@@ -0,0 +1,91 @@
+from collections import deque
+from typing import Generic, Optional, TypeVar
+
+from .types import ValidationError
+
+T = TypeVar("T")
+
+
+def build_error(errors: deque[str]) -> str:
+    text = "\n"
+    indent_level = 0
+
+    # drain contents
+    while errors:
+        error = errors.popleft()
+        if error == "indent":
+            indent_level += 1
+            continue
+        elif error == "unindent":
+            indent_level -= 1
+            continue
+
+        if indent_level:
+            text += f"{'  ' * indent_level}• {error}\n"
+        else:
+            text += f"{error}\n"
+
+    return text.rstrip()
+
+
+class Result(Generic[T]):
+    """Represents a result."""
+
+    ok_data: Optional[T]
+    errors: Optional["deque[str]"]  # deque is O(1)
+
+    def __init__(self, okd: Optional[T], errors: Optional["deque[str]"]):
+        self.ok_data = okd
+        self.errors = errors
+
+    @classmethod
+    def Ok(cls, data: T) -> "Result[T]":
+        return cls(data, None)
+
+    @classmethod
+    def Err(cls, *errors: str) -> "Result[T]":
+        return cls(None, deque(errors))
+
+    def unwrap(self) -> T:
+        """Unwrap the OK data."""
+        # cheap operation lmfao
+        return self.ok_data  # type: ignore
+
+    def unwrap_err(self) -> "deque[str]":
+        """Unwrap the Err data."""
+        # AGAIN. lmfao! you gotta be responsible.
+        return self.errors  # type: ignore
+
+    def is_ok(self) -> bool:
+        """CALL."""
+        return not self.errors
+
+    def trace(self, upper: str) -> "Result[T]":
+        if self.errors is not None:
+            self.errors.appendleft("indent")
+            self.errors.appendleft(upper)
+            self.errors.append("unindent")
+
+        return self
+
+    @classmethod
+    def trace_below(cls, upper: str, *items: str) -> "Result[T]":
+        errors = deque(items)
+        errors.appendleft("indent")
+        errors.appendleft(upper)
+        errors.append("unindent")
+
+        return cls(okd=None, errors=errors)
+
+    def raise_for_err(self) -> None:
+        if self.is_ok():
+            return
+
+        error = build_error(self.unwrap_err())
+        raise ValidationError(error)
+
+    def __repr__(self) -> str:
+        if self.is_ok():
+            return f"Result.Ok({self.unwrap()!r})"
+        else:
+            return f"Result.Err({self.unwrap_err()!r})"

python/exacting/types.py

@@ -1,65 +1,6 @@
 import dataclasses as std_dc
 
-from collections import deque
-from typing import Any, Dict, Generic, Optional, Protocol, Type, TypeVar, Union
-
-T = TypeVar("T")
-
-
-class Result(Generic[T]):
-    """Represents a result."""
-
-    ok_data: Optional[T]
-    errors: Optional["deque[str]"]  # O(1)
-
-    def __init__(self, okd: Optional[T], errors: Optional["deque[str]"]):
-        self.ok_data = okd
-        self.errors = errors
-
-    @classmethod
-    def Ok(cls, data: T) -> "Result[T]":
-        return cls(data, None)
-
-    @classmethod
-    def Err(cls, *errors: str) -> "Result[T]":
-        return cls(None, deque(errors))
-
-    def unwrap(self) -> T:
-        """Unwrap the OK data."""
-        # cheap operation lmfao
-        return self.ok_data  # type: ignore
-
-    def unwrap_err(self) -> "deque[str]":
-        """Unwrap the Err data."""
-        # AGAIN. lmfao! you gotta be responsible.
-        return self.errors  # type: ignore
-
-    def is_ok(self) -> bool:
-        """CALL."""
-        return not self.errors
-
-    def trace(self, upper: str) -> "Result[T]":
-        if self.errors is not None:
-            self.errors.appendleft("indent")
-            self.errors.appendleft(upper)
-            self.errors.append("unindent")
-
-        return self
-
-    @classmethod
-    def trace_below(cls, upper: str, *items: str) -> "Result[T]":
-        errors = deque(items)
-        errors.appendleft("indent")
-        errors.appendleft(upper)
-        errors.append("unindent")
-
-        return cls(okd=None, errors=errors)
-
-    def __repr__(self) -> str:
-        if self.is_ok():
-            return f"Result.Ok({self.unwrap()!r})"
-        else:
-            return f"Result.Err({self.unwrap_err()!r})"
+from typing import Any, Dict, Protocol, Type, Union
 
 
 class Dataclass(Protocol):
@@ -72,6 +13,7 @@ class Dataclass(Protocol):
 class _Indexable:
     def __getitem__(self, k: str): ...
     def __setitem__(self, k: str, data: Any): ...
+    def get(self, k: str): ...
     def as_dict(self) -> dict: ...
     def as_dc(self) -> Dataclass: ...
 
@@ -80,8 +22,11 @@ class _DefinitelyDict(_Indexable):
     def __init__(self, d: Dict):
         self.data = d
 
+    def get(self, k: str):
+        return self.data.get(k, std_dc.MISSING)
+
     def __getitem__(self, k: str):
-        self.data[k]
+        return self.data[k]
 
     def __setitem__(self, k: str, data: Any):
         self.data[k] = data
@@ -97,6 +42,9 @@ class _DefinitelyDataclass(_Indexable):
     def __init__(self, dc: Dataclass):
         self.dc = dc
 
+    def get(self, k: str):
+        return getattr(self.dc, k, std_dc.MISSING)
+
     def __getitem__(self, k: str):
         return getattr(self.dc, k)
 
@@ -115,3 +63,7 @@ def indexable(item: Any) -> "_Indexable":
         return _DefinitelyDict(item)
     else:
         return _DefinitelyDataclass(item)
+
+
+class ValidationError(RuntimeError):
+    """Validation error for `exacting`."""

python/exacting/utils.py

@@ -0,0 +1,39 @@
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import MISSING, Field, _MISSING_TYPE
+from typing import TypeVar, Union
+
+unsafe_mode = ContextVar("unsafe_mode", default=False)
+
+
+@contextmanager
+def unsafe(on: bool = True, /):
+    """Enable unsafe mode for unsafe operations.
+
+    Example:
+    ```python
+    with unsafe():
+        ...  # do stuff
+    ```
+    """
+
+    token = unsafe_mode.set(on)
+    try:
+        yield
+    finally:
+        unsafe_mode.reset(token)
+
+
+T = TypeVar("T")
+
+
+def get_field_value(item: Union[T, _MISSING_TYPE], field: Field) -> T:
+    if item is MISSING:
+        if field.default is not MISSING:
+            return field.default
+        elif field.default_factory is not MISSING:
+            return field.default_factory()
+        else:
+            raise KeyError(field.name)
+    else:
+        return item