Перейти к содержанию

API справочник

cwtch.cwtch

is_cwtch_model(cls) -> bool

Check if class is a cwtch model.

Source code in cwtch/cwtch.py
def is_cwtch_model(cls) -> bool:
    """Check if class is a cwtch model."""

    return bool(getattr(cls, "__cwtch_model__", None) and not getattr(cls, "__cwtch_view__", None))

is_cwtch_view(cls) -> bool

Check if class is a cwtch view.

Source code in cwtch/cwtch.py
def is_cwtch_view(cls) -> bool:
    """Check if class is a cwtch view."""

    return bool(getattr(cls, "__cwtch_model__", None) and getattr(cls, "__cwtch_view__", None))

field(default: Any = _MISSING, *, default_factory: _Missing[Callable] = _MISSING, init: bool = True, init_alias: Unset[str] = UNSET, repr: Unset[Literal[False]] = UNSET, compare: Unset[bool] = UNSET, property: Unset[Literal[True]] = UNSET, validate: Unset[bool] = UNSET, metadata: Unset[dict] = UNSET, kw_only: Unset[dict] = UNSET) -> Any

Return an object to identify dataclass fields.

Parameters:

Name Type Description Default
default Any

The default value of the field.

_MISSING
default_factory _Missing[Callable]

A 0-argument function called to initialize a field's value.

_MISSING
init bool

If init is true, the field will be a parameter to the class's __init__() function.

True
repr Unset[Literal[False]]

If repr is true, the field will be included in the object's repr().

UNSET
compare Unset[bool]

If compare is true, the field will be used in comparison functions.

UNSET
property Unset[Literal[True]]
UNSET
validate Unset[bool]
UNSET
metadata Unset[dict]

If specified, must be a mapping which is stored but not otherwise examined by dataclass.

UNSET
kw_only Unset[dict]

If kw_only true, the field will become a keyword-only parameter to __init__().

UNSET

It is an error to specify both default and default_factory.

Source code in cwtch/cwtch.py
def field(
    default: Any = _MISSING,
    *,
    default_factory: _Missing[Callable] = _MISSING,
    init: bool = True,
    init_alias: Unset[str] = UNSET,
    repr: Unset[Literal[False]] = UNSET,
    compare: Unset[bool] = UNSET,
    property: Unset[Literal[True]] = UNSET,
    validate: Unset[bool] = UNSET,
    metadata: Unset[dict] = UNSET,
    kw_only: Unset[dict] = UNSET,
) -> Any:
    """
    Return an object to identify dataclass fields.

    Args:
        default: The default value of the field.
        default_factory: A 0-argument function called to initialize a field's value.
        init: If init is true, the field will be a parameter to the class's `__init__()` function.
        repr: If repr is true, the field will be included in the object's repr().
        compare: If compare is true, the field will be used in comparison functions.
        property:
        validate:
        metadata: If specified, must be a mapping which is stored but not otherwise examined by dataclass.
        kw_only: If kw_only true, the field will become a keyword-only parameter to `__init__()`.

    It is an error to specify both default and default_factory.
    """
    return Field(
        default=default,
        default_factory=default_factory,
        init=init,
        init_alias=init_alias,
        repr=repr,
        compare=compare,
        property=property,
        validate=validate,
        metadata=metadata,
        kw_only=kw_only,
    )

dataclass(cls=None, *, slots: Unset[bool] = UNSET, kw_only: Unset[bool] = UNSET, env_prefix: Unset[str | Sequence[str]] = UNSET, env_source: Unset[Callable] = UNSET, validate: Unset[bool] = UNSET, add_disable_validation_to_init: Unset[bool] = UNSET, show_input_value_on_error: Unset[bool] = UNSET, extra: Unset[Literal['ignore', 'forbid']] = UNSET, repr: Unset[bool] = UNSET, eq: Unset[bool] = UNSET, recursive: Unset[bool | Sequence[str]] = UNSET, handle_circular_refs: Unset[bool] = UNSET) -> Callable[[Type[T]], Type[T]]

Parameters:

Name Type Description Default
slots Unset[bool]

If true, __slots__ attribute will be generated and new class will be returned instead of the original one. If __slots__ is already defined in the class, then TypeError is raised.

UNSET
kw_only Unset[bool]

If kw_only is true, then by default all fields are keyword-only.

UNSET
env_prefix Unset[str | Sequence[str]]

Prefix(or list of prefixes) for environment variables.

UNSET
env_source Unset[Callable]

Environment variables source factory. By default os.environ.

UNSET
validate Unset[bool]

Validate or not fields.

UNSET
add_disable_validation_to_init Unset[bool]

Add disable_validation keywoard argument to __init__() method to disable validation.

UNSET
extra Unset[Literal['ignore', 'forbid']]

Ignore or forbid extra arguments passed to init.

UNSET
repr Unset[bool]

If true, a __rich_repr__() method will be generated and rich.repr.auto decorator applied to the class.

UNSET
eq Unset[bool]

If true, an __eq__() method will be generated. This method compares the class as if it were a tuple of its fields, in order. Both instances in the comparison must be of the identical type.

UNSET
recursive Unset[bool | Sequence[str]]

...

UNSET
handle_circular_refs Unset[bool]

Handle or not circular refs.

UNSET
Source code in cwtch/cwtch.py
@dataclass_transform(field_specifiers=(field,))
def dataclass(
    cls=None,
    *,
    slots: Unset[bool] = UNSET,
    kw_only: Unset[bool] = UNSET,
    env_prefix: Unset[str | Sequence[str]] = UNSET,
    env_source: Unset[Callable] = UNSET,
    validate: Unset[bool] = UNSET,
    add_disable_validation_to_init: Unset[bool] = UNSET,
    show_input_value_on_error: Unset[bool] = UNSET,
    extra: Unset[Literal["ignore", "forbid"]] = UNSET,
    repr: Unset[bool] = UNSET,
    eq: Unset[bool] = UNSET,
    recursive: Unset[bool | Sequence[str]] = UNSET,
    handle_circular_refs: Unset[bool] = UNSET,
) -> Callable[[Type[T]], Type[T]]:
    """
    Args:
        slots: If true, `__slots__` attribute will be generated
            and new class will be returned instead of the original one.
            If `__slots__` is already defined in the class, then TypeError is raised.
        kw_only: If kw_only is true, then by default all fields are keyword-only.
        env_prefix: Prefix(or list of prefixes) for environment variables.
        env_source: Environment variables source factory. By default os.environ.
        validate: Validate or not fields.
        add_disable_validation_to_init: Add disable_validation keywoard argument to `__init__()` method
            to disable validation.
        extra: Ignore or forbid extra arguments passed to init.
        repr: If true, a `__rich_repr__()` method will be generated and rich.repr.auto decorator applied to the class.
        eq: If true, an `__eq__()` method will be generated.
            This method compares the class as if it were a tuple of its fields, in order.
            Both instances in the comparison must be of the identical type.
        recursive: ...
        handle_circular_refs: Handle or not circular refs.
    """

    if slots is UNSET:
        slots = SLOTS
    if kw_only is UNSET:
        kw_only = KW_ONLY
    if validate is UNSET:
        validate = VALIDATE
    if add_disable_validation_to_init is UNSET:
        add_disable_validation_to_init = ADD_DISABLE_VALIDATION_TO_INIT
    if show_input_value_on_error is UNSET:
        show_input_value_on_error = SHOW_INPUT_VALUE_ON_ERROR
    if extra is UNSET:
        extra = EXTRA
    if repr is UNSET:
        repr = REPR
    if eq is UNSET:
        eq = EQ
    if recursive is UNSET:
        recursive = RECURSIVE
    if handle_circular_refs is UNSET:
        handle_circular_refs = HANDLE_CIRCULAR_REFS

    def wrapper(
        cls,
        slots=slots,
        kw_only=kw_only,
        env_prefix=env_prefix,
        env_source=env_source,
        validate=validate,
        add_disable_validation_to_init=add_disable_validation_to_init,
        extra=extra,
        repr=repr,
        eq=eq,
        recursive=recursive,
        handle_circular_refs=handle_circular_refs,
    ):
        return _build(
            cls,
            cast(bool, slots),
            cast(bool, kw_only),
            cast(
                Unset[Sequence[str]],
                env_prefix if env_prefix is UNSET or isinstance(env_prefix, (list, tuple, set)) else [env_prefix],
            ),
            env_source,
            cast(bool, validate),
            cast(bool, add_disable_validation_to_init),
            cast(Literal["ignore", "forbid"], extra),
            cast(bool, repr),
            cast(bool, eq),
            cast(bool, recursive),
            cast(bool, handle_circular_refs),
        )

    if cls is None:
        return wrapper

    return wrapper(cls)

view(base_cls, name: Unset[str] = UNSET, *, attach: Unset[bool] = UNSET, include: Unset[Sequence[str]] = UNSET, exclude: Unset[Sequence[str]] = UNSET, slots: Unset[bool] = UNSET, kw_only: Unset[bool] = UNSET, env_prefix: Unset[str | Sequence[str]] = UNSET, env_source: Unset[Callable] = UNSET, validate: Unset[bool] = UNSET, add_disable_validation_to_init: Unset[bool] = UNSET, extra: Unset[Literal['ignore', 'forbid']] = UNSET, repr: Unset[bool] = UNSET, eq: Unset[bool] = UNSET, recursive: Unset[bool | Sequence[str]] = UNSET, handle_circular_refs: Unset[bool] = UNSET) -> Callable[[Type[T]], Type[T]]

Parameters:

Name Type Description Default
name Unset[str]

View name.

UNSET
attach Unset[bool]

If true, view will be attached to base cls.

UNSET
include Unset[Sequence[str]]

List of fields to include in view.

UNSET
exclude Unset[Sequence[str]]

List of fields to exclude from view.

UNSET
slots Unset[bool]

If true, __slots__ attribute will be generated and new class will be returned instead of the original one. If __slots__ is already defined in the class, then TypeError is raised. If UNSET value from base view model will be used.

UNSET
kw_only Unset[bool]

If kw_only is true, then by default all fields are keyword-only.

UNSET
env_prefix Unset[str | Sequence[str]]

Prefix(or list of prefixes) for environment variables. If UNSET value from base view model will be used.

UNSET
env_source Unset[Callable]

Environment variables source factory. If UNSET value from base view model will be used.

UNSET
validate Unset[bool]

Validate or not fields. If UNSET value from base view model will be used.

UNSET
add_disable_validation_to_init Unset[bool]

Add disable_validation keywoard argument to __init__() method to disable validation. If UNSET value from base view model will be used.

UNSET
extra Unset[Literal['ignore', 'forbid']]

Ignore or forbid extra arguments passed to init. If UNSET value from base view model will be used.

UNSET
repr Unset[bool]

If true, a __rich_repr__() method will be generated and rich.repr.auto decorator applied to the class. If UNSET value from base view model will be used.

UNSET
eq Unset[bool]

If true, an __eq__() method will be generated. This method compares the class as if it were a tuple of its fields, in order. Both instances in the comparison must be of the identical type. If UNSET value from base view model will be used.

UNSET
recursive Unset[bool | Sequence[str]]

...

UNSET
handle_circular_refs Unset[bool]

Handle or not circular refs. If UNSET value from base view model will be used.

UNSET
Source code in cwtch/cwtch.py
@dataclass_transform(field_specifiers=(field,))
def view(
    base_cls,
    name: Unset[str] = UNSET,
    *,
    attach: Unset[bool] = UNSET,
    include: Unset[Sequence[str]] = UNSET,
    exclude: Unset[Sequence[str]] = UNSET,
    slots: Unset[bool] = UNSET,
    kw_only: Unset[bool] = UNSET,
    env_prefix: Unset[str | Sequence[str]] = UNSET,
    env_source: Unset[Callable] = UNSET,
    validate: Unset[bool] = UNSET,
    add_disable_validation_to_init: Unset[bool] = UNSET,
    extra: Unset[Literal["ignore", "forbid"]] = UNSET,
    repr: Unset[bool] = UNSET,
    eq: Unset[bool] = UNSET,
    recursive: Unset[bool | Sequence[str]] = UNSET,
    handle_circular_refs: Unset[bool] = UNSET,
) -> Callable[[Type[T]], Type[T]]:
    """
    Args:
        name: View name.
        attach: If true, view will be attached to base cls.
        include: List of fields to include in view.
        exclude: List of fields to exclude from view.
        slots: If true, `__slots__` attribute will be generated
            and new class will be returned instead of the original one.
            If `__slots__` is already defined in the class, then TypeError is raised.
            If UNSET value from base view model will be used.
        kw_only: If kw_only is true, then by default all fields are keyword-only.
        env_prefix: Prefix(or list of prefixes) for environment variables.
            If UNSET value from base view model will be used.
        env_source: Environment variables source factory.
            If UNSET value from base view model will be used.
        validate: Validate or not fields.
            If UNSET value from base view model will be used.
        add_disable_validation_to_init: Add disable_validation keywoard argument to `__init__()` method
            to disable validation.
            If UNSET value from base view model will be used.
        extra: Ignore or forbid extra arguments passed to init.
            If UNSET value from base view model will be used.
        repr: If true, a `__rich_repr__()` method will be generated and rich.repr.auto decorator applied to the class.
            If UNSET value from base view model will be used.
        eq: If true, an `__eq__()` method will be generated.
            This method compares the class as if it were a tuple of its fields, in order.
            Both instances in the comparison must be of the identical type.
            If UNSET value from base view model will be used.
        recursive: ...
        handle_circular_refs: Handle or not circular refs.
            If UNSET value from base view model will be used.
    """

    if not (is_cwtch_model(base_cls) or is_cwtch_view(base_cls)):
        raise TypeError(f"{base_cls} is not a valid cwtch model or view")

    def wrapper(
        view_cls,
        *,
        base_cls=base_cls,
        name=name,
        attach=attach,
        include=include,
        exclude=exclude,
        slots=slots,
        kw_only=kw_only,
        env_prefix=env_prefix,
        env_source=env_source,
        validate=validate,
        add_disable_validation_to_init=add_disable_validation_to_init,
        extra=extra,
        repr=repr,
        eq=eq,
        recursive=recursive,
        handle_circular_refs=handle_circular_refs,
    ):
        if exclude and set(exclude) & view_cls.__annotations__.keys():  # type: ignore
            raise ValueError(f"unable to exclude fields {list(set(exclude) & view_cls.__annotations__.keys())}")  # type: ignore

        return _build_view(
            base_cls,
            view_cls,
            name,
            attach,
            include,
            exclude,
            slots,
            kw_only,
            cast(
                Unset[Sequence[str]],
                env_prefix if env_prefix is UNSET or isinstance(env_prefix, (list, tuple, str)) else [env_prefix],
            ),
            env_source,
            validate,
            add_disable_validation_to_init,
            extra,
            repr,
            eq,
            recursive,
            handle_circular_refs,
        )

    return wrapper

from_attributes(cls, obj: Any, data: dict | None = None, exclude: Sequence | None = None, suffix: str | None = None, reset_circular_refs: bool | None = None)

Build model from attributes of other object.

Parameters:

Name Type Description Default
obj Any

Object from which to build.

required
data dict | None

Additional data to build.

None
exclude Sequence | None

List of fields to exclude.

None
suffix str | None

Fields suffix.

None
reset_circular_refs bool | None

Reset circular references to None.

None
Source code in cwtch/cwtch.py
def from_attributes(
    cls,
    obj: Any,
    data: dict | None = None,
    exclude: Sequence | None = None,
    suffix: str | None = None,
    reset_circular_refs: bool | None = None,
):
    """
    Build model from attributes of other object.

    Args:
      obj: Object from which to build.
      data: Additional data to build.
      exclude: List of fields to exclude.
      suffix: Fields suffix.
      reset_circular_refs: Reset circular references to None.
    """

    kwds = {
        f.name: getattr(obj, f"{f_name}{suffix}" if suffix else f_name)
        for f_name, f in cls.__dataclass_fields__.items()
        if (not exclude or f_name not in exclude) and hasattr(obj, f"{f.name}{suffix}" if suffix else f_name)
    }
    if data:
        kwds.update(data)
    if exclude:
        kwds = {k: v for k, v in kwds.items() if k not in exclude}

    cache = _CACHE.get()
    cache["reset_circular_refs"] = reset_circular_refs
    try:
        return cls(__cwtch_cache_key=(cls, id(obj)), **kwds)
    finally:
        del cache["reset_circular_refs"]

asdict(inst, include: Sequence[str] | None = None, exclude: Sequence[str] | None = None, exclude_none: bool | None = None, exclude_unset: bool | None = None, context: dict | None = None) -> dict

Return cwtch model as dict.

Parameters:

Name Type Description Default
inst

cwtch model.

required
include Sequence[str] | None

List of field names to include.

None
exclude Sequence[str] | None

List of field names to exclude.

None
exclude_none bool | None

If true, fields with None value will be excluded.

None
exclude_unset bool | None

If true, unset fields will be excluded.

None
context dict | None

If specified, must be a mapping.

None
Source code in cwtch/cwtch.py
def asdict(
    inst,
    include: Sequence[str] | None = None,
    exclude: Sequence[str] | None = None,
    exclude_none: bool | None = None,
    exclude_unset: bool | None = None,
    context: dict | None = None,
) -> dict:
    """
    Return cwtch model as dict.

    Args:
        inst: cwtch model.
        include: List of field names to include.
        exclude: List of field names to exclude.
        exclude_none: If true, fields with None value will be excluded.
        exclude_unset: If true, unset fields will be excluded.
        context: If specified, must be a mapping.
    """

    return _asdict(
        inst,
        include_=include,
        exclude_=exclude,
        exclude_none=exclude_none,
        exclude_unset=exclude_unset,
        context=context,
    )

dumps_json(inst, encoder: Callable[[Any], Any] | None = None, context: dict | None = None)

Dump cwtch model to json.

Parameters:

Name Type Description Default
encoder Callable[[Any], Any] | None

Custom JSON encoder as callable.

None
context dict | None

If specified, must be a mapping.

None
Source code in cwtch/cwtch.py
def dumps_json(inst, encoder: Callable[[Any], Any] | None = None, context: dict | None = None):
    """
    Dump cwtch model to json.

    Args:
        encoder: Custom JSON encoder as callable.
        context: If specified, must be a mapping.
    """

    return _dumps_json(inst, encoder, context)

validate_args(fn: Callable, args: tuple, kwds: dict) -> tuple[tuple, dict]

Helper to convert and validate function arguments.

Parameters:

Name Type Description Default
args tuple

function positional arguments.

required
kwds dict

function keyword arguments.

required
Source code in cwtch/cwtch.py
def validate_args(fn: Callable, args: tuple, kwds: dict) -> tuple[tuple, dict]:
    """
    Helper to convert and validate function arguments.

    Args:
      args: function positional arguments.
      kwds: function keyword arguments.
    """

    annotations = {k: v.annotation for k, v in signature(fn).parameters.items()}

    validated_args = []
    for v, (arg_name, T) in zip(args, annotations.items()):
        if T != _empty:
            try:
                validated_args.append(validate_value(v, T))
            except ValidationError as e:
                raise TypeError(f"{fn.__name__}() expects {T} for argument {arg_name}") from e
        else:
            validated_args.append(v)

    validated_kwds = {}
    for arg_name, v in kwds.items():
        T = annotations[arg_name]
        if T != _empty:
            try:
                validated_kwds[arg_name] = validate_value(v, T)
            except ValidationError as e:
                raise TypeError(f"{fn.__name__}() expects {T} for argument {arg_name}") from e
        else:
            validated_kwds[arg_name] = v

    return tuple(validated_args), validated_kwds

validate_call(fn)

Decorator to convert and validate function arguments.

Source code in cwtch/cwtch.py
def validate_call(fn):
    """Decorator to convert and validate function arguments."""

    def wrapper(*args, **kwds):
        validate_args(fn, args, kwds)
        return fn(*args, **kwds)

    return wrapper

cwtch.metadata

Validator

Bases: TypeMetadata

Validator object.

Attributes:

Name Type Description
json_schema dict

Additional custom JSON schema.

before Callable

Validator to validate input data before base validation.

after Callable

Validator to validate value after base validation.

Source code in cwtch/metadata.py
@typing.final
@dataclass(slots=True)
class Validator(TypeMetadata):
    """Validator object.

    Attributes:
        json_schema: Additional custom JSON schema.
        before: Validator to validate input data before base validation.
        after: Validator to validate value after base validation.
    """

    json_schema: dict = field(default_factory=dict, repr=False)  # type: ignore
    before: typing.Callable = field(default=nop, kw_only=True)
    after: typing.Callable = field(default=nop, kw_only=True)

    def __init_subclass__(cls, **kwds):
        raise TypeError("Validator class cannot be inherited")

Ge

Bases: TypeMetadata

Validator to check that the input data is greater than or equal to the specified value.

Example:

Annotated[int, Ge(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Ge(TypeMetadata):
    """
    Validator to check that the input data is greater than or equal to the specified value.

    Example:

        Annotated[int, Ge(1)]
    """

    value: Any

    def json_schema(self) -> dict:
        return {"minimum": self.value}

    def after(self, value, /):
        if value < self.value:
            raise ValueError(f"value should be >= {self.value}")
        return value

Gt

Bases: TypeMetadata

Validator to check if input is greater than specified value.

Example:

Annotated[int, Gt(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Gt(TypeMetadata):
    """
    Validator to check if input is greater than specified value.

    Example:

        Annotated[int, Gt(1)]
    """

    value: Any

    def json_schema(self) -> dict:
        return {"minimum": self.value, "exclusiveMinimum": True}

    def after(self, value, /):
        if value <= self.value:
            raise ValueError(f"value should be > {self.value}")
        return value

Le

Bases: TypeMetadata

Validator to check that the input data is less than or equal to the specified value.

Example:

Annotated[int, Le(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Le(TypeMetadata):
    """
    Validator to check that the input data is less than or equal to the specified value.

    Example:

        Annotated[int, Le(1)]
    """

    value: Any

    def json_schema(self) -> dict:
        return {"maximum": self.value}

    def after(self, value, /):
        if value > self.value:
            raise ValueError(f"value should be <= {self.value}")
        return value

Lt

Bases: TypeMetadata

Validator to check if input is less than specified value.

Example:

Annotated[int, Lt(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Lt(TypeMetadata):
    """
    Validator to check if input is less than specified value.

    Example:

        Annotated[int, Lt(1)]
    """

    value: Any

    def json_schema(self) -> dict:
        return {"maximum": self.value, "exclusiveMaximum": True}

    def after(self, value, /):
        if value >= self.value:
            raise ValueError(f"value should be < {self.value}")
        return value

MinLen

Bases: TypeMetadata

Validator to check that the length of the input data is greater than or equal to the specified value.

Example:

Annotated[str, MinLen(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class MinLen(TypeMetadata):
    """
    Validator to check that the length of the input data is greater than or equal to the specified value.

    Example:

        Annotated[str, MinLen(1)]
    """

    value: int

    def json_schema(self) -> dict:
        return {"minLength": self.value}

    def after(self, value, /):
        if len(value) < self.value:
            raise ValueError(f"value length should be >= {self.value}")
        return value

MaxLen

Bases: TypeMetadata

Validator to check that the length of the input data is less than or equal to the specified value.

Example:

Annotated[str, MaxLen(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class MaxLen(TypeMetadata):
    """
    Validator to check that the length of the input data is less than or equal to the specified value.

    Example:

        Annotated[str, MaxLen(1)]
    """

    value: int

    def json_schema(self) -> dict:
        return {"maxLength": self.value}

    def after(self, value, /):
        if len(value) > self.value:
            raise ValueError(f"value length should be <= {self.value}")
        return value

Len

Bases: TypeMetadata

Validator to check that the input length is within the specified range.

Example:

Annotated[str, Len(1, 10)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Len(TypeMetadata):
    """
    Validator to check that the input length is within the specified range.

    Example:

        Annotated[str, Len(1, 10)]
    """

    min_value: int
    max_value: int

    def json_schema(self) -> dict:
        return {"minLength": self.min_value, "maxLength": self.max_value}

    def after(self, value, /):
        if len(value) < self.min_value:
            raise ValueError(f"value length should be >= {self.min_value}")
        if len(value) > self.max_value:
            raise ValueError(f"value length should be  {self.max_value}")
        return value

MinItems

Bases: TypeMetadata

Validator to check that the number of elements in the input is greater than or equal to the specified value.

Example:

Annotated[list, MinItems(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class MinItems(TypeMetadata):
    """
    Validator to check that the number of elements in the input is greater than or equal to the specified value.

    Example:

        Annotated[list, MinItems(1)]
    """

    value: int

    def json_schema(self) -> dict:
        return {"minItems": self.value}

    def after(self, value, /):
        if len(value) < self.value:
            raise ValueError(f"items count should be >= {self.value}")
        return value

MaxItems

Bases: TypeMetadata

Validator to check that the number of elements in the input is less than or equal to the specified value.

Example:

Annotated[list, MaxItems(1)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class MaxItems(TypeMetadata):
    """
    Validator to check that the number of elements in the input is less than or equal to the specified value.

    Example:

        Annotated[list, MaxItems(1)]
    """

    value: int

    def json_schema(self) -> dict:
        return {"maxItems": self.value}

    def after(self, value, /):
        if len(value) > self.value:
            raise ValueError(f"items count should be <= {self.value}")
        return value

Match

Bases: TypeMetadata

Validator to check that an input value matches a regular expression.

Example:

Annotated[str, Match(r".*")]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Match(TypeMetadata):
    """
    Validator to check that an input value matches a regular expression.

    Example:

        Annotated[str, Match(r".*")]
    """

    pattern: re.Pattern

    def json_schema(self) -> dict:
        return {"pattern": self.pattern.pattern}

    def after(self, value: str, /):
        if not self.pattern.match(value):
            raise ValueError(f"value doesn't match pattern {self.pattern}")
        return value

UrlConstraints

Bases: TypeMetadata

URL constraints.

Attributes:

Name Type Description
schemes list[str] | None

List of valid schemes.

ports list[int] | None

list of valid ports.

Example:

Annotated[Url, UrlConstraints(schemes=["http", "https"])]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class UrlConstraints(TypeMetadata):
    """
    URL constraints.

    Attributes:
        schemes: List of valid schemes.
        ports: list of valid ports.


    Example:

        Annotated[Url, UrlConstraints(schemes=["http", "https"])]
    """

    schemes: list[str] | None = field(default=None, kw_only=True)
    ports: list[int] | None = field(default=None, kw_only=True)

    def after(self, value, /):
        if self.schemes is not None and value.scheme not in self.schemes:
            raise ValueError(f"URL scheme should be one of {self.schemes}")
        if self.ports is not None and value.port is not None and value.port not in self.ports:
            raise ValueError(f"port number should be one of {self.ports}")
        return value

    def __hash__(self):
        return hash(f"{sorted(self.schemes or [])}{sorted(self.ports or [])}")

JsonLoads

Bases: TypeMetadata

Validator to try load value from json.

Example:

Annotated[list[int], JsonLoads()]
Source code in cwtch/metadata.py
@dataclass(slots=True, repr=False)
class JsonLoads(TypeMetadata):
    """
    Validator to try load value from json.

    Example:

        Annotated[list[int], JsonLoads()]
    """

    def before(self, value, /):
        try:
            return orjson.loads(value)
        except orjson.JSONDecodeError:
            return value

ToLower

Bases: TypeMetadata

Convert input to lower case.

Attributes:

Name Type Description
mode Literal['before', 'after']

Validation mode, before or after base validation. Default: after.

Example:

Annotated[str, ToLower()]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class ToLower(TypeMetadata):
    """
    Convert input to lower case.

    Attributes:
        mode: Validation mode, before or after base validation. Default: after.

    Example:

        Annotated[str, ToLower()]
    """

    mode: Literal["before", "after"] = "after"

    def before(self, value, /):
        if self.mode == "before":
            return value.lower()
        return value

    def after(self, value, /):
        if self.mode == "after":
            return value.lower()
        return value

ToUpper

Bases: TypeMetadata

Convert input to upper case.

Attributes:

Name Type Description
mode Literal['before', 'after']

Validation mode, before or after base validation. Default: after.

Example:

Annotated[str, ToUpper()]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class ToUpper(TypeMetadata):
    """
    Convert input to upper case.

    Attributes:
        mode: Validation mode, before or after base validation. Default: after.

    Example:

        Annotated[str, ToUpper()]
    """

    mode: Literal["before", "after"] = "after"

    def before(self, value, /):
        if self.mode == "before":
            return value.upper()
        return value

    def after(self, value, /):
        if self.mode == "after":
            return value.upper()
        return value

Strict

Bases: TypeMetadata

Validator to strict input type.

Example:

Annotated[int, Strict(int)]
Source code in cwtch/metadata.py
@dataclass(slots=True)
class Strict(TypeMetadata):
    """
    Validator to strict input type.

    Example:

        Annotated[int, Strict(int)]
    """

    type: Type

    def __post_init__(self):
        def fn(tp):
            tps = []
            if __args__ := getattr(tp, "__args__", None):
                if tp.__class__ not in [types.UnionType, typing._UnionGenericAlias]:  # type: ignore
                    raise ValueError(f"{self.type} is unsupported by {self.__class__}")
                for arg in __args__:
                    tps.extend(fn(arg))
            else:
                tps.append(tp)
            return tps

        object.__setattr__(self, "type", fn(self.type))

    def __hash__(self):
        return hash(f"{self.type}")

    def before(self, value, /):
        for tp in typing.cast(list, self.type):
            if isinstance(value, tp) and type(value) == tp:  # noqa: E721
                return value
        raise ValueError(f"invalid value for {' | '.join(map(str, typing.cast(list, self.type)))}")

EmailValidator

Bases: TypeMetadata

Email address validator.

Source code in cwtch/metadata.py
@dataclass(slots=True)
class EmailValidator(TypeMetadata):
    """Email address validator."""

    validator: emval.EmailValidator = field(
        default_factory=lambda: emval.EmailValidator(
            allow_smtputf8=True,
            allow_empty_local=True,
            allow_quoted_local=True,
            allow_domain_literal=True,
            deliverable_address=False,
        )
    )

    def json_schema(self) -> dict:
        return {"format": "email"}

    def after(self, value, /):
        return self.validator.validate_email(value)

cwtch.types

UNSET = UNSET module

Type to mark value as unset.

Number = int | float module-attribute

Number type. Example:

i: Number = 1
f: Number = 1.1

Positive = Annotated[T, Ge(1)] module-attribute

Positive type(Generic). Example:

i: Positive[int] = 1
n: Positive[Number] = 1.1

NonNegative = Annotated[T, Ge(0)] module-attribute

Non negative type(Generict). Example:

i: NonNegative[int] = 0
n: NonNegative[Number] = 0.0

NonEmpty = Annotated[T, MinItems(1)] module-attribute

Non empty container. Example:

l: NonEmpty[list] = [1]

NonZeroLen = Annotated[T, MinLen(1)] module-attribute

Non zero length object. Example:

l: NonZeroLen[str] = "a"

LowerStr = Annotated[str, ToLower()] module-attribute

Lower case string. Example:

s: LowerStr = "a"

UpperStr = Annotated[str, ToUpper()] module-attribute

Upper case string. Example:

s: LowerStr = "A"

StrictInt = Annotated[int, Strict(int)] module-attribute

Strict integer. Example:

i: StrictInt = 1

StrictFloat = Annotated[float, Strict(float)] module-attribute

Strict float. Example:

f: StrictFloat = 1.1

StrictNumber = StrictInt | StrictFloat module-attribute

Strict number. Example:

n: StrictNumber = 1.1

StrictStr = Annotated[str, Strict(str)] module-attribute

Strict string. Example:

s: StrictString = "a"

StrictBool = Annotated[bool, Strict(bool)] module-attribute

Strict bool. Example:

b: StrictBool = True

HttpUrl = Annotated[Url, UrlConstraints(shemes=['http', 'https'])] module-attribute

Type for HTTP URL.

SecretHttpUrl = Annotated[SecretUrl, UrlConstraints(shemes=['http', 'https'])] module-attribute

Type for secret HTTP URL.

WebsocketpUrl = Annotated[Url, UrlConstraints(shemes=['ws', 'wss'])] module-attribute

Type for websocket URL.

FileUrl = Annotated[Url, UrlConstraints(shemes=['file'])] module-attribute

Type for file URL.

FtpUrl = Annotated[Url, UrlConstraints(shemes=['ftp'])] module-attribute

Type for FTP URL.

SecretBytes

Bases: bytes

Type to represent secret bytes.

Source code in cwtch/types.py
class SecretBytes(bytes):
    """Type to represent secret bytes."""

    def __new__(cls, value):
        obj = super().__new__(cls, b"***")
        obj._value = value
        return obj

    def __repr__(self):
        return f"{self.__class__.__name__}(***)"

    def __hash__(self):
        return hash(self._value)

    def __eq__(self, other):
        if not isinstance(other, SecretBytes):
            return False
        return self._value == other._value

    def __ne__(self, other):
        if not isinstance(other, SecretBytes):
            return True
        return self._value != other._value

    def __len__(self):
        return len(self._value)

    # @classmethod
    # def __cwtch_json_schema__(cls, **kwds) -> dict:
    #     return {"type": "string"}

    def __cwtch_asdict__(self, handler, kwds: AsDictKwds):
        if (kwds.context or {}).get("show_secrets"):
            return self.get_secret_value()
        return self

    def get_secret_value(self) -> bytes:
        return self._value

SecretStr

Bases: str

Type to represent secret string.

Source code in cwtch/types.py
class SecretStr(str):
    """Type to represent secret string."""

    __slots__ = ("_value",)

    def __new__(cls, value):
        obj = super().__new__(cls, "***")
        obj._value = value
        return obj

    def __repr__(self):
        return f"{self.__class__.__name__}(***)"

    def __hash__(self):
        return hash(self._value)

    def __eq__(self, other):
        if not isinstance(other, SecretStr):
            return False
        return self._value == other._value

    def __ne__(self, other):
        if not isinstance(other, SecretStr):
            return True
        return self._value != other._value

    def __len__(self):
        return len(self._value)

    @classmethod
    def __cwtch_json_schema__(cls, **kwds) -> dict:
        return {"type": "string"}

    def __cwtch_asdict__(self, handler, kwds: AsDictKwds):
        if (kwds.context or {}).get("show_secrets"):
            return self.get_secret_value()
        return self

    def __cwtch_asjson__(self, context: dict | None = None):
        if (context or {}).get("show_secrets"):
            return self.get_secret_value()
        return f"{self}"

    def get_secret_value(self) -> str:
        return self._value

Url

Bases: str, _UrlMixIn

Type to represent URL.

Source code in cwtch/types.py
class Url(str, _UrlMixIn):
    """Type to represent URL."""

    __slots__ = ("_parsed",)

    def __new__(cls, value):
        try:
            parsed = urlparse(value)
        except Exception as e:
            raise ValueError(e)
        if parsed.hostname:
            _validate_hostname(parsed.hostname)

        obj = super().__new__(cls, parsed.geturl())
        obj._parsed = parsed
        return obj

    def __repr__(self):
        return f"{self.__class__.__name__}({self})"

    @classmethod
    def __cwtch_json_schema__(cls, **kwds) -> dict:
        return {"type": "string", "format": "uri"}

    def __cwtch_asjson__(self, context: dict | None = None):
        return f"{self}"

SecretUrl

Bases: str, _UrlMixIn

Type to represent secret URL.

Source code in cwtch/types.py
class SecretUrl(str, _UrlMixIn):
    """Type to represent secret URL."""

    __slots__ = ("_parsed", "_value")

    def __new__(cls, value):
        try:
            parsed = urlparse(value)
        except Exception as e:
            raise ValueError(e)
        if parsed.hostname:
            _validate_hostname(parsed.hostname)

        obj = super().__new__(
            cls,
            (
                parsed._replace(
                    netloc=f"***:***@{parsed.hostname}" + (f":{parsed.port}" if parsed.port is not None else "")
                ).geturl()
                if parsed.scheme
                else parsed.geturl()
            ),
        )
        obj._parsed = parsed
        obj._value = parsed.geturl()
        return obj

    def __repr__(self):
        parsed = self._parsed
        value = (
            parsed._replace(
                netloc=f"***:***@{parsed.hostname}" + (f":{parsed.port}" if parsed.port is not None else "")
            ).geturl()
            if parsed.scheme
            else parsed.geturl()
        )
        return f"{self.__class__.__name__}({value})"

    def __hash__(self):
        return hash(self._value)

    def __eq__(self, other):
        if not isinstance(other, SecretUrl):
            return False
        return self._value == other._value

    def __ne__(self, other):
        if not isinstance(other, SecretUrl):
            return True
        return self._value != other._value

    def __len__(self):
        return len(self._value)

    @property
    def username(self):
        return "***" if self._parsed.username else None

    @property
    def password(self):
        return "***" if self._parsed.password else None

    @classmethod
    def __cwtch_json_schema__(cls, **kwds) -> dict:
        return {"type": "string", "format": "uri"}

    def __cwtch_asdict__(self, handler, kwds: AsDictKwds):
        if (kwds.context or {}).get("show_secrets"):
            return self.get_secret_value()
        return self

    def __cwtch_asjson__(self, context: dict | None = None):
        if (context or {}).get("show_secrets"):
            return self.get_secret_value()
        return f"{self}"

    def get_secret_value(self) -> str:
        return self._value