Selectors

plateforme.core.selectors

This module provides utilities for managing selectors used in resources within the Plateforme framework.

SelectorType module-attribute

SelectorType = Type['BaseSelector[BaseResource]']

A type alias for a selector class.

SelectorConfig

SelectorConfig(
    __owner: Any | None = None,
    __defaults: dict[str, Any] | None = None,
    __partial_init: bool = False,
    /,
    **data: Any,
)

Bases: ConfigWrapper

A selector configuration.

Initialize the configuration class with the given data.

Parameters:

Name	Type	Description	Default
__owner	Any | None	The owner of the configuration instance. It can be any object or type that uses the configuration instance. Defaults to None.	None
__defaults	dict[str, Any] | None	The default values to initialize the configuration instance with. Defaults to None.	None
__partial_init	bool	Flag indicating whether to partially initialize the configuration instance. Defaults to False.	False
**data	Any	The data as keyword arguments to initialize the configuration instance with.	{}

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def __init__(
    self,
    __owner: Any | None = None,
    __defaults: dict[str, Any] | None = None,
    __partial_init: bool = False,
    /,
    **data: Any,
) -> None:
    """Initialize the configuration class with the given data.

    Args:
        __owner: The owner of the configuration instance. It can be any
            object or type that uses the configuration instance.
            Defaults to ``None``.
        __defaults: The default values to initialize the configuration
            instance with. Defaults to ``None``.
        __partial_init: Flag indicating whether  to partially initialize
            the configuration instance. Defaults to ``False``.
        **data: The data as keyword arguments to initialize the
            configuration instance with.
    """
    # Initialize configuration instance
    self.__config_owner__ = __owner \
        or getattr(self, '__config_owner__', None)
    self.__config_defaults__: dict[str, Any] = __defaults or {}
    with self.context(allow_mutation=True):
        self.update(data)

    # Validate configuration instance
    if not __partial_init:
        self.validate()

type_ class-attribute instance-attribute

type_: str = ConfigField(
    default="selector", frozen=True, init=False
)

The configuration owner type set to selector. It is a protected field that is typically used with check_config to validate an object type without using isinstance in order to avoid circular imports.

collection class-attribute instance-attribute

collection: bool | None = ConfigField(
    default=None, frozen=True
)

Whether the selection should return a collection or a single instance of the underlying resource class. Defaults to None.

create classmethod

create(
    owner: Any | None = None,
    defaults: dict[str, Any] | None = None,
    partial_init: bool = False,
    *,
    data: dict[str, Any] | None = None,
) -> Self

Create a new configuration instance.

This method is typically used internally to create a new configuration class with a specific owner and partial initialization flag. It is an alternative to the __init__ method for creating a new configuration instance.

Parameters:

Name	Type	Description	Default
owner	Any | None	The owner of the configuration instance.	None
defaults	dict[str, Any] | None	The default values to initialize the configuration instance with. Defaults to None.	None
partial_init	bool	Flag indicating whether to partially initialize the configuration instance. Defaults to False.	False
data	dict[str, Any] | None	The data to initialize the configuration instance with. Defaults to None.	None

Returns:

Type	Description
Self	The new configuration instance created.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

@classmethod
def create(
    cls,
    owner: Any | None = None,
    defaults: dict[str, Any] | None = None,
    partial_init: bool = False,
    *,
    data: dict[str, Any] | None = None,
) -> Self:
    """Create a new configuration instance.

    This method is typically used internally to create a new configuration
    class with a specific owner and partial initialization flag. It is an
    alternative to the `__init__` method for creating a new configuration
    instance.

    Args:
        owner: The owner of the configuration instance.
        defaults: The default values to initialize the configuration
            instance with. Defaults to ``None``.
        partial_init: Flag indicating whether to partially initialize the
            configuration instance. Defaults to ``False``.
        data: The data to initialize the configuration instance with.
            Defaults to ``None``.

    Returns:
        The new configuration instance created.
    """
    return cls(owner, defaults, partial_init, **(data or {}))

from_meta classmethod

from_meta(
    owner: type[Any],
    bases: tuple[type, ...],
    namespace: dict[str, Any],
    /,
    config_attr: str = "__config__",
    partial_init: bool = False,
    data: dict[str, Any] | None = None,
) -> Self

Create a new configuration instance from a class constructor.

This method is typically used internally to create a new configuration class from the meta configuration of a model, package, resource, or service. It merges the configuration of the given bases, the namespace, and the keyword arguments to create a new configuration class.

Parameters:

Name	Type	Description	Default
owner	type[Any]	The owner of the configuration instance. It should be the class that is being created from the meta configuration.	required
bases	tuple[type, ...]	The configurable base classes to merge.	required
namespace	dict[str, Any]	The configurable namespace to merge.	required
config_attr	str	The configurable attribute name used to extract the configuration dictionary from the bases and the namespace of the configurable class. Defaults to '__config__'.	'__config__'
partial_init	bool	Flag indicating whether to partially initialize the configuration instance. Defaults to False.	False
data	dict[str, Any] | None	The data to initialize the configuration instance with. Defaults to None.	None

Returns:

Type	Description
Self	The new configuration instance created from the given meta
Self	configuration.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

@classmethod
def from_meta(
    cls,
    owner: type[Any],
    bases: tuple[type, ...],
    namespace: dict[str, Any],
    /,
    config_attr: str = '__config__',
    partial_init: bool = False,
    data: dict[str, Any] | None = None,
) -> Self:
    """Create a new configuration instance from a class constructor.

    This method is typically used internally to create a new configuration
    class from the meta configuration of a model, package, resource, or
    service. It merges the configuration of the given bases, the namespace,
    and the keyword arguments to create a new configuration class.

    Args:
        owner: The owner of the configuration instance. It should be the
            class that is being created from the meta configuration.
        bases: The configurable base classes to merge.
        namespace: The configurable namespace to merge.
        config_attr: The configurable attribute name used to extract the
            configuration dictionary from the bases and the namespace of
            the configurable class. Defaults to ``'__config__'``.
        partial_init: Flag indicating whether to partially initialize the
            configuration instance. Defaults to ``False``.
        data: The data to initialize the configuration instance with.
            Defaults to ``None``.

    Returns:
        The new configuration instance created from the given meta
        configuration.
    """
    with cls.context(allow_mutation=True):
        # Build configuration instance
        config = cls(owner, {}, True)

        for base in bases:
            base_config = getattr(base, config_attr, {})
            if callable(base_config):
                base_config = base_config()
            config.merge(base_config)

        namespace_config = namespace.get(config_attr, {})
        if callable(namespace_config):
            namespace_config = namespace_config()
        config.merge(namespace_config)

        config.merge(data or {})

        # Validate configuration instance
        if not partial_init:
            config.validate()

    return config

validate

validate(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> None

Validate the configuration instance.

It post-initializes the configuration instance, checks for any missing required fields and validates the assignments of the configuration values based on the configuration fields information and the current validation context. This is performed automatically upon initialization of the configuration instance.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Raises:

Type	Description
ValueError	If the configuration instance has undefined values for required fields.
ValidationError	If the assignment of a value is invalid based on the configuration fields information and the current validation context.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def validate(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> None:
    """Validate the configuration instance.

    It post-initializes the configuration instance, checks for any missing
    required fields and validates the assignments of the configuration
    values based on the configuration fields information and the current
    validation context. This is performed automatically upon initialization
    of the configuration instance.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Raises:
        ValueError: If the configuration instance has undefined values for
            required fields.
        ValidationError: If the assignment of a value is invalid based on
            the configuration fields information and the current validation
            context.
    """
    # Perform post-initialization
    self.__post_init__()

    # Validate for missing required fields
    config_missing = [
        key for key, field in self.__config_fields__.items()
        if field.is_required() and self[key] is Undefined
    ]
    if config_missing:
        raise ValueError(
            f"Undefined values for configuration "
            f"{self.__class__.__qualname__!r} required fields: "
            f"{', '.join(config_missing)}."
        )

    # Validate assignments
    for key, value in self.entries(scope='set').items():
        if not strict and value is Deferred:
            continue
        if key in self.__config_validators__:
            validator = self.__config_validators__[key].validate_python
            value = validator(value, strict=strict, context=context)
        self.__dict__[key] = value

context staticmethod

context(
    *, allow_mutation: bool | None = None
) -> Iterator[bool]

Context manager for the configuration instance.

If the frozen mutation flag is not specified, the current frozen mutation flag is used if available, otherwise it defaults to False.

Parameters:

Name	Type	Description	Default
allow_mutation	bool | None	Flag indicating whether to allow frozen mutation of the configuration instance. When set to False, it prevents any changes by setting the frozen flag to True. If not specified, the current frozen mutation flag is used if available, otherwise it resolves to False. Defaults to None.	None

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

@staticmethod
@contextmanager
def context(
    *, allow_mutation: bool | None = None,
) -> Iterator[bool]:
    """Context manager for the configuration instance.

    If the frozen mutation flag is not specified, the current frozen
    mutation flag is used if available, otherwise it defaults to ``False``.

    Args:
        allow_mutation: Flag indicating whether to allow frozen mutation
            of the configuration instance. When set to ``False``, it
            prevents any changes by setting the frozen flag to ``True``.
            If not specified, the current frozen mutation flag is used if
            available, otherwise it resolves to ``False``.
            Defaults to ``None``.
    """
    context = FROZEN_CONTEXT.get()

    # Set frozen mutation flag
    if allow_mutation is None:
        if context is not None:
            frozen = context
        frozen = True
    else:
        frozen = not allow_mutation

    # Yield and reset frozen mutation flag
    token = FROZEN_CONTEXT.set(frozen)
    try:
        yield frozen
    finally:
        FROZEN_CONTEXT.reset(token)

clear

clear() -> None

Clear the configuration dictionary and reset all values.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def clear(self) -> None:
    """Clear the configuration dictionary and reset all values."""
    for key in self.__config_fields__:
        self.__dict__.pop(key, None)

copy

copy() -> Self

Return a shallow copy of the configuration dictionary.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def copy(self) -> Self:
    """Return a shallow copy of the configuration dictionary."""
    return self.__class__(
        self.__config_owner__,
        self.__config_defaults__.copy(),
        False,
        **self.entries(scope='set')
    )

merge

merge(
    *configs: Self | dict[str, Any],
    setdefault: bool = False,
) -> None

Merge the configuration with other configurations.

It merges the configuration with the provided configuration instances or dictionaries. The precedence of the rightmost configuration is higher than the leftmost configuration. This can be changed by setting the setdefault argument to True.

Parameters:

Name	Type	Description	Default
*configs	Self | dict[str, Any]	The configuration instances or dictionaries to merge with the target configuration dictionary.	()
setdefault	bool	Flag indicating whether to set the default values for the configuration keys if they are not already set. This modifies the behavior of the merge operation, making the precedence of the leftmost configuration higher than the rightmost configuration. Defaults to False (rightmost precedence).	False

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def merge(
    self,
    *configs: Self | dict[str, Any],
    setdefault: bool = False,
) -> None:
    """Merge the configuration with other configurations.

    It merges the configuration with the provided configuration instances
    or dictionaries. The precedence of the rightmost configuration is
    higher than the leftmost configuration. This can be changed by setting
    the `setdefault` argument to ``True``.

    Args:
        *configs: The configuration instances or dictionaries to merge with
            the target configuration dictionary.
        setdefault: Flag indicating whether to set the default values for
            the configuration keys if they are not already set.
            This modifies the behavior of the merge operation, making the
            precedence of the leftmost configuration higher than the
            rightmost configuration.
            Defaults to ``False`` (rightmost precedence).
    """
    for config in configs:
        # Retrieve the configuration dictionary
        if isinstance(config, self.__class__):
            config_dict = {
                key: value
                for key, value in config.__dict__.items()
                if key in config.__config_fields__
            }
        elif isinstance(config, dict):
            config_dict = config.copy()
        else:
            raise TypeError(
                f"Invalid configuration type {type(config).__name__!r} "
                f"for {self.__class__.__qualname__!r}, it must be a "
                f"dictionary or a configuration instance."
            )
        # Merge with the target configuration dictionary
        if setdefault:
            for key, value in config_dict.items():
                self.setdefault(key, value)
        else:
            self.update(config_dict)

check

check(
    key: str,
    *,
    scope: Literal["all", "default", "set"] = "all",
    raise_errors: bool = True,
) -> bool

Check if the configuration key exists in the given scope.

Parameters:

Name	Type	Description	Default
key	str	The configuration key to check for.	required
scope	Literal['all', 'default', 'set']	The scope to check for the configuration key. It can be either 'all' to check in all configuration entries, 'default' to check in configuration entries with default values not undefined, or 'set' to check in only the configuration entries that have been explicitly set. Defaults to 'all'.	'all'
raise_errors	bool	Flag indicating whether to raise an error if the configuration key is not defined for the configuration wrapper. Defaults to True.	True

Returns:

Type	Description
bool	A boolean indicating whether the configuration key exists in the
bool	specified scope.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def check(
    self,
    key: str,
    *,
    scope: Literal['all', 'default', 'set'] = 'all',
    raise_errors: bool = True,
) -> bool:
    """Check if the configuration key exists in the given scope.

    Args:
        key: The configuration key to check for.
        scope: The scope to check for the configuration key. It can be
            either ``'all'`` to check in all configuration entries,
            ``'default'`` to check in configuration entries with default
            values not undefined, or ``'set'`` to check in only the
            configuration entries that have been explicitly set.
            Defaults to ``'all'``.
        raise_errors: Flag indicating whether to raise an error if the
            configuration key is not defined for the configuration wrapper.
            Defaults to ``True``.

    Returns:
        A boolean indicating whether the configuration key exists in the
        specified scope.
    """
    if key not in self.__config_fields__:
        if not raise_errors:
            return False
        raise KeyError(
            f"Configuration key {key!r} cannot be checked as it is not "
            f"defined for wrapper {self.__class__.__qualname__!r}."
        )
    if scope == 'default':
        return key not in self.__dict__
    if scope == 'set':
        return key in self.__dict__
    return True

entries

entries(
    *,
    scope: Literal["all", "default", "set"] = "all",
    default_mode: Literal[
        "preserve", "unwrap", "wrap"
    ] = "unwrap",
    include_keys: Iterable[str] | None = None,
    exclude_keys: Iterable[str] | None = None,
    include_metadata: Iterable[Any] | None = None,
    exclude_metadata: Iterable[Any] | None = None,
) -> dict[str, Any]

Return the configuration dictionary.

It returns the configuration dictionary based on the specified scope, and keys and extra information to filter the configuration dictionary entries.

Parameters:

Name	Type	Description	Default
scope	Literal['all', 'default', 'set']	The scope of the configuration dictionary to return. It can be either 'all' to return all configuration entries, 'default' to return all configuration entries with their default values, or 'set' to return only the configuration entries that have been explicitly set. Defaults to 'all'.	'all'
default_mode	Literal['preserve', 'unwrap', 'wrap']	The default mode to use when returning a default entry from the configuration dictionary. It can be either 'preserve' to keep the default value as is, 'unwrap' to unwrap the default value, or 'wrap' to wrap the default value with a default placeholder. Defaults to 'unwrap'.	'unwrap'
include_keys	Iterable[str] | None	The keys to include from the configuration dictionary entries. Defaults to None.	None
exclude_keys	Iterable[str] | None	The keys to exclude from the configuration dictionary entries. Defaults to None.	None
include_metadata	Iterable[Any] | None	The metadata information to include from the configuration dictionary entries. Defaults to None.	None
exclude_metadata	Iterable[Any] | None	The metadata information to exclude from the configuration dictionary entries. Defaults to None.	None

Returns:

Type	Description
dict[str, Any]	A dictionary containing the configuration entries based on the
dict[str, Any]	specified scope and extra information.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def entries(
    self,
    *,
    scope: Literal['all', 'default', 'set'] = 'all',
    default_mode: Literal['preserve', 'unwrap', 'wrap'] = 'unwrap',
    include_keys: Iterable[str] | None = None,
    exclude_keys: Iterable[str] | None = None,
    include_metadata: Iterable[Any] | None = None,
    exclude_metadata: Iterable[Any] | None = None,
) -> dict[str, Any]:
    """Return the configuration dictionary.

    It returns the configuration dictionary based on the specified scope,
    and keys and extra information to filter the configuration dictionary
    entries.

    Args:
        scope: The scope of the configuration dictionary to return. It can
            be either ``'all'`` to return all configuration entries,
            ``'default'`` to return all configuration entries with their
            default values, or ``'set'`` to return only the configuration
            entries that have been explicitly set. Defaults to ``'all'``.
        default_mode: The default mode to use when returning a default
            entry from the configuration dictionary. It can be either
            ``'preserve'`` to keep the default value as is, ``'unwrap'`` to
            unwrap the default value, or ``'wrap'`` to wrap the default
            value with a default placeholder. Defaults to ``'unwrap'``.
        include_keys: The keys to include from the configuration dictionary
            entries. Defaults to ``None``.
        exclude_keys: The keys to exclude from the configuration dictionary
            entries. Defaults to ``None``.
        include_metadata: The metadata information to include from the
            configuration dictionary entries. Defaults to ``None``.
        exclude_metadata: The metadata information to exclude from the
            configuration dictionary entries. Defaults to ``None``.

    Returns:
        A dictionary containing the configuration entries based on the
        specified scope and extra information.
    """
    # Retrieve the configuration dictionary based on the specified scope
    config_dict: dict[str, Any] = {}
    if scope == 'default':
        for key in self.__config_fields__:
            if key in self.__dict__:
                continue
            config_dict[key] = self.get(key, default_mode=default_mode)
    elif scope == 'set':
        for key in self.__config_fields__:
            if key not in self.__dict__:
                continue
            config_dict[key] = self.get(key, default_mode=default_mode)
    else:
        for key in self.__config_fields__:
            config_dict[key] = self.get(key, default_mode=default_mode)

    # Build the keys and metadata sets to include and exclude from the
    # configuration dictionary entries.
    incex_keys = [include_keys, exclude_keys]
    for count, value in enumerate(incex_keys):
        if value is not None:
            if isinstance(value, type) and \
                    issubclass(value, ConfigWrapper):
                value = value.__config_fields__.keys()
            incex_keys[count] = set(value)

    incex_metadata = [include_metadata, exclude_metadata]
    for count, value in enumerate(incex_metadata):
        if value is not None:
            incex_metadata[count] = set(value)

    # Return directly if no keys or metadata filtering is provided.
    if not any(incex_keys) and not any(incex_metadata):
        return config_dict

    # Filter the configuration dictionary based on the information and keys
    # if provided in the "with" arguments.
    for key in list(config_dict.keys()):
        if incex_keys[0] is not None and key not in incex_keys[0]:
            config_dict.pop(key)
        elif incex_keys[1] is not None and key in incex_keys[1]:
            config_dict.pop(key)
        elif incex_metadata[0] is not None and not all(
            metadata in self.__config_fields__[key].metadata
            for metadata in incex_metadata[0]
        ):
            config_dict.pop(key)
        elif incex_metadata[1] is not None and any(
            metadata in self.__config_fields__[key].metadata
            for metadata in incex_metadata[1]
        ):
            config_dict.pop(key)

    return config_dict

keys

keys() -> KeysView[str]

Return the configuration keys.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def keys(self) -> KeysView[str]:
    """Return the configuration keys."""
    return KeysView(self.entries())

values

values() -> ValuesView[Any]

Return the configuration values.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def values(self) -> ValuesView[Any]:
    """Return the configuration values."""
    return ValuesView(self.entries())

items

items() -> ItemsView[str, Any]

Return the configuration items.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def items(self) -> ItemsView[str, Any]:
    """Return the configuration items."""
    return ItemsView(self.entries())

get

get(key: str) -> Any

get(key: str, default: Any) -> Any

get(
    key: str,
    default: Any = Undefined,
    *,
    default_mode: Literal[
        "preserve", "unwrap", "wrap"
    ] = "unwrap",
) -> Any

get(
    key: str,
    default: Any = Undefined,
    *,
    default_mode: Literal[
        "preserve", "unwrap", "wrap"
    ] = "unwrap",
) -> Any

Get the value for the specified key if set otherwise the default.

Parameters:

Name	Type	Description	Default
key	str	The configuration key to get the value for.	required
default	Any	The default value to return if the key is not set.	Undefined
default_mode	Literal['preserve', 'unwrap', 'wrap']	The default mode to use when returning a default value from the configuration dictionary. It can be either 'preserve' to keep the default value as is, 'unwrap' to unwrap the default value, or 'wrap' to wrap the default value with a placeholder. Defaults to 'unwrap'.	'unwrap'

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def get(
    self, key: str,
    default: Any = Undefined,
    *,
    default_mode: Literal['preserve', 'unwrap', 'wrap'] = 'unwrap',
) -> Any:
    """Get the value for the specified key if set otherwise the default.

    Args:
        key: The configuration key to get the value for.
        default: The default value to return if the key is not set.
        default_mode: The default mode to use when returning a default
            value from the configuration dictionary. It can be either
            ``'preserve'`` to keep the default value as is, ``'unwrap'`` to
            unwrap the default value, or ``'wrap'`` to wrap the default
            value with a placeholder. Defaults to ``'unwrap'``.
    """
    if key not in self.__config_fields__:
        raise KeyError(
            f"Configuration key {key!r} cannot be retrieved as it is not "
            f"defined for wrapper {self.__class__.__qualname__!r}."
        )
    if key in self.__dict__:
        return self.__dict__[key]
    if default is Undefined:
        return self.getdefault(key, mode=default_mode)
    if default_mode == 'wrap':
        return Default(default)
    if default_mode == 'unwrap':
        return get_value_or_default(default)
    return default

pop

pop(key: str) -> Any

pop(key: str, default: Any) -> Any

pop(
    key: str,
    default: Any = Undefined,
    *,
    default_mode: Literal[
        "preserve", "unwrap", "wrap"
    ] = "unwrap",
) -> Any

pop(
    key: str,
    default: Any = Undefined,
    *,
    default_mode: Literal[
        "preserve", "unwrap", "wrap"
    ] = "unwrap",
) -> Any

Pop the specified key if set and return its corresponding value.

Parameters:

Name	Type	Description	Default
key	str	The configuration key to pop the value for.	required
default	Any	The default value to return if the key is not set.	Undefined
default_mode	Literal['preserve', 'unwrap', 'wrap']	The default mode to use when returning a default value from the configuration dictionary. It can be either 'preserve' to keep the default value as is, 'unwrap' to unwrap the default value, or 'wrap' to wrap the default value with a placeholder. Defaults to 'unwrap'.	'unwrap'

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def pop(
    self,
    key: str,
    default: Any = Undefined,
    *,
    default_mode: Literal['preserve', 'unwrap', 'wrap'] = 'unwrap',
) -> Any:
    """Pop the specified key if set and return its corresponding value.

    Args:
        key: The configuration key to pop the value for.
        default: The default value to return if the key is not set.
        default_mode: The default mode to use when returning a default
            value from the configuration dictionary. It can be either
            ``'preserve'`` to keep the default value as is, ``'unwrap'`` to
            unwrap the default value, or ``'wrap'`` to wrap the default
            value with a placeholder. Defaults to ``'unwrap'``.
    """
    if key not in self.__config_fields__:
        raise KeyError(
            f"Configuration key {key!r} cannot be popped as it is not "
            f"defined for wrapper {self.__class__.__qualname__!r}."
        )
    if key in self.__dict__:
        return self.__dict__.pop(key)
    if default is Undefined:
        return self.getdefault(key, mode=default_mode)
    if default_mode == 'wrap':
        return Default(default)
    if default_mode == 'unwrap':
        return get_value_or_default(default)
    return default

getdefault

getdefault(
    key: str,
    *,
    mode: Literal["preserve", "unwrap", "wrap"] = "unwrap",
) -> Any

Get the default value for the specified key.

Parameters:

Name	Type	Description	Default
key	str	The configuration key to get the default value for.	required
mode	Literal['preserve', 'unwrap', 'wrap']	The mode to use when returning the default value. It can be either 'preserve' to keep the default value as is, 'unwrap' to unwrap the default value, or 'wrap' to wrap the default value with a placeholder. Defaults to 'unwrap'.	'unwrap'

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def getdefault(
    self,
    key: str,
    *,
    mode: Literal['preserve', 'unwrap', 'wrap'] = 'unwrap',
) -> Any:
    """Get the default value for the specified key.

    Args:
        key: The configuration key to get the default value for.
        mode: The mode to use when returning the default value. It can be
            either ``'preserve'`` to keep the default value as is,
            ``'unwrap'`` to unwrap the default value, or ``'wrap'`` to wrap
            the default value with a placeholder. Defaults to ``'unwrap'``.
    """
    if key not in self.__config_fields__:
        raise KeyError(
            f"Default for configuration key {key!r} cannot be retrieved "
            f"as it is not defined for wrapper "
            f"{self.__class__.__qualname__!r}."
        )
    elif key in self.__config_defaults__:
        default = self.__config_defaults__[key]
    else:
        default = self.__config_fields__[key].get_default()

    if mode == 'wrap':
        return Default(default)
    if mode == 'unwrap':
        return get_value_or_default(default)
    return default

setdefault

setdefault(key: str, default: Any) -> Any

Set the default value for the specified key.

Parameters:

Name	Type	Description	Default
key	str	The configuration key to set the default value for.	required
default	Any	The default value to set for the key.	required

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def setdefault(self, key: str, default: Any) -> Any:
    """Set the default value for the specified key.

    Args:
        key: The configuration key to set the default value for.
        default: The default value to set for the key.
    """
    if key not in self.__config_fields__:
        raise KeyError(
            f"Default for configuration key {key!r} cannot be set as it "
            f"is not defined for wrapper {self.__class__.__qualname__!r}."
        )
    self.__config_defaults__[key] = default
    return default

update

update(
    *args: tuple[str, Any] | Mapping[str, Any],
    **kwargs: Any,
) -> None

Update the config dictionary with new data.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def update(
    self,
    *args: tuple[str, Any] | Mapping[str, Any],
    **kwargs: Any,
) -> None:
    """Update the config dictionary with new data."""
    # Helper function to update configuration entries
    def update_entry(key: str, value: Any) -> None:
        if key not in self.__config_fields__:
            raise KeyError(
                f"Configuration key {key!r} cannot be updated as it is "
                f"not defined for wrapper {self.__class__.__qualname__!r}."
            )
        setattr(self, key, value)

    # Update args data
    for arg in args:
        if isinstance(arg, tuple):
            if len(arg) != 2:
                raise ValueError(
                    f"Configuration update arguments must be provided as "
                    f"key-value pairs. Got: {arg}."
                )
            update_entry(arg[0], arg[1])
        else:
            for key, value in arg.items():
                update_entry(key, value)
    # Update kwargs data
    for key, value in kwargs.items():
        update_entry(key, value)

SelectorMeta

Bases: ABCMeta, ConfigurableMeta

A metaclass for selector classes.

collect_config_wrappers

collect_config_wrappers(
    bases: tuple[type, ...], namespace: dict[str, Any]
) -> list[ConfigType]

Collect configuration wrappers from the given bases and namespace.

It collects the configuration wrappers from the given bases and namespace by extracting the configuration wrapper type from the original bases annotation if it is a generic subclass of the configurable class or metaclass, and from the configuration attribute if present in the class and bases namespace.

Parameters:

Name	Type	Description	Default
bases	tuple[type, ...]	The class bases.	required
namespace	dict[str, Any]	The class namespace.	required

Returns:

Type	Description
list[ConfigType]	A list of configuration wrapper classes found in the given bases
list[ConfigType]	and namespace.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/config.py

def collect_config_wrappers(
    cls, bases: tuple[type, ...], namespace: dict[str, Any], /,
) -> list[ConfigType]:
    """Collect configuration wrappers from the given bases and namespace.

    It collects the configuration wrappers from the given bases and
    namespace by extracting the configuration wrapper type from the
    original bases annotation if it is a generic subclass of the
    configurable class or metaclass, and from the configuration attribute
    if present in the class and bases namespace.

    Args:
        bases: The class bases.
        namespace: The class namespace.

    Returns:
        A list of configuration wrapper classes found in the given bases
        and namespace.
    """
    config_attr = getattr(cls, '__config_attr__', '__config__')
    config_wrappers: list[ConfigType] = []

    # The collection process is done in two steps to ensure that the
    # configuration attribute is extracted from the annotation first, and
    # then from the class and bases namespace.

    # Extract the configuration wrapper type from the annotation if it is a
    # generic subclass of the configurable class.
    meta_bases = get_meta_orig_bases(bases, namespace)
    for meta_base in meta_bases:
        origin = typing.get_origin(meta_base)
        if origin is None:
            continue
        if not isbaseclass_lenient(origin, 'Configurable'):
            continue
        args = typing.get_args(meta_base)
        if len(args) == 1:
            if isinstance(args[0], TypeVar):
                break
            if issubclass(args[0], ConfigWrapper):
                config_wrappers.append(args[0])
                break
        raise TypeError(
            f"Generic argument for the `Configurable` class must be a "
            f"subclass of the base configuration wrapper. Got: {args}."
        )

    # Extract the configuration wrapper type from the configuration
    # attribute if present in the class and bases namespace.
    meta_namespaces = get_meta_namespaces(bases, namespace)
    for meta_namespace in meta_namespaces:
        if config_attr not in meta_namespace:
            continue
        config_wrapper = meta_namespace[config_attr]
        if callable(config_wrapper):
            config_wrapper = config_wrapper()
        if isinstance(config_wrapper, ConfigWrapper):
            config_wrappers.append(config_wrapper.__class__)
            continue
        if isinstance(config_wrapper, dict):
            continue
        raise TypeError(
            f"Configuration attribute must be a dictionary or an "
            f"instance of the base configuration wrapper. Got: "
            f"{type(config_wrapper).__qualname__}."
        )

    return config_wrappers

collect_resources

collect_resources(
    bases: tuple[type, ...], namespace: dict[str, Any]
) -> list[ResourceType]

Collect selector resources from the given objects.

It collects the selector resources from the given bases and namespace by extracting the selector resource type from the original bases annotation if it is a generic subclass of the base selector class or metaclass, and from the resource configuration attribute if present in the class and bases namespace.

Parameters:

Name	Type	Description	Default
bases	tuple[type, ...]	The class bases.	required
namespace	dict[str, Any]	The class namespace.	required

Returns:

Type	Description
list[ResourceType]	A list of selector resource classes found in the given objects.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def collect_resources(
    cls, bases: tuple[type, ...], namespace: dict[str, Any], /,
) -> list['ResourceType']:
    """Collect selector resources from the given objects.

    It collects the selector resources from the given bases and namespace
    by extracting the selector resource type from the original bases
    annotation if it is a generic subclass of the base selector class or
    metaclass, and from the resource configuration attribute if present in
    the class and bases namespace.

    Args:
        bases: The class bases.
        namespace: The class namespace.

    Returns:
        A list of selector resource classes found in the given objects.
    """
    resources: list['ResourceType'] = []

    # The collection process is done in two steps to ensure that the
    # resource configuration attribute is extracted from the annotation
    # first, and then from the class and bases namespace.

    # Extract the selector resource class from the annotation if it is a
    # generic subclass of the selector class.
    meta_bases = get_meta_orig_bases(bases, namespace)
    for meta_base in meta_bases:
        origin = typing.get_origin(meta_base)
        if origin is None:
            continue
        if not isbaseclass_lenient(origin, 'BaseSelector'):
            continue
        resource, update = cls.parse_annotation(meta_base)
        if not update:
            if resource is not None:
                resources.append(resource)
            break
        raise TypeError(
            f"Invalid resource class for resource selector "
            f"{meta_base.__qualname__!r}. The resource class cannot have "
            f"additional metadata update when used with the base selector "
            f"class. Got: {update}."
        )

    # Extract the selector resource class from the resource configuration
    # attribute if present in the class and bases namespace.
    meta_namespaces = get_meta_namespaces(bases, namespace)
    for meta_namespace in meta_namespaces:
        if '__config_resource__' not in meta_namespace:
            continue
        resource = meta_namespace['__config_resource__']
        if resource is None:
            continue
        if is_resource(resource):
            resources.append(resource)
            continue
        raise TypeError(
            f"Invalid selector resource class found in the class and "
            f"bases namespace of {cls.__qualname__!r}. Got: {resource!r}."
        )

    return resources

parse_annotation staticmethod

parse_annotation(
    annotation: Any,
) -> tuple[ResourceType | None, dict[str, Any]]

Parse resource and metadata update from a selector annotation.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@staticmethod
def parse_annotation(
    annotation: Any
) -> tuple['ResourceType | None', dict[str, Any]]:
    """Parse resource and metadata update from a selector annotation."""
    # Extract annotation selector
    annotation_selector = typing.get_origin(annotation)
    if not issubclass(annotation_selector, BaseSelector):
        raise TypeError(
            f"Invalid annotation type for resource selector. "
            f"Got: {annotation!r}. "
        )

    # Extract annotation argument
    annotation_args = typing.get_args(annotation)
    if len(annotation_args) != 1:
        raise TypeError(
            f"Invalid annotation argument for resource selector. "
            f"Got: {annotation!r}. "
        )
    annotation_arg = annotation_args[0]

    # Parse annotation argument
    if is_annotated(annotation_arg):
        annotation_resource, *annotation_metadata = \
            typing.get_args(annotation_arg)
    else:
        annotation_resource = annotation_arg
        annotation_metadata = []

    # Validate annotation resource
    if isinstance(annotation_resource, TypeVar):
        annotation_resource = None
    elif not is_resource(annotation_resource):
        raise TypeError(
            f"Invalid selector resource class found in the annotation "
            f"{annotation!r}. The resource class must be a subclass of "
            f"`BaseResource` class. Got: {annotation_resource}."
        )

    # Validate annotation metadata update
    annotation_update = {}
    for update in annotation_metadata:
        if update is None:
            continue
        elif isinstance(update, str):
            annotation_update[update] = Deferred
        elif isinstance(update, dict):
            annotation_update.update(update)
        else:
            raise TypeError(
                f"Invalid metadata update type for resource selector "
                f"{annotation.__qualname__!r}. The metadata update must "
                f"be either a string or a dictionary. Got: {update}."
            )

    return annotation_resource, annotation_update

BaseSelector

BaseSelector(*args: Any, **kwargs: Any)

Bases: dict[str, Any], Configurable[SelectorConfig], Generic[_T]

A base selector class to uniquely identify resource instances.

This class is used as a type annotation to define a resource selector for a specific resource class. It is used as a proxy of the dict type to represent selection assignments for a specific resource type.

It can be inherited by applying a concrete selector resource class as a generic argument to the base selector class.

Note

The route auto-generation system will use this type hint to generate if needed the correct path parameter for the resource selector in the API routes.

Initialize a new selector.

It initializes a new resource selector with the provided assignment arguments and keyword arguments. The selector represents a dictionary of aliases and their associated values that identify resource instances.

To validate the selector against a resource class, the validate method should be called with a specific resource and collection instead of directly initializing the selector.

Parameters:

Name	Type	Description	Default
*args	Any	The provided selector assignments as a list of string arguments with the following format alias=value optionally separated by semicolons. If an argument is not of string type, it is assigned to the wildcard entry *.	()
**kwargs	Any	The provided selector assignments as a dictionary of assignment values with the aliases as keys.	{}

Note

Only one wildcard entry * is allowed within the selector assignments. The wildcard entry is used to match any field alias in a resource selector. If the wildcard entry is provided, it will be matched against the selectors of the resource when specified to determine the correct field alias.

A selector assignment value can be Deferred to represent an entry that should be inferred when initializing a resource instance. Those entries can be filtered out when collecting the selector assignment entries using the scope 'set'.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize a new selector.

    It initializes a new resource selector with the provided assignment
    arguments and keyword arguments. The selector represents a dictionary
    of aliases and their associated values that identify resource
    instances.

    To validate the selector against a resource class, the `validate`
    method should be called with a specific resource and collection instead
    of directly initializing the selector.

    Args:
        *args: The provided selector assignments as a list of string
            arguments with the following format ``alias=value`` optionally
            separated by semicolons. If an argument is not of string type,
            it is assigned to the wildcard entry ``*``.
        **kwargs: The provided selector assignments as a dictionary of
            assignment values with the aliases as keys.

    Note:
        Only one wildcard entry ``*`` is allowed within the selector
        assignments. The wildcard entry is used to match any field alias in
        a resource selector. If the wildcard entry is provided, it will
        be matched against the selectors of the resource when specified to
        determine the correct field alias.

        A selector assignment value can be `Deferred` to represent an
        entry that should be inferred when initializing a resource
        instance. Those entries can be filtered out when collecting the
        selector assignment entries using the scope ``'set'``.
    """
    # Collect all selector assignments from the given arguments
    for arg in args:
        # Parse selector
        if isinstance(arg, str):
            separator = r'[' + Symbol.STATEMENT + r']'
            selector = parse_selection(arg, separator=separator)
        else:
            selector = {WILDCARD: arg}
        # Validate assignments
        for alias, value in selector.items():
            if alias in self:
                raise KeyError(
                    f"Duplicate selector assignment aliases found for "
                    f"{alias}` from argument `{arg!r}."
                )
            self[alias] = value

    # Collect all selector assignments from the given keyword arguments
    for alias, value in kwargs.items():
        # Validate assignment alias
        alias = alias.lower()
        if not re.match(RegexPattern.ALIAS, alias):
            raise ValueError(
                f"Invalid selector assignment alias {alias!r} found in "
                f"keyword argument '{alias}={value}'. Aliases must match "
                f"a specific pattern `ALIAS` defined in the framework's "
                f"regular expressions repository."
            )
        # Validate assignment value
        if alias in self:
            raise KeyError(
                f"Duplicate selector assignment aliases found for "
                f"{alias!r} from keyword argument '{alias}={value}'."
            )
        self[alias] = value

resource property

resource: type[_T]

The resource associated with the selector.

It is used to validate the selection assignments against this resource class and to generate the correct path parameter for the resource selector in the API routes.

config

config() -> SelectorConfig

The selector configuration.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classproperty
def config(cls) -> SelectorConfig:
    """The selector configuration."""
    return cls.__config__

build_query

build_query(
    *,
    filter_criteria: Filter | None = None,
    sort_criteria: Sort | None = None,
    options: Sequence[ExecutableOption] | None = None,
    raise_errors: bool = True,
) -> Select[tuple[_T]]

Build a query for the selector.

Recursively builds a query for the selector by performing a join operation on the resource relationships found in the selector and filters the query based on the selector assignments. Optionally, the function can apply additional filter and sort criteria to the query.

Parameters:

Name	Type	Description	Default
filter_criteria	Filter | None	The filter criteria to apply to the query. Defaults to None.	None
sort_criteria	Sort | None	The sort criteria to apply to the query. Defaults to None.	None
options	Sequence[ExecutableOption] | None	The options to apply to the query. Defaults to None.	None
raise_errors	bool	Whether to raise errors when invalid attributes or values are found in the selector assignments. Defaults to True.	True

Returns:

Type	Description
Select[tuple[_T]]	The built selector query with the applied filter and sort criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def build_query(
    self,
    *,
    filter_criteria: Filter | None = None,
    sort_criteria: Sort | None = None,
    options: Sequence[ExecutableOption] | None = None,
    raise_errors: bool = True,
) -> Select[tuple[_T]]:
    """Build a query for the selector.

    Recursively builds a query for the selector by performing a join
    operation on the resource relationships found in the selector and
    filters the query based on the selector assignments. Optionally, the
    function can apply additional filter and sort criteria to the query.

    Args:
        filter_criteria: The filter criteria to apply to the query.
            Defaults to ``None``.
        sort_criteria: The sort criteria to apply to the query.
            Defaults to ``None``.
        options: The options to apply to the query.
            Defaults to ``None``.
        raise_errors: Whether to raise errors when invalid attributes or
            values are found in the selector assignments.
            Defaults to ``True``.

    Returns:
        The built selector query with the applied filter and sort criteria.
    """
    return build_query(
        self.resource,
        selection=self,
        filter_criteria=filter_criteria,
        sort_criteria=sort_criteria,
        raise_errors=raise_errors,
        options=options,
    )

copy

copy() -> Self

Return a shallow copy of the selector.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def copy(self) -> Self:
    """Return a shallow copy of the selector."""
    return copy(self)

entries

entries(
    *, scope: Literal["all", "deferred", "set"] = "all"
) -> dict[str, Any]

Return the selector assignments.

It returns the assignments dictionary based on the specified scope.

Parameters:

Name	Type	Description	Default
scope	Literal['all', 'deferred', 'set']	The scope of the selector assignments to return. It can be either 'all' to return all selector assignments entries, 'deferred' to return all deferred assignments entries, or 'set' to return only the assignments entries that have been explicitly set. Defaults to 'all'.	'all'

Returns:

Type	Description
dict[str, Any]	A dictionary containing the selector assignments entries based on
dict[str, Any]	the specified scope.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def entries(
    self, *, scope: Literal['all', 'deferred', 'set'] = 'all',
) -> dict[str, Any]:
    """Return the selector assignments.

    It returns the assignments dictionary based on the specified scope.

    Args:
        scope: The scope of the selector assignments to return. It can be
            either ``'all'`` to return all selector assignments entries,
            ``'deferred'`` to return all deferred assignments entries, or
            ``'set'`` to return only the assignments entries that have
            been explicitly set. Defaults to ``'all'``.

    Returns:
        A dictionary containing the selector assignments entries based on
        the specified scope.
    """
    if scope == 'deferred':
        return {k: v for k, v in self.items() if v is Deferred}
    elif scope == 'set':
        return {k: v for k, v in self.items() if v is not Deferred}
    else:
        return {k: v for k, v in self.items()}

resolve async

resolve(
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
) -> Sequence[_T] | _T

Resolve the selector.

It resolves the selector against the associated resource class and returns the matching resource instances. If the selector is not associated with a collection, it returns a single resource instance.

Parameters:

Name	Type	Description	Default
__session	AsyncSession | None	The async session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None
limit	int | None	The maximum number of resource instances to return. If not provided, all matching resource instances are returned. Defaults to None.	None
options	Sequence[ExecutableOption] | None	The options to apply to the query. Defaults to None.	None

Returns:

Type	Description
Sequence[_T] | _T	The resolved resource instances or single instance.

Note

This method is similar to the resource manager get_one and get_many database operations, enforcing at least one matching resource instance for the latter.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

async def resolve(
    self,
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
) -> Sequence[_T] | _T:
    """Resolve the selector.

    It resolves the selector against the associated resource class and
    returns the matching resource instances. If the selector is not
    associated with a collection, it returns a single resource instance.

    Args:
        __session: The async session to use for the operation. If not
            provided, the session in the current context is used.
            Defaults to ``None``.
        limit: The maximum number of resource instances to return. If not
            provided, all matching resource instances are returned.
            Defaults to ``None``.
        options: The options to apply to the query.
            Defaults to ``None``.

    Returns:
        The resolved resource instances or single instance.

    Note:
        This method is similar to the resource manager `get_one` and
        `get_many` database operations, enforcing at least one matching
        resource instance for the latter.
    """
    async def query(session: AsyncSession) -> Sequence[_T] | _T:
        query = self.build_query(options=options)

        # Handle collection
        if self.config.collection:
            if limit is not None:
                query = query.limit(limit)
            buffer = await session.execute(query)
            result = buffer.unique().scalars().all()
            if len(result) == 0:
                raise ValueError(
                    f"Expected at least one resource instance matching "
                    f"the provided selector. Got no instance with "
                    f"selector: {self!r}."
                )
            return result

        # Handle single instance
        buffer = await session.execute(query.limit(2))
        result = buffer.unique().scalars().all()
        if len(result) != 1:
            raise ValueError(
                f"Expected exactly one resource instance matching the "
                f"provided selector. Got {len(result)} instances with "
                f"selector: {self!r}."
            )
        return result[0]

    if __session:
        return await query(__session)
    async with async_session_manager(on_missing='raise') as session:
        return await query(session)

resolve_lenient async

resolve_lenient(
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
) -> Sequence[_T] | _T | None

Resolve the selector.

It resolves the selector in a lenient way against the associated resource class and returns the matching resource instances. If the selector is not associated with a collection, it returns a single resource instance.

Parameters:

Name	Type	Description	Default
__session	AsyncSession | None	The async session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None
limit	int | None	The maximum number of resource instances to return. If not provided, all matching resource instances are returned. Defaults to None.	None
options	Sequence[ExecutableOption] | None	The options to apply to the query. Defaults to None.	None

Returns:

Type	Description
Sequence[_T] | _T | None	The resolved resource instances or single instance.

Note

This method is similar to the resource manager get and get_many database operations.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

async def resolve_lenient(
    self,
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
) -> Sequence[_T] | _T | None:
    """Resolve the selector.

    It resolves the selector in a lenient way against the associated
    resource class and returns the matching resource instances. If the
    selector is not associated with a collection, it returns a single
    resource instance.

    Args:
        __session: The async session to use for the operation. If not
            provided, the session in the current context is used.
            Defaults to ``None``.
        limit: The maximum number of resource instances to return. If not
            provided, all matching resource instances are returned.
            Defaults to ``None``.
        options: The options to apply to the query.
            Defaults to ``None``.

    Returns:
        The resolved resource instances or single instance.

    Note:
        This method is similar to the resource manager `get` and `get_many`
        database operations.
    """
    async def query(session: AsyncSession) -> Sequence[_T] | _T | None:
        query = self.build_query(options=options)

        # Handle collection
        if self.config.collection:
            if limit is not None:
                query = query.limit(limit)
            buffer = await session.execute(query)
            return buffer.unique().scalars().all()

        # Handle single instance
        buffer = await session.execute(query.limit(1))
        return buffer.scalar()

    if __session:
        return await query(__session)
    async with async_session_manager(on_missing='raise') as session:
        return await query(session)

create classmethod

create(resource: type[_T]) -> type[Self]

Create a new selector class for the given resource.

It constructs a new selector class for the provided resource class and set the __config_resource__ attribute to the resource class.

Parameters:

Name	Type	Description	Default
resource	type[_T]	The resource class to create the selector for.	required

Returns:

Type	Description
type[Self]	The new selector class for the given resource.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def create(cls, resource: type[_T]) -> type[Self]:
    """Create a new selector class for the given resource.

    It constructs a new selector class for the provided resource class and
    set the `__config_resource__` attribute to the resource class.

    Args:
        resource: The resource class to create the selector for.

    Returns:
        The new selector class for the given resource.
    """
    name = cls.__name__
    bases = (cls,)
    namespace = {
        '__module__': __name__,
        '__config_resource__': resource,
    }

    return type(name, bases, namespace)

create_type_adapter classmethod

create_type_adapter(
    alias: str, *, resource: type[_T] | None = None
) -> TypeAdapter[Any]

Create a type adapter for the given alias.

Parameters:

Name	Type	Description	Default
alias	str	The resource alias to create the type adapter for.	required
resource	type[_T] | None	The resource associated with the selector. If not provided, the resource is retrieved from the selector class when available. Defaults to None.	None

Returns:

Type	Description
TypeAdapter[Any]	The type adapter for the given alias.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def create_type_adapter(
    cls,
    alias: str,
    *,
    resource: type[_T] | None = None,
) -> TypeAdapter[Any]:
    """Create a type adapter for the given alias.

    Args:
        alias: The resource alias to create the type adapter for.
        resource: The resource associated with the selector. If not
            provided, the resource is retrieved from the selector
            class when available. Defaults to ``None``.

    Returns:
        The type adapter for the given alias.
    """
    resource = cls.validate_resource(resource)
    annotation_type = None
    for field in resource.resource_fields.values():
        if field.alias == alias:
            if field.rel_attribute is not None:
                assert isinstance(field.target, type)
                annotation_type = Key[
                    Annotated[field.target, field.alias]  # type: ignore
                ]
            else:
                annotation_type = field.annotation
            break

    if annotation_type is None:
        raise TypeError(
            f"Invalid alias {alias!r} found when creating a type adapter "
            f"for the resource selector {cls.__qualname__!r}. The alias "
            f"does not match any of the resource field aliases."
        )

    return TypeAdapter(annotation_type)

serialize classmethod

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["json"],
    scope: Literal["all", "deferred", "set"] = "all",
) -> str

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["python"] | None = None,
    scope: Literal["all", "deferred", "set"] = "all",
) -> dict[str, Any]

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["json", "python"] | None = None,
    scope: Literal["all", "deferred", "set"] = "all",
) -> dict[str, Any] | str

Serialize the selector to a string or a dictionary.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	The selector to serialize.	required
info	SerializationInfo | None	Additional information to pass to the serializer. Defaults to None.	None
mode	Literal['json', 'python'] | None	The mode in which the serializer should run: - If mode is json, the output is a string that can be used as a resource field reference. - If mode is python, the output is a dictionary that can be used to represent the selector object. Defaults to None which resolves to python if no mode is provided in the info.	None
scope	Literal['all', 'deferred', 'set']	The scope of the serialization to perform. It can be either 'all' to process all selector assignments entries, 'deferred' to process all deferred assignments entries, or 'set' to process only the assignments entries that have been explicitly set. Defaults to 'all'.	'all'

Returns:

Type	Description
dict[str, Any] | str	The serialized selector as a string if the serialization mode is
dict[str, Any] | str	set to json, otherwise as a dictionary.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def serialize(
    cls,
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal['json', 'python'] | None = None,
    scope: Literal['all', 'deferred', 'set'] = 'all',
) -> dict[str, Any] | str:
    """Serialize the selector to a string or a dictionary.

    Args:
        obj: The selector to serialize.
        info: Additional information to pass to the serializer.
            Defaults to ``None``.
        mode: The mode in which the serializer should run:
            - If mode is ``json``, the output is a string that can be
                used as a resource field reference.
            - If mode is ``python``, the output is a dictionary that
                can be used to represent the selector object.
            Defaults to ``None`` which resolves to ``python`` if no mode
            is provided in the info.
        scope: The scope of the serialization to perform. It can be either
            ``'all'`` to process all selector assignments entries,
            ``'deferred'`` to process all deferred assignments entries, or
            ``'set'`` to process only the assignments entries that have
            been explicitly set. Defaults to ``'all'``.

    Returns:
        The serialized selector as a string if the serialization mode is
        set to ``json``, otherwise as a dictionary.
    """
    assert isinstance(obj, dict)

    mode = mode or (info and info.mode) or 'python'  # type: ignore

    # Validate scope
    if scope != 'all':
        if not isinstance(obj, cls):
            raise TypeError(
                f"Invalid selector object found when serializing the "
                f"selector {cls.__qualname__!r}. The object must be an "
                f"instance of the selector class when specifying a scope. "
                f"Got: {obj!r} with scope {scope!r}."
            )
        obj = obj.entries(scope=scope)

    # Serialize selector
    if mode == 'json':
        return ';'.join(
            f'{k}={"?" if v is Deferred else v}'
            for k, v in obj.items()
        )
    else:
        return obj.copy()

validate classmethod

validate(
    obj: Any,
    *,
    resource: type[_T] | None = None,
    update: dict[str, Any] | None = None,
    validate_assignment: bool | None = None,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Self

Validate the given object against the selector.

Parameters:

Name	Type	Description	Default
obj	Any	The object to validate.	required
resource	type[_T] | None	The resource associated with the selector. If not provided, the resource is retrieved from the selector class. Defaults to None.	None
update	dict[str, Any] | None	Values to add/modify within the selector assignments. Defaults to None.	None
validate_assignment	bool | None	Whether to validate assignments against the resource. Defaults to None.	None
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Raises:

Type	Description
ValidationError	If the object could not be validated.

Returns:

Type	Description
Self	A validated selector instance.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def validate(
    cls,
    obj: Any,
    *,
    resource: type[_T] | None = None,
    update: dict[str, Any] | None = None,
    validate_assignment: bool | None = None,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Self:
    """Validate the given object against the selector.

    Args:
        obj: The object to validate.
        resource: The resource associated with the selector. If not
            provided, the resource is retrieved from the selector class.
            Defaults to ``None``.
        update: Values to add/modify within the selector assignments.
            Defaults to ``None``.
        validate_assignment: Whether to validate assignments against the
            resource. Defaults to ``None``.
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Raises:
        ValidationError: If the object could not be validated.

    Returns:
        A validated selector instance.
    """
    resource = cls.validate_resource(resource)

    # Check validation settings
    if not validate_assignment and (strict or context):
        raise ValueError(
            f"Cannot set `strict` or `context` when assignment validation "
            f"is set to `False`. Got strict={strict} and "
            f"context={context}."
        )

    update = update or {}

    # Create a new selector instance
    if isinstance(obj, dict):
        self = cls(**{**obj, **update})
    else:
        self = cls(obj, **update)

    # Validate aliases and assignments
    setattr(self, '__config_resource__', resource)
    wildcard = self.validate_aliases(strict=strict, context=context)
    if validate_assignment:
        self.validate_assignments(
            strict=strict,
            context=context,
            exclude={wildcard} if wildcard else None,
        )

    return self

validate_resource classmethod

validate_resource(
    resource: type[_T] | None = None,
) -> type[_T]

Validate a resource against the selector.

It validates the provided resource against the selector configuration resource class. If no resource is provided, it retrieves the resource from the selector class configuration if available. Otherwise, it raises an error if no resource is found.

Parameters:

Name	Type	Description	Default
resource	type[_T] | None	The resource to validate. If not provided, the resource is retrieved from the selector class configuration if available. Defaults to None.	None

Returns:

Type	Description
type[_T]	The validated resource class.

Raises:

Type	Description
TypeError	If the resource type is invalid for the selector.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def validate_resource(cls, resource: type[_T] | None = None) -> type[_T]:
    """Validate a resource against the selector.

    It validates the provided resource against the selector configuration
    resource class. If no resource is provided, it retrieves the resource
    from the selector class configuration if available. Otherwise, it
    raises an error if no resource is found.

    Args:
        resource: The resource to validate. If not provided, the resource
            is retrieved from the selector class configuration if
            available. Defaults to ``None``.

    Returns:
        The validated resource class.

    Raises:
        TypeError: If the resource type is invalid for the selector.
    """
    # Retrieve resource from the selector configuration if not provided
    if resource is None:
        if cls.__config_resource__ is None:
            raise TypeError(
                f"No resource found for the validation of the resource "
                f"selector {cls.__qualname__!r}. Please provide a valid "
                f"resource for the validation."
            )
        resource = cls.__config_resource__

    # Validate the resource type against the selector configuration
    if cls.__config_resource__ is not None \
            and not issubclass(resource, cls.__config_resource__):
        raise TypeError(
            f"Invalid resource type for resource selector "
            f"{cls.__qualname__!r}. The provided resource type must be a "
            f"subclass of the selector configuration resource type. Got "
            f"{resource!r} and expected a subclass of "
            f"{cls.__config_resource__!r}."
        )

    return resource

validate_aliases

validate_aliases(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> str | None

Validate the selector aliases against the resource.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Returns:

Type	Description
str | None	The wildcard selector resolved alias if found.

Raises:

Type	Description
ValidationError	If any alias is invalid.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def validate_aliases(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> str | None:
    """Validate the selector aliases against the resource.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Returns:
        The wildcard selector resolved alias if found.

    Raises:
        ValidationError: If any alias is invalid.
    """
    # Return for collection selectors with no wildcard
    if self.config.collection and WILDCARD not in self:
        return None

    class SelectorMatch(TypedDict):
        aliases: set[str]
        wildcard: str | None

    # Collect all potential matching selectors
    selectors: list[SelectorMatch] = []
    selectors_check: list[SelectorMatch] = [
        SelectorMatch(aliases=aliases, wildcard=None)
        for aliases in self.resource.resource_identifiers
    ]

    # Filter out selectors that do not match the provided aliases
    selectors_check.sort(key=lambda s: len(s['aliases']), reverse=True)
    for selector in selectors_check:
        # Loop through the selector aliases and append if matched
        remaining = [*selector['aliases']]
        for alias in selector['aliases']:
            if alias in self:
                remaining.remove(alias)
            elif selector['wildcard'] is None and WILDCARD in self:
                selector['wildcard'] = alias
                remaining.remove(alias)
        # Append selector if all aliases are matched
        if len(remaining) == 0:
            selectors.append(selector)

    # Check if at least one matching selector is found
    if len(selectors) == 0:
        raise ValueError(
            f"Invalid resource selector {self!r} for resource class "
            f"{self.resource.__qualname__!r}. No valid resource selectors "
            f"matches the provided aliases."
        )

    # Skip if no wildcard selector assignment is found
    if WILDCARD not in self:
        return None
    value = self.pop(WILDCARD)

    # Validate wildcard selector assignment against resource
    for selector in selectors:
        wildcard = selector['wildcard']
        assert wildcard is not None
        validator = self.create_type_adapter(
            wildcard, resource=self.resource
        )
        try:
            self[wildcard] = validator.validate_python(
                value, strict=strict, context=context
            )
            return wildcard
        except ValidationError:
            pass

    raise ValueError(
        f"Invalid resource selector {self!r} for resource class "
        f"{self.resource.__qualname__!r}. The type of the wildcard "
        f"selector assignment value {value!r} does not match any of the "
        f"following field aliases: "
        f"{', '.join(str(i['wildcard']) for i in selectors)}."
    )

validate_assignments

validate_assignments(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
) -> None

Validate the selector assignments against the resource.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None
include	Iterable[str] | None	The aliases to include in the validation. If not provided, all aliases are included. Defaults to None.	None
exclude	Iterable[str] | None	The aliases to exclude from the validation. If not provided, no aliases are excluded. Defaults to None.	None

Raises:

Type	Description
ValidationError	If any assignment is invalid.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def validate_assignments(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
) -> None:
    """Validate the selector assignments against the resource.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.
        include: The aliases to include in the validation. If not provided,
            all aliases are included. Defaults to ``None``.
        exclude: The aliases to exclude from the validation. If not
            provided, no aliases are excluded. Defaults to ``None``.

    Raises:
        ValidationError: If any assignment is invalid.
    """
    for alias, value in self.items():
        if include is not None and alias not in include:
            continue
        if exclude is not None and alias in exclude:
            continue
        validator = self.create_type_adapter(alias, resource=self.resource)
        try:
            self[alias] = validator.validate_python(
                value, strict=strict, context=context
            )
        except ValidationError as error:
            raise ValueError(
                f"Invalid resource selector {self!r} for resource class "
                f"{self.resource.__qualname__!r}. The selector assignment "
                f"value {value!r} does not match the expected type for "
                f"the field {alias!r}."
            ) from error

Key

Key(*args: Any, **kwargs: Any)

Bases: BaseSelector[_T], Generic[_T]

A resource key to uniquely identify a resource instance.

This class is used as type annotation to define a resource key for a specific resource class. It must be used with a generic argument to specify the resource class associated with the underlying key within annotations, or it should be directly instantiated with the validate method by providing a resource class as an argument.

Note

The route auto-generation system will use this type hint to generate if needed the correct path parameter for the resource selector in the API routes.

This class if final and is not meant to be inherited. Use the base selector class instead to create custom selector classes.

Initialize a new selector.

It initializes a new resource selector with the provided assignment arguments and keyword arguments. The selector represents a dictionary of aliases and their associated values that identify resource instances.

To validate the selector against a resource class, the validate method should be called with a specific resource and collection instead of directly initializing the selector.

Parameters:

Name	Type	Description	Default
*args	Any	The provided selector assignments as a list of string arguments with the following format alias=value optionally separated by semicolons. If an argument is not of string type, it is assigned to the wildcard entry *.	()
**kwargs	Any	The provided selector assignments as a dictionary of assignment values with the aliases as keys.	{}

Note

Only one wildcard entry * is allowed within the selector assignments. The wildcard entry is used to match any field alias in a resource selector. If the wildcard entry is provided, it will be matched against the selectors of the resource when specified to determine the correct field alias.

A selector assignment value can be Deferred to represent an entry that should be inferred when initializing a resource instance. Those entries can be filtered out when collecting the selector assignment entries using the scope 'set'.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize a new selector.

    It initializes a new resource selector with the provided assignment
    arguments and keyword arguments. The selector represents a dictionary
    of aliases and their associated values that identify resource
    instances.

    To validate the selector against a resource class, the `validate`
    method should be called with a specific resource and collection instead
    of directly initializing the selector.

    Args:
        *args: The provided selector assignments as a list of string
            arguments with the following format ``alias=value`` optionally
            separated by semicolons. If an argument is not of string type,
            it is assigned to the wildcard entry ``*``.
        **kwargs: The provided selector assignments as a dictionary of
            assignment values with the aliases as keys.

    Note:
        Only one wildcard entry ``*`` is allowed within the selector
        assignments. The wildcard entry is used to match any field alias in
        a resource selector. If the wildcard entry is provided, it will
        be matched against the selectors of the resource when specified to
        determine the correct field alias.

        A selector assignment value can be `Deferred` to represent an
        entry that should be inferred when initializing a resource
        instance. Those entries can be filtered out when collecting the
        selector assignment entries using the scope ``'set'``.
    """
    # Collect all selector assignments from the given arguments
    for arg in args:
        # Parse selector
        if isinstance(arg, str):
            separator = r'[' + Symbol.STATEMENT + r']'
            selector = parse_selection(arg, separator=separator)
        else:
            selector = {WILDCARD: arg}
        # Validate assignments
        for alias, value in selector.items():
            if alias in self:
                raise KeyError(
                    f"Duplicate selector assignment aliases found for "
                    f"{alias}` from argument `{arg!r}."
                )
            self[alias] = value

    # Collect all selector assignments from the given keyword arguments
    for alias, value in kwargs.items():
        # Validate assignment alias
        alias = alias.lower()
        if not re.match(RegexPattern.ALIAS, alias):
            raise ValueError(
                f"Invalid selector assignment alias {alias!r} found in "
                f"keyword argument '{alias}={value}'. Aliases must match "
                f"a specific pattern `ALIAS` defined in the framework's "
                f"regular expressions repository."
            )
        # Validate assignment value
        if alias in self:
            raise KeyError(
                f"Duplicate selector assignment aliases found for "
                f"{alias!r} from keyword argument '{alias}={value}'."
            )
        self[alias] = value

resource property

resource: type[_T]

The resource associated with the selector.

It is used to validate the selection assignments against this resource class and to generate the correct path parameter for the resource selector in the API routes.

config

config() -> SelectorConfig

The selector configuration.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classproperty
def config(cls) -> SelectorConfig:
    """The selector configuration."""
    return cls.__config__

build_query

build_query(
    *,
    filter_criteria: Filter | None = None,
    sort_criteria: Sort | None = None,
    options: Sequence[ExecutableOption] | None = None,
    raise_errors: bool = True,
) -> Select[tuple[_T]]

Build a query for the selector.

Recursively builds a query for the selector by performing a join operation on the resource relationships found in the selector and filters the query based on the selector assignments. Optionally, the function can apply additional filter and sort criteria to the query.

Parameters:

Name	Type	Description	Default
filter_criteria	Filter | None	The filter criteria to apply to the query. Defaults to None.	None
sort_criteria	Sort | None	The sort criteria to apply to the query. Defaults to None.	None
options	Sequence[ExecutableOption] | None	The options to apply to the query. Defaults to None.	None
raise_errors	bool	Whether to raise errors when invalid attributes or values are found in the selector assignments. Defaults to True.	True

Returns:

Type	Description
Select[tuple[_T]]	The built selector query with the applied filter and sort criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def build_query(
    self,
    *,
    filter_criteria: Filter | None = None,
    sort_criteria: Sort | None = None,
    options: Sequence[ExecutableOption] | None = None,
    raise_errors: bool = True,
) -> Select[tuple[_T]]:
    """Build a query for the selector.

    Recursively builds a query for the selector by performing a join
    operation on the resource relationships found in the selector and
    filters the query based on the selector assignments. Optionally, the
    function can apply additional filter and sort criteria to the query.

    Args:
        filter_criteria: The filter criteria to apply to the query.
            Defaults to ``None``.
        sort_criteria: The sort criteria to apply to the query.
            Defaults to ``None``.
        options: The options to apply to the query.
            Defaults to ``None``.
        raise_errors: Whether to raise errors when invalid attributes or
            values are found in the selector assignments.
            Defaults to ``True``.

    Returns:
        The built selector query with the applied filter and sort criteria.
    """
    return build_query(
        self.resource,
        selection=self,
        filter_criteria=filter_criteria,
        sort_criteria=sort_criteria,
        raise_errors=raise_errors,
        options=options,
    )

copy

copy() -> Self

Return a shallow copy of the selector.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def copy(self) -> Self:
    """Return a shallow copy of the selector."""
    return copy(self)

entries

entries(
    *, scope: Literal["all", "deferred", "set"] = "all"
) -> dict[str, Any]

Return the selector assignments.

It returns the assignments dictionary based on the specified scope.

Parameters:

Name	Type	Description	Default
scope	Literal['all', 'deferred', 'set']	The scope of the selector assignments to return. It can be either 'all' to return all selector assignments entries, 'deferred' to return all deferred assignments entries, or 'set' to return only the assignments entries that have been explicitly set. Defaults to 'all'.	'all'

Returns:

Type	Description
dict[str, Any]	A dictionary containing the selector assignments entries based on
dict[str, Any]	the specified scope.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def entries(
    self, *, scope: Literal['all', 'deferred', 'set'] = 'all',
) -> dict[str, Any]:
    """Return the selector assignments.

    It returns the assignments dictionary based on the specified scope.

    Args:
        scope: The scope of the selector assignments to return. It can be
            either ``'all'`` to return all selector assignments entries,
            ``'deferred'`` to return all deferred assignments entries, or
            ``'set'`` to return only the assignments entries that have
            been explicitly set. Defaults to ``'all'``.

    Returns:
        A dictionary containing the selector assignments entries based on
        the specified scope.
    """
    if scope == 'deferred':
        return {k: v for k, v in self.items() if v is Deferred}
    elif scope == 'set':
        return {k: v for k, v in self.items() if v is not Deferred}
    else:
        return {k: v for k, v in self.items()}

create classmethod

create(resource: type[_T]) -> type[Self]

Create a new selector class for the given resource.

It constructs a new selector class for the provided resource class and set the __config_resource__ attribute to the resource class.

Parameters:

Name	Type	Description	Default
resource	type[_T]	The resource class to create the selector for.	required

Returns:

Type	Description
type[Self]	The new selector class for the given resource.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def create(cls, resource: type[_T]) -> type[Self]:
    """Create a new selector class for the given resource.

    It constructs a new selector class for the provided resource class and
    set the `__config_resource__` attribute to the resource class.

    Args:
        resource: The resource class to create the selector for.

    Returns:
        The new selector class for the given resource.
    """
    name = cls.__name__
    bases = (cls,)
    namespace = {
        '__module__': __name__,
        '__config_resource__': resource,
    }

    return type(name, bases, namespace)

create_type_adapter classmethod

create_type_adapter(
    alias: str, *, resource: type[_T] | None = None
) -> TypeAdapter[Any]

Create a type adapter for the given alias.

Parameters:

Name	Type	Description	Default
alias	str	The resource alias to create the type adapter for.	required
resource	type[_T] | None	The resource associated with the selector. If not provided, the resource is retrieved from the selector class when available. Defaults to None.	None

Returns:

Type	Description
TypeAdapter[Any]	The type adapter for the given alias.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def create_type_adapter(
    cls,
    alias: str,
    *,
    resource: type[_T] | None = None,
) -> TypeAdapter[Any]:
    """Create a type adapter for the given alias.

    Args:
        alias: The resource alias to create the type adapter for.
        resource: The resource associated with the selector. If not
            provided, the resource is retrieved from the selector
            class when available. Defaults to ``None``.

    Returns:
        The type adapter for the given alias.
    """
    resource = cls.validate_resource(resource)
    annotation_type = None
    for field in resource.resource_fields.values():
        if field.alias == alias:
            if field.rel_attribute is not None:
                assert isinstance(field.target, type)
                annotation_type = Key[
                    Annotated[field.target, field.alias]  # type: ignore
                ]
            else:
                annotation_type = field.annotation
            break

    if annotation_type is None:
        raise TypeError(
            f"Invalid alias {alias!r} found when creating a type adapter "
            f"for the resource selector {cls.__qualname__!r}. The alias "
            f"does not match any of the resource field aliases."
        )

    return TypeAdapter(annotation_type)

serialize classmethod

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["json"],
    scope: Literal["all", "deferred", "set"] = "all",
) -> str

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["python"] | None = None,
    scope: Literal["all", "deferred", "set"] = "all",
) -> dict[str, Any]

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["json", "python"] | None = None,
    scope: Literal["all", "deferred", "set"] = "all",
) -> dict[str, Any] | str

Serialize the selector to a string or a dictionary.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	The selector to serialize.	required
info	SerializationInfo | None	Additional information to pass to the serializer. Defaults to None.	None
mode	Literal['json', 'python'] | None	The mode in which the serializer should run: - If mode is json, the output is a string that can be used as a resource field reference. - If mode is python, the output is a dictionary that can be used to represent the selector object. Defaults to None which resolves to python if no mode is provided in the info.	None
scope	Literal['all', 'deferred', 'set']	The scope of the serialization to perform. It can be either 'all' to process all selector assignments entries, 'deferred' to process all deferred assignments entries, or 'set' to process only the assignments entries that have been explicitly set. Defaults to 'all'.	'all'

Returns:

Type	Description
dict[str, Any] | str	The serialized selector as a string if the serialization mode is
dict[str, Any] | str	set to json, otherwise as a dictionary.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def serialize(
    cls,
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal['json', 'python'] | None = None,
    scope: Literal['all', 'deferred', 'set'] = 'all',
) -> dict[str, Any] | str:
    """Serialize the selector to a string or a dictionary.

    Args:
        obj: The selector to serialize.
        info: Additional information to pass to the serializer.
            Defaults to ``None``.
        mode: The mode in which the serializer should run:
            - If mode is ``json``, the output is a string that can be
                used as a resource field reference.
            - If mode is ``python``, the output is a dictionary that
                can be used to represent the selector object.
            Defaults to ``None`` which resolves to ``python`` if no mode
            is provided in the info.
        scope: The scope of the serialization to perform. It can be either
            ``'all'`` to process all selector assignments entries,
            ``'deferred'`` to process all deferred assignments entries, or
            ``'set'`` to process only the assignments entries that have
            been explicitly set. Defaults to ``'all'``.

    Returns:
        The serialized selector as a string if the serialization mode is
        set to ``json``, otherwise as a dictionary.
    """
    assert isinstance(obj, dict)

    mode = mode or (info and info.mode) or 'python'  # type: ignore

    # Validate scope
    if scope != 'all':
        if not isinstance(obj, cls):
            raise TypeError(
                f"Invalid selector object found when serializing the "
                f"selector {cls.__qualname__!r}. The object must be an "
                f"instance of the selector class when specifying a scope. "
                f"Got: {obj!r} with scope {scope!r}."
            )
        obj = obj.entries(scope=scope)

    # Serialize selector
    if mode == 'json':
        return ';'.join(
            f'{k}={"?" if v is Deferred else v}'
            for k, v in obj.items()
        )
    else:
        return obj.copy()

validate classmethod

validate(
    obj: Any,
    *,
    resource: type[_T] | None = None,
    update: dict[str, Any] | None = None,
    validate_assignment: bool | None = None,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Self

Validate the given object against the selector.

Parameters:

Name	Type	Description	Default
obj	Any	The object to validate.	required
resource	type[_T] | None	The resource associated with the selector. If not provided, the resource is retrieved from the selector class. Defaults to None.	None
update	dict[str, Any] | None	Values to add/modify within the selector assignments. Defaults to None.	None
validate_assignment	bool | None	Whether to validate assignments against the resource. Defaults to None.	None
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Raises:

Type	Description
ValidationError	If the object could not be validated.

Returns:

Type	Description
Self	A validated selector instance.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def validate(
    cls,
    obj: Any,
    *,
    resource: type[_T] | None = None,
    update: dict[str, Any] | None = None,
    validate_assignment: bool | None = None,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Self:
    """Validate the given object against the selector.

    Args:
        obj: The object to validate.
        resource: The resource associated with the selector. If not
            provided, the resource is retrieved from the selector class.
            Defaults to ``None``.
        update: Values to add/modify within the selector assignments.
            Defaults to ``None``.
        validate_assignment: Whether to validate assignments against the
            resource. Defaults to ``None``.
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Raises:
        ValidationError: If the object could not be validated.

    Returns:
        A validated selector instance.
    """
    resource = cls.validate_resource(resource)

    # Check validation settings
    if not validate_assignment and (strict or context):
        raise ValueError(
            f"Cannot set `strict` or `context` when assignment validation "
            f"is set to `False`. Got strict={strict} and "
            f"context={context}."
        )

    update = update or {}

    # Create a new selector instance
    if isinstance(obj, dict):
        self = cls(**{**obj, **update})
    else:
        self = cls(obj, **update)

    # Validate aliases and assignments
    setattr(self, '__config_resource__', resource)
    wildcard = self.validate_aliases(strict=strict, context=context)
    if validate_assignment:
        self.validate_assignments(
            strict=strict,
            context=context,
            exclude={wildcard} if wildcard else None,
        )

    return self

validate_resource classmethod

validate_resource(
    resource: type[_T] | None = None,
) -> type[_T]

Validate a resource against the selector.

It validates the provided resource against the selector configuration resource class. If no resource is provided, it retrieves the resource from the selector class configuration if available. Otherwise, it raises an error if no resource is found.

Parameters:

Name	Type	Description	Default
resource	type[_T] | None	The resource to validate. If not provided, the resource is retrieved from the selector class configuration if available. Defaults to None.	None

Returns:

Type	Description
type[_T]	The validated resource class.

Raises:

Type	Description
TypeError	If the resource type is invalid for the selector.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def validate_resource(cls, resource: type[_T] | None = None) -> type[_T]:
    """Validate a resource against the selector.

    It validates the provided resource against the selector configuration
    resource class. If no resource is provided, it retrieves the resource
    from the selector class configuration if available. Otherwise, it
    raises an error if no resource is found.

    Args:
        resource: The resource to validate. If not provided, the resource
            is retrieved from the selector class configuration if
            available. Defaults to ``None``.

    Returns:
        The validated resource class.

    Raises:
        TypeError: If the resource type is invalid for the selector.
    """
    # Retrieve resource from the selector configuration if not provided
    if resource is None:
        if cls.__config_resource__ is None:
            raise TypeError(
                f"No resource found for the validation of the resource "
                f"selector {cls.__qualname__!r}. Please provide a valid "
                f"resource for the validation."
            )
        resource = cls.__config_resource__

    # Validate the resource type against the selector configuration
    if cls.__config_resource__ is not None \
            and not issubclass(resource, cls.__config_resource__):
        raise TypeError(
            f"Invalid resource type for resource selector "
            f"{cls.__qualname__!r}. The provided resource type must be a "
            f"subclass of the selector configuration resource type. Got "
            f"{resource!r} and expected a subclass of "
            f"{cls.__config_resource__!r}."
        )

    return resource

validate_aliases

validate_aliases(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> str | None

Validate the selector aliases against the resource.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Returns:

Type	Description
str | None	The wildcard selector resolved alias if found.

Raises:

Type	Description
ValidationError	If any alias is invalid.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def validate_aliases(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> str | None:
    """Validate the selector aliases against the resource.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Returns:
        The wildcard selector resolved alias if found.

    Raises:
        ValidationError: If any alias is invalid.
    """
    # Return for collection selectors with no wildcard
    if self.config.collection and WILDCARD not in self:
        return None

    class SelectorMatch(TypedDict):
        aliases: set[str]
        wildcard: str | None

    # Collect all potential matching selectors
    selectors: list[SelectorMatch] = []
    selectors_check: list[SelectorMatch] = [
        SelectorMatch(aliases=aliases, wildcard=None)
        for aliases in self.resource.resource_identifiers
    ]

    # Filter out selectors that do not match the provided aliases
    selectors_check.sort(key=lambda s: len(s['aliases']), reverse=True)
    for selector in selectors_check:
        # Loop through the selector aliases and append if matched
        remaining = [*selector['aliases']]
        for alias in selector['aliases']:
            if alias in self:
                remaining.remove(alias)
            elif selector['wildcard'] is None and WILDCARD in self:
                selector['wildcard'] = alias
                remaining.remove(alias)
        # Append selector if all aliases are matched
        if len(remaining) == 0:
            selectors.append(selector)

    # Check if at least one matching selector is found
    if len(selectors) == 0:
        raise ValueError(
            f"Invalid resource selector {self!r} for resource class "
            f"{self.resource.__qualname__!r}. No valid resource selectors "
            f"matches the provided aliases."
        )

    # Skip if no wildcard selector assignment is found
    if WILDCARD not in self:
        return None
    value = self.pop(WILDCARD)

    # Validate wildcard selector assignment against resource
    for selector in selectors:
        wildcard = selector['wildcard']
        assert wildcard is not None
        validator = self.create_type_adapter(
            wildcard, resource=self.resource
        )
        try:
            self[wildcard] = validator.validate_python(
                value, strict=strict, context=context
            )
            return wildcard
        except ValidationError:
            pass

    raise ValueError(
        f"Invalid resource selector {self!r} for resource class "
        f"{self.resource.__qualname__!r}. The type of the wildcard "
        f"selector assignment value {value!r} does not match any of the "
        f"following field aliases: "
        f"{', '.join(str(i['wildcard']) for i in selectors)}."
    )

validate_assignments

validate_assignments(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
) -> None

Validate the selector assignments against the resource.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None
include	Iterable[str] | None	The aliases to include in the validation. If not provided, all aliases are included. Defaults to None.	None
exclude	Iterable[str] | None	The aliases to exclude from the validation. If not provided, no aliases are excluded. Defaults to None.	None

Raises:

Type	Description
ValidationError	If any assignment is invalid.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def validate_assignments(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
) -> None:
    """Validate the selector assignments against the resource.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.
        include: The aliases to include in the validation. If not provided,
            all aliases are included. Defaults to ``None``.
        exclude: The aliases to exclude from the validation. If not
            provided, no aliases are excluded. Defaults to ``None``.

    Raises:
        ValidationError: If any assignment is invalid.
    """
    for alias, value in self.items():
        if include is not None and alias not in include:
            continue
        if exclude is not None and alias in exclude:
            continue
        validator = self.create_type_adapter(alias, resource=self.resource)
        try:
            self[alias] = validator.validate_python(
                value, strict=strict, context=context
            )
        except ValidationError as error:
            raise ValueError(
                f"Invalid resource selector {self!r} for resource class "
                f"{self.resource.__qualname__!r}. The selector assignment "
                f"value {value!r} does not match the expected type for "
                f"the field {alias!r}."
            ) from error

resolve async

resolve(
    __session: AsyncSession | None = None,
    *,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> _T

Resolve the key.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

async def resolve(
    self,
    __session: AsyncSession | None = None,
    *,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> _T:
    """Resolve the key."""
    result = await super().resolve(__session, options=options)
    assert not isinstance(result, Sequence)
    return result

resolve_lenient async

resolve_lenient(
    __session: AsyncSession | None = None,
    *,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> _T | None

Resolve the key in a lenient way.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

async def resolve_lenient(
    self,
    __session: AsyncSession | None = None,
    *,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> _T | None:
    """Resolve the key in a lenient way."""
    result = await super().resolve_lenient(__session, options=options)
    assert result is None or not isinstance(result, Sequence)
    return result

KeyList

KeyList(*args: Any, **kwargs: Any)

Bases: BaseSelector[_T], Generic[_T]

A resource key list to identify a list of resource instances.

This class is used as type annotation to define a resource key list for a specific resource class. It must be used with a generic argument to specify the resource class associated with the underlying key within annotations, or it should be directly instantiated with the validate method by providing a resource class as an argument.

Note

The route auto-generation system will use this type hint to generate if needed the correct path parameter for the resource selector in the API routes.

This class if final and is not meant to be inherited. Use the base selector class instead to create custom selector classes.

Initialize a new selector.

It initializes a new resource selector with the provided assignment arguments and keyword arguments. The selector represents a dictionary of aliases and their associated values that identify resource instances.

To validate the selector against a resource class, the validate method should be called with a specific resource and collection instead of directly initializing the selector.

Parameters:

Name	Type	Description	Default
*args	Any	The provided selector assignments as a list of string arguments with the following format alias=value optionally separated by semicolons. If an argument is not of string type, it is assigned to the wildcard entry *.	()
**kwargs	Any	The provided selector assignments as a dictionary of assignment values with the aliases as keys.	{}

Note

Only one wildcard entry * is allowed within the selector assignments. The wildcard entry is used to match any field alias in a resource selector. If the wildcard entry is provided, it will be matched against the selectors of the resource when specified to determine the correct field alias.

A selector assignment value can be Deferred to represent an entry that should be inferred when initializing a resource instance. Those entries can be filtered out when collecting the selector assignment entries using the scope 'set'.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize a new selector.

    It initializes a new resource selector with the provided assignment
    arguments and keyword arguments. The selector represents a dictionary
    of aliases and their associated values that identify resource
    instances.

    To validate the selector against a resource class, the `validate`
    method should be called with a specific resource and collection instead
    of directly initializing the selector.

    Args:
        *args: The provided selector assignments as a list of string
            arguments with the following format ``alias=value`` optionally
            separated by semicolons. If an argument is not of string type,
            it is assigned to the wildcard entry ``*``.
        **kwargs: The provided selector assignments as a dictionary of
            assignment values with the aliases as keys.

    Note:
        Only one wildcard entry ``*`` is allowed within the selector
        assignments. The wildcard entry is used to match any field alias in
        a resource selector. If the wildcard entry is provided, it will
        be matched against the selectors of the resource when specified to
        determine the correct field alias.

        A selector assignment value can be `Deferred` to represent an
        entry that should be inferred when initializing a resource
        instance. Those entries can be filtered out when collecting the
        selector assignment entries using the scope ``'set'``.
    """
    # Collect all selector assignments from the given arguments
    for arg in args:
        # Parse selector
        if isinstance(arg, str):
            separator = r'[' + Symbol.STATEMENT + r']'
            selector = parse_selection(arg, separator=separator)
        else:
            selector = {WILDCARD: arg}
        # Validate assignments
        for alias, value in selector.items():
            if alias in self:
                raise KeyError(
                    f"Duplicate selector assignment aliases found for "
                    f"{alias}` from argument `{arg!r}."
                )
            self[alias] = value

    # Collect all selector assignments from the given keyword arguments
    for alias, value in kwargs.items():
        # Validate assignment alias
        alias = alias.lower()
        if not re.match(RegexPattern.ALIAS, alias):
            raise ValueError(
                f"Invalid selector assignment alias {alias!r} found in "
                f"keyword argument '{alias}={value}'. Aliases must match "
                f"a specific pattern `ALIAS` defined in the framework's "
                f"regular expressions repository."
            )
        # Validate assignment value
        if alias in self:
            raise KeyError(
                f"Duplicate selector assignment aliases found for "
                f"{alias!r} from keyword argument '{alias}={value}'."
            )
        self[alias] = value

resource property

resource: type[_T]

The resource associated with the selector.

It is used to validate the selection assignments against this resource class and to generate the correct path parameter for the resource selector in the API routes.

config

config() -> SelectorConfig

The selector configuration.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classproperty
def config(cls) -> SelectorConfig:
    """The selector configuration."""
    return cls.__config__

build_query

build_query(
    *,
    filter_criteria: Filter | None = None,
    sort_criteria: Sort | None = None,
    options: Sequence[ExecutableOption] | None = None,
    raise_errors: bool = True,
) -> Select[tuple[_T]]

Build a query for the selector.

Recursively builds a query for the selector by performing a join operation on the resource relationships found in the selector and filters the query based on the selector assignments. Optionally, the function can apply additional filter and sort criteria to the query.

Parameters:

Name	Type	Description	Default
filter_criteria	Filter | None	The filter criteria to apply to the query. Defaults to None.	None
sort_criteria	Sort | None	The sort criteria to apply to the query. Defaults to None.	None
options	Sequence[ExecutableOption] | None	The options to apply to the query. Defaults to None.	None
raise_errors	bool	Whether to raise errors when invalid attributes or values are found in the selector assignments. Defaults to True.	True

Returns:

Type	Description
Select[tuple[_T]]	The built selector query with the applied filter and sort criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def build_query(
    self,
    *,
    filter_criteria: Filter | None = None,
    sort_criteria: Sort | None = None,
    options: Sequence[ExecutableOption] | None = None,
    raise_errors: bool = True,
) -> Select[tuple[_T]]:
    """Build a query for the selector.

    Recursively builds a query for the selector by performing a join
    operation on the resource relationships found in the selector and
    filters the query based on the selector assignments. Optionally, the
    function can apply additional filter and sort criteria to the query.

    Args:
        filter_criteria: The filter criteria to apply to the query.
            Defaults to ``None``.
        sort_criteria: The sort criteria to apply to the query.
            Defaults to ``None``.
        options: The options to apply to the query.
            Defaults to ``None``.
        raise_errors: Whether to raise errors when invalid attributes or
            values are found in the selector assignments.
            Defaults to ``True``.

    Returns:
        The built selector query with the applied filter and sort criteria.
    """
    return build_query(
        self.resource,
        selection=self,
        filter_criteria=filter_criteria,
        sort_criteria=sort_criteria,
        raise_errors=raise_errors,
        options=options,
    )

copy

copy() -> Self

Return a shallow copy of the selector.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def copy(self) -> Self:
    """Return a shallow copy of the selector."""
    return copy(self)

entries

entries(
    *, scope: Literal["all", "deferred", "set"] = "all"
) -> dict[str, Any]

Return the selector assignments.

It returns the assignments dictionary based on the specified scope.

Parameters:

Name	Type	Description	Default
scope	Literal['all', 'deferred', 'set']	The scope of the selector assignments to return. It can be either 'all' to return all selector assignments entries, 'deferred' to return all deferred assignments entries, or 'set' to return only the assignments entries that have been explicitly set. Defaults to 'all'.	'all'

Returns:

Type	Description
dict[str, Any]	A dictionary containing the selector assignments entries based on
dict[str, Any]	the specified scope.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def entries(
    self, *, scope: Literal['all', 'deferred', 'set'] = 'all',
) -> dict[str, Any]:
    """Return the selector assignments.

    It returns the assignments dictionary based on the specified scope.

    Args:
        scope: The scope of the selector assignments to return. It can be
            either ``'all'`` to return all selector assignments entries,
            ``'deferred'`` to return all deferred assignments entries, or
            ``'set'`` to return only the assignments entries that have
            been explicitly set. Defaults to ``'all'``.

    Returns:
        A dictionary containing the selector assignments entries based on
        the specified scope.
    """
    if scope == 'deferred':
        return {k: v for k, v in self.items() if v is Deferred}
    elif scope == 'set':
        return {k: v for k, v in self.items() if v is not Deferred}
    else:
        return {k: v for k, v in self.items()}

create classmethod

create(resource: type[_T]) -> type[Self]

Create a new selector class for the given resource.

It constructs a new selector class for the provided resource class and set the __config_resource__ attribute to the resource class.

Parameters:

Name	Type	Description	Default
resource	type[_T]	The resource class to create the selector for.	required

Returns:

Type	Description
type[Self]	The new selector class for the given resource.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def create(cls, resource: type[_T]) -> type[Self]:
    """Create a new selector class for the given resource.

    It constructs a new selector class for the provided resource class and
    set the `__config_resource__` attribute to the resource class.

    Args:
        resource: The resource class to create the selector for.

    Returns:
        The new selector class for the given resource.
    """
    name = cls.__name__
    bases = (cls,)
    namespace = {
        '__module__': __name__,
        '__config_resource__': resource,
    }

    return type(name, bases, namespace)

create_type_adapter classmethod

create_type_adapter(
    alias: str, *, resource: type[_T] | None = None
) -> TypeAdapter[Any]

Create a type adapter for the given alias.

Parameters:

Name	Type	Description	Default
alias	str	The resource alias to create the type adapter for.	required
resource	type[_T] | None	The resource associated with the selector. If not provided, the resource is retrieved from the selector class when available. Defaults to None.	None

Returns:

Type	Description
TypeAdapter[Any]	The type adapter for the given alias.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def create_type_adapter(
    cls,
    alias: str,
    *,
    resource: type[_T] | None = None,
) -> TypeAdapter[Any]:
    """Create a type adapter for the given alias.

    Args:
        alias: The resource alias to create the type adapter for.
        resource: The resource associated with the selector. If not
            provided, the resource is retrieved from the selector
            class when available. Defaults to ``None``.

    Returns:
        The type adapter for the given alias.
    """
    resource = cls.validate_resource(resource)
    annotation_type = None
    for field in resource.resource_fields.values():
        if field.alias == alias:
            if field.rel_attribute is not None:
                assert isinstance(field.target, type)
                annotation_type = Key[
                    Annotated[field.target, field.alias]  # type: ignore
                ]
            else:
                annotation_type = field.annotation
            break

    if annotation_type is None:
        raise TypeError(
            f"Invalid alias {alias!r} found when creating a type adapter "
            f"for the resource selector {cls.__qualname__!r}. The alias "
            f"does not match any of the resource field aliases."
        )

    return TypeAdapter(annotation_type)

serialize classmethod

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["json"],
    scope: Literal["all", "deferred", "set"] = "all",
) -> str

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["python"] | None = None,
    scope: Literal["all", "deferred", "set"] = "all",
) -> dict[str, Any]

serialize(
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal["json", "python"] | None = None,
    scope: Literal["all", "deferred", "set"] = "all",
) -> dict[str, Any] | str

Serialize the selector to a string or a dictionary.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	The selector to serialize.	required
info	SerializationInfo | None	Additional information to pass to the serializer. Defaults to None.	None
mode	Literal['json', 'python'] | None	The mode in which the serializer should run: - If mode is json, the output is a string that can be used as a resource field reference. - If mode is python, the output is a dictionary that can be used to represent the selector object. Defaults to None which resolves to python if no mode is provided in the info.	None
scope	Literal['all', 'deferred', 'set']	The scope of the serialization to perform. It can be either 'all' to process all selector assignments entries, 'deferred' to process all deferred assignments entries, or 'set' to process only the assignments entries that have been explicitly set. Defaults to 'all'.	'all'

Returns:

Type	Description
dict[str, Any] | str	The serialized selector as a string if the serialization mode is
dict[str, Any] | str	set to json, otherwise as a dictionary.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def serialize(
    cls,
    obj: dict[str, Any],
    info: SerializationInfo | None = None,
    *,
    mode: Literal['json', 'python'] | None = None,
    scope: Literal['all', 'deferred', 'set'] = 'all',
) -> dict[str, Any] | str:
    """Serialize the selector to a string or a dictionary.

    Args:
        obj: The selector to serialize.
        info: Additional information to pass to the serializer.
            Defaults to ``None``.
        mode: The mode in which the serializer should run:
            - If mode is ``json``, the output is a string that can be
                used as a resource field reference.
            - If mode is ``python``, the output is a dictionary that
                can be used to represent the selector object.
            Defaults to ``None`` which resolves to ``python`` if no mode
            is provided in the info.
        scope: The scope of the serialization to perform. It can be either
            ``'all'`` to process all selector assignments entries,
            ``'deferred'`` to process all deferred assignments entries, or
            ``'set'`` to process only the assignments entries that have
            been explicitly set. Defaults to ``'all'``.

    Returns:
        The serialized selector as a string if the serialization mode is
        set to ``json``, otherwise as a dictionary.
    """
    assert isinstance(obj, dict)

    mode = mode or (info and info.mode) or 'python'  # type: ignore

    # Validate scope
    if scope != 'all':
        if not isinstance(obj, cls):
            raise TypeError(
                f"Invalid selector object found when serializing the "
                f"selector {cls.__qualname__!r}. The object must be an "
                f"instance of the selector class when specifying a scope. "
                f"Got: {obj!r} with scope {scope!r}."
            )
        obj = obj.entries(scope=scope)

    # Serialize selector
    if mode == 'json':
        return ';'.join(
            f'{k}={"?" if v is Deferred else v}'
            for k, v in obj.items()
        )
    else:
        return obj.copy()

validate classmethod

validate(
    obj: Any,
    *,
    resource: type[_T] | None = None,
    update: dict[str, Any] | None = None,
    validate_assignment: bool | None = None,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Self

Validate the given object against the selector.

Parameters:

Name	Type	Description	Default
obj	Any	The object to validate.	required
resource	type[_T] | None	The resource associated with the selector. If not provided, the resource is retrieved from the selector class. Defaults to None.	None
update	dict[str, Any] | None	Values to add/modify within the selector assignments. Defaults to None.	None
validate_assignment	bool | None	Whether to validate assignments against the resource. Defaults to None.	None
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Raises:

Type	Description
ValidationError	If the object could not be validated.

Returns:

Type	Description
Self	A validated selector instance.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def validate(
    cls,
    obj: Any,
    *,
    resource: type[_T] | None = None,
    update: dict[str, Any] | None = None,
    validate_assignment: bool | None = None,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Self:
    """Validate the given object against the selector.

    Args:
        obj: The object to validate.
        resource: The resource associated with the selector. If not
            provided, the resource is retrieved from the selector class.
            Defaults to ``None``.
        update: Values to add/modify within the selector assignments.
            Defaults to ``None``.
        validate_assignment: Whether to validate assignments against the
            resource. Defaults to ``None``.
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Raises:
        ValidationError: If the object could not be validated.

    Returns:
        A validated selector instance.
    """
    resource = cls.validate_resource(resource)

    # Check validation settings
    if not validate_assignment and (strict or context):
        raise ValueError(
            f"Cannot set `strict` or `context` when assignment validation "
            f"is set to `False`. Got strict={strict} and "
            f"context={context}."
        )

    update = update or {}

    # Create a new selector instance
    if isinstance(obj, dict):
        self = cls(**{**obj, **update})
    else:
        self = cls(obj, **update)

    # Validate aliases and assignments
    setattr(self, '__config_resource__', resource)
    wildcard = self.validate_aliases(strict=strict, context=context)
    if validate_assignment:
        self.validate_assignments(
            strict=strict,
            context=context,
            exclude={wildcard} if wildcard else None,
        )

    return self

validate_resource classmethod

validate_resource(
    resource: type[_T] | None = None,
) -> type[_T]

Validate a resource against the selector.

It validates the provided resource against the selector configuration resource class. If no resource is provided, it retrieves the resource from the selector class configuration if available. Otherwise, it raises an error if no resource is found.

Parameters:

Name	Type	Description	Default
resource	type[_T] | None	The resource to validate. If not provided, the resource is retrieved from the selector class configuration if available. Defaults to None.	None

Returns:

Type	Description
type[_T]	The validated resource class.

Raises:

Type	Description
TypeError	If the resource type is invalid for the selector.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

@classmethod
def validate_resource(cls, resource: type[_T] | None = None) -> type[_T]:
    """Validate a resource against the selector.

    It validates the provided resource against the selector configuration
    resource class. If no resource is provided, it retrieves the resource
    from the selector class configuration if available. Otherwise, it
    raises an error if no resource is found.

    Args:
        resource: The resource to validate. If not provided, the resource
            is retrieved from the selector class configuration if
            available. Defaults to ``None``.

    Returns:
        The validated resource class.

    Raises:
        TypeError: If the resource type is invalid for the selector.
    """
    # Retrieve resource from the selector configuration if not provided
    if resource is None:
        if cls.__config_resource__ is None:
            raise TypeError(
                f"No resource found for the validation of the resource "
                f"selector {cls.__qualname__!r}. Please provide a valid "
                f"resource for the validation."
            )
        resource = cls.__config_resource__

    # Validate the resource type against the selector configuration
    if cls.__config_resource__ is not None \
            and not issubclass(resource, cls.__config_resource__):
        raise TypeError(
            f"Invalid resource type for resource selector "
            f"{cls.__qualname__!r}. The provided resource type must be a "
            f"subclass of the selector configuration resource type. Got "
            f"{resource!r} and expected a subclass of "
            f"{cls.__config_resource__!r}."
        )

    return resource

validate_aliases

validate_aliases(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> str | None

Validate the selector aliases against the resource.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None

Returns:

Type	Description
str | None	The wildcard selector resolved alias if found.

Raises:

Type	Description
ValidationError	If any alias is invalid.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def validate_aliases(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> str | None:
    """Validate the selector aliases against the resource.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.

    Returns:
        The wildcard selector resolved alias if found.

    Raises:
        ValidationError: If any alias is invalid.
    """
    # Return for collection selectors with no wildcard
    if self.config.collection and WILDCARD not in self:
        return None

    class SelectorMatch(TypedDict):
        aliases: set[str]
        wildcard: str | None

    # Collect all potential matching selectors
    selectors: list[SelectorMatch] = []
    selectors_check: list[SelectorMatch] = [
        SelectorMatch(aliases=aliases, wildcard=None)
        for aliases in self.resource.resource_identifiers
    ]

    # Filter out selectors that do not match the provided aliases
    selectors_check.sort(key=lambda s: len(s['aliases']), reverse=True)
    for selector in selectors_check:
        # Loop through the selector aliases and append if matched
        remaining = [*selector['aliases']]
        for alias in selector['aliases']:
            if alias in self:
                remaining.remove(alias)
            elif selector['wildcard'] is None and WILDCARD in self:
                selector['wildcard'] = alias
                remaining.remove(alias)
        # Append selector if all aliases are matched
        if len(remaining) == 0:
            selectors.append(selector)

    # Check if at least one matching selector is found
    if len(selectors) == 0:
        raise ValueError(
            f"Invalid resource selector {self!r} for resource class "
            f"{self.resource.__qualname__!r}. No valid resource selectors "
            f"matches the provided aliases."
        )

    # Skip if no wildcard selector assignment is found
    if WILDCARD not in self:
        return None
    value = self.pop(WILDCARD)

    # Validate wildcard selector assignment against resource
    for selector in selectors:
        wildcard = selector['wildcard']
        assert wildcard is not None
        validator = self.create_type_adapter(
            wildcard, resource=self.resource
        )
        try:
            self[wildcard] = validator.validate_python(
                value, strict=strict, context=context
            )
            return wildcard
        except ValidationError:
            pass

    raise ValueError(
        f"Invalid resource selector {self!r} for resource class "
        f"{self.resource.__qualname__!r}. The type of the wildcard "
        f"selector assignment value {value!r} does not match any of the "
        f"following field aliases: "
        f"{', '.join(str(i['wildcard']) for i in selectors)}."
    )

validate_assignments

validate_assignments(
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
) -> None

Validate the selector assignments against the resource.

Parameters:

Name	Type	Description	Default
strict	bool | None	Whether to enforce strict validation. Defaults to None.	None
context	dict[str, Any] | None	The context to use for validation. Defaults to None.	None
include	Iterable[str] | None	The aliases to include in the validation. If not provided, all aliases are included. Defaults to None.	None
exclude	Iterable[str] | None	The aliases to exclude from the validation. If not provided, no aliases are excluded. Defaults to None.	None

Raises:

Type	Description
ValidationError	If any assignment is invalid.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

def validate_assignments(
    self,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
) -> None:
    """Validate the selector assignments against the resource.

    Args:
        strict: Whether to enforce strict validation. Defaults to ``None``.
        context: The context to use for validation. Defaults to ``None``.
        include: The aliases to include in the validation. If not provided,
            all aliases are included. Defaults to ``None``.
        exclude: The aliases to exclude from the validation. If not
            provided, no aliases are excluded. Defaults to ``None``.

    Raises:
        ValidationError: If any assignment is invalid.
    """
    for alias, value in self.items():
        if include is not None and alias not in include:
            continue
        if exclude is not None and alias in exclude:
            continue
        validator = self.create_type_adapter(alias, resource=self.resource)
        try:
            self[alias] = validator.validate_python(
                value, strict=strict, context=context
            )
        except ValidationError as error:
            raise ValueError(
                f"Invalid resource selector {self!r} for resource class "
                f"{self.resource.__qualname__!r}. The selector assignment "
                f"value {value!r} does not match the expected type for "
                f"the field {alias!r}."
            ) from error

resolve async

resolve(
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> Sequence[_T]

Resolve the key list.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

async def resolve(
    self,
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> Sequence[_T]:
    """Resolve the key list."""
    result = await super().resolve(__session, limit=limit, options=options)
    assert isinstance(result, Sequence)
    return result

resolve_lenient async

resolve_lenient(
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> Sequence[_T]

Resolve the key list in a lenient way.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/selectors.py

async def resolve_lenient(
    self,
    __session: AsyncSession | None = None,
    *,
    limit: int | None = None,
    options: Sequence[ExecutableOption] | None = None,
    **kwargs: Any,
) -> Sequence[_T]:
    """Resolve the key list in a lenient way."""
    result = await super().resolve_lenient(
        __session, limit=limit, options=options
    )
    assert isinstance(result, Sequence)
    return result

plateforme.core.expressions

This module provides utilities for managing expressions such as filters and sorts used in database and API queries within the Plateforme framework.

IncExKey module-attribute

IncExKey = TypeVar('IncExKey', int, str)

A type variable for inclusive and exclusive schema keys.

IncExValue module-attribute

IncExValue = Union[bool, 'IncExSet[Any]', 'IncExDict[Any]']

A type alias for inclusive and exclusive schema values.

IncExSet module-attribute

IncExSet = Set[IncExKey]

A type alias for inclusive and exclusive schema of field sets.

IncExDict module-attribute

IncExDict = Dict[IncExKey, IncExValue]

A type alias for inclusive and exclusive schema of field dictionaries.

IncExObj module-attribute

IncExObj = Union[IncExSet[IncExKey], IncExDict[IncExKey]]

A type alias for inclusive and exclusive schema of field objects.

IncExPredicate module-attribute

IncExPredicate = Union[
    Any,
    Tuple[Sequence[Any], Any],
    Tuple[Mapping[str, Any], Any],
    Tuple[Sequence[Any], Mapping[str, Any], Any],
]

A type alias for inclusive and exclusive predicate values.

This type represents different ways to specify matching criteria for values. It supports both direct value matching and callable predicates with various argument patterns:

1. Any Direct predicate where it checks whether the target matches the value, or is contained within the collection if the provided value is a list, set, or tuple, e.g.: 42 or [1, 2, 3].

2. Tuple[Sequence[Any], Any] Callable predicate with positional arguments where it calls the function with the given positional arguments and matches against the value(s), e.g.: (['foo', 'bar'], 42) or (['foo', 'bar'], [1, 2, 3]).

3. Tuple[Mapping[str, Any], Any] Callable predicate with keyword arguments where it calls the function with the given keyword arguments and matches against the value(s), e.g.: ({'min': 0, 'max': 100}, 42) or ({'foo': 'bar'}, [1, 2, 3]).

4. Tuple[Sequence[Any], Mapping[str, Any], Any] Callable predicate with both positional and keyword arguments where it calls the function with both positional and keyword arguments and matches against the value(s), e.g.: (['foo', 'bar'], {'min': 0}, 42).

Note

For all forms, the last value can be either a single value or a collection (list, set, tuple) of values. The predicate matches if the result equals the single value or is contained within the collection.

FilterDict module-attribute

FilterDict = Dict[str, 'FilterValue']

A type alias for a filter criteria dictionary.

FilterValue module-attribute

FilterValue = Union[List[Condition], FilterDict]

A type alias for a filter criterion or nested criteria dictionary.

FilterVar module-attribute

FilterVar = TypeVar('FilterVar', FilterDict, FilterValue)

A type variable for a filter dictionary or value.

SortList module-attribute

SortList = List['SortValue']

A type alias for a sort criteria list.

SortValue module-attribute

SortValue = Tuple[str, Ordering]

A type alias for a sort criterion value.

SortVar module-attribute

SortVar = TypeVar('SortVar', SortList, SortValue)

A type variable for a sort list or a sort value.

IncEx

IncEx()

Bases: Generic[IncExKey]

Inclusive and exclusive field schema factory.

This class is used to define the inclusive and exclusive field schema with a set or nested dictionary structure of integer field indexes or string field keys. For dictionaries, the values can be of type boolean to include or exclude the field based on the context.

Examples:

• The expression {1, 2, 3} when inclusive will only include the values from the first three fields.
• The expression {'user': {'id': True}, 'groups': True} when exclusive will exclude the user id and groups field.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

def __init__(self) -> None:
    raise ValueError(
        "Cannot directly instantiate the inclusive/exclusive schema "
        "factory class. Use the `from_args` method instead."
    )

from_args classmethod

from_args(
    *args: IncExKey | IncExSet[IncExKey],
) -> IncExSet[IncExKey]

from_args(
    *args: IncExKey | IncExDict[IncExKey],
    **kwargs: IncExValue,
) -> IncExDict[IncExKey]

from_args(
    *args: IncExKey | IncExObj[IncExKey],
    **kwargs: IncExValue,
) -> IncExObj[IncExKey]

Create a new schema from the given arguments.

Parameters:

Name	Type	Description	Default
*args	IncExKey | IncExObj[IncExKey]	The inclusive/exclusive set or nested dictionary structure of integer field indexes or string field keys.	()
**kwargs	IncExValue	The inclusive/exclusive field keys with either boolean values to include or exclude the field based on the context, or set or nested dictionary structure of integer field indexes or string field keys.	{}

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@classmethod
def from_args(
    cls, *args: IncExKey | IncExObj[IncExKey], **kwargs: IncExValue
) -> IncExObj[IncExKey]:
    """Create a new schema from the given arguments.

    Args:
        *args: The inclusive/exclusive set or nested dictionary
            structure of integer field indexes or string field keys.
        **kwargs: The inclusive/exclusive field keys with either
            boolean values to include or exclude the field based on the
            context, or set or nested dictionary structure of integer
            field indexes or string field keys.
    """
    # Initialize schema and type
    schema: IncExObj[IncExKey]
    schema_type: type[IncExKey] | None
    if kwargs:
        schema = dict()
        schema_type = str  # type: ignore[assignment]
    elif any(isinstance(arg, dict) for arg in args):
        schema = dict()
        schema_type = None
    else:
        schema = set()
        schema_type = None

    # Validate schema type
    def validate_schema_type(*keys: IncExKey) ->  None:
        nonlocal schema_type
        for key in keys:
            if not isinstance(key, (int, str)):
                raise ValueError(
                    f"Invalid inclusive/exclusive schema key type: "
                    f"{type(key)!r}."
                )
            if schema_type is None:
                schema_type = type(key)
            elif not isinstance(key, schema_type):
                raise ValueError(
                    f"Incompatible inclusive/exclusive schema key "
                    f"types: {schema_type!r} and {type(key)!r}."
                )

    # Parse schema arguments
    for arg in args:
        if isinstance(arg, dict):
            validate_schema_type(*arg.keys())
            assert isinstance(schema, dict)
            for arg_key, arg_value in arg.items():
                if isinstance(arg_value, bool):
                    schema[arg_key] = arg_value
                else:
                    schema[arg_key] = cls.from_args(arg_value)
        elif isinstance(arg, (list, set, tuple)):
            validate_schema_type(*arg)
            if isinstance(schema, dict):
                schema.update({key: True for key in arg})
            else:
                schema.update(arg)
        else:
            validate_schema_type(arg)
            if isinstance(schema, dict):
                schema[arg] = True
            else:
                schema.add(arg)

    # Parse schema keyword arguments
    assert isinstance(schema, dict)
    for kwarg_key, kwarg_value in kwargs.items():
        if isinstance(kwarg_value, bool):
            schema[kwarg_key] = kwarg_value  # type: ignore[index]
        else:
            schema[kwarg_key] = cls.from_args(  # type: ignore[index]
                kwarg_value
            )

    return schema

ExpressionIssue dataclass

ExpressionIssue(obj: Any, message: str)

Bases: Representation

An issue with an expression.

obj instance-attribute

obj: Any

The expression inner object with the issue.

message instance-attribute

message: str

The issue message.

OperatorDefinition dataclass

OperatorDefinition(
    symbol: str,
    types: tuple[type, ...],
    sql: Callable[
        [InstrumentedAttribute[Any], Any],
        BinaryExpression[Any],
    ],
)

Bases: Representation

An operator definition.

symbol instance-attribute

symbol: str

The symbol of the operator.

types instance-attribute

types: tuple[type, ...]

The types of values the operator can be applied to.

sql instance-attribute

sql: Callable[
    [InstrumentedAttribute[Any], Any], BinaryExpression[Any]
]

The SQL representation of the operator.

Operator

Bases: Enum

An enumeration of operators used in expressions.

EQUAL class-attribute instance-attribute

EQUAL = OperatorDefinition(
    symbol="eq",
    types=(float, int, str),
    sql=lambda attr, comp: attr == comp,
)

Equal to (default)

NOT_EQUAL class-attribute instance-attribute

NOT_EQUAL = OperatorDefinition(
    symbol="ne",
    types=(float, int, str),
    sql=lambda attr, comp: attr != comp,
)

Not equal to

GREATER_THAN class-attribute instance-attribute

GREATER_THAN = OperatorDefinition(
    symbol="gt",
    types=(float, int, str),
    sql=lambda attr, comp: attr > comp,
)

Greater than

GREATER_THAN_OR_EQUAL class-attribute instance-attribute

GREATER_THAN_OR_EQUAL = OperatorDefinition(
    symbol="ge",
    types=(float, int, str),
    sql=lambda attr, comp: attr >= comp,
)

Greater than or equal to

LESS_THAN class-attribute instance-attribute

LESS_THAN = OperatorDefinition(
    symbol="lt",
    types=(float, int, str),
    sql=lambda attr, comp: attr < comp,
)

Less than

LESS_THAN_OR_EQUAL class-attribute instance-attribute

LESS_THAN_OR_EQUAL = OperatorDefinition(
    symbol="le",
    types=(float, int, str),
    sql=lambda attr, comp: attr <= comp,
)

Less than or equal to

LIKE class-attribute instance-attribute

LIKE = OperatorDefinition(
    symbol="like",
    types=(str,),
    sql=lambda attr, comp: like(
        replace("*", "%") if isinstance(comp, str) else comp
    ),
)

Matching the like pattern

NOT_LIKE class-attribute instance-attribute

NOT_LIKE = OperatorDefinition(
    symbol="nlike",
    types=(str,),
    sql=lambda attr, comp: ~like(
        replace("*", "%") if isinstance(comp, str) else comp
    ),
)

Not matching the like pattern

IN class-attribute instance-attribute

IN = OperatorDefinition(
    symbol="in",
    types=(list,),
    sql=lambda attr, comp: in_(comp),
)

In the list of

NOT_IN class-attribute instance-attribute

NOT_IN = OperatorDefinition(
    symbol="nin",
    types=(list,),
    sql=lambda attr, comp: ~in_(comp),
)

Not in the list of

get staticmethod

get(symbol: str) -> Operator

Get the operator associated with the given symbol.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def get(symbol: str) -> 'Operator':
    """Get the operator associated with the given symbol."""
    for operator in Operator:
        if operator.value.symbol == symbol:
            return operator
    raise ValueError(f"Invalid operator symbol: {symbol!r}.")

Symbol

Bases: StrEnum

An enumeration of symbols used in expressions.

AND class-attribute instance-attribute

AND = '&'

The query separator symbol.

ASSIGNMENT class-attribute instance-attribute

ASSIGNMENT = '='

The assignment symbol.

LIST class-attribute instance-attribute

LIST = ','

The list separator symbol.

OPERATOR class-attribute instance-attribute

OPERATOR = '~'

The operator separator symbol.

PIPE class-attribute instance-attribute

PIPE = ':'

The pipe symbol.

REFERENCE class-attribute instance-attribute

REFERENCE = '$'

The field reference symbol.

SPREAD class-attribute instance-attribute

SPREAD = '*'

The spread symbol.

STATEMENT class-attribute instance-attribute

STATEMENT = ';'

The statement separator symbol.

Condition dataclass

Condition(value: Any, operator: Operator = EQUAL)

Bases: Representation

Condition options for a filter clause in a query.

value instance-attribute

value: Any

The value to filter the field.

operator class-attribute instance-attribute

operator: Operator = EQUAL

The operator to filter the field.

reference class-attribute instance-attribute

reference: bool = field(default=False, init=False)

Whether the value is a reference to another field.

apply

apply(
    query: Select[Any],
    attr: InstrumentedAttribute[Any],
    *,
    refs: dict[str, InstrumentedAttribute[Any]]
    | None = None,
) -> Select[Any]

Apply the filtering condition to the given query and attribute.

Parameters:

Name	Type	Description	Default
query	Select[Any]	The query to apply the filtering condition to.	required
attr	InstrumentedAttribute[Any]	The attribute to apply the filtering condition to.	required
refs	dict[str, InstrumentedAttribute[Any]] | None	The reference attributes mapping used to resolve the condition reference value.	None

Returns:

Type	Description
Select[Any]	The query with the filtering condition applied.

Raises:

Type	Description
ValueError	When the condition reference value does not exist in the reference attributes mapping.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

def apply(
    self,
    query: 'Select[Any]',
    attr: 'InstrumentedAttribute[Any]',
    *,
    refs: dict[str, 'InstrumentedAttribute[Any]'] | None = None,
) -> 'Select[Any]':
    """Apply the filtering condition to the given query and attribute.

    Args:
        query: The query to apply the filtering condition to.
        attr: The attribute to apply the filtering condition to.
        refs: The reference attributes mapping used to resolve the
            condition reference value.

    Returns:
        The query with the filtering condition applied.

    Raises:
        ValueError: When the condition reference value does not exist in
            the reference attributes mapping.
    """
    # Resolve condition value
    if self.reference:
        if refs is None or self.value not in refs:
            raise ValueError(
                f"The reference field {self.value!r} does not exist in "
                f"the references mapping."
            )
        value = refs[self.value]
    else:
        value = self.value

    return query.filter(self.operator.value.sql(attr, value))

Ordering dataclass

Ordering(
    direction: Literal["asc", "desc"] = "asc",
    nulls: Literal["first", "last"] | None = None,
)

Bases: Representation

Options to sort a field in a query.

direction class-attribute instance-attribute

direction: Literal['asc', 'desc'] = 'asc'

The direction to sort the field.

nulls class-attribute instance-attribute

nulls: Literal['first', 'last'] | None = None

The position of null values in the sorted results.

apply

apply(
    query: Select[Any], attr: InstrumentedAttribute[Any]
) -> Select[Any]

Apply the ordering options to the given query and attribute.

Parameters:

Name	Type	Description	Default
query	Select[Any]	The query to apply the ordering options to.	required
attr	InstrumentedAttribute[Any]	The attribute to apply the ordering options to.	required

Returns:

Type	Description
Select[Any]	The query with the ordering options applied.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

def apply(
    self, query: 'Select[Any]', attr: 'InstrumentedAttribute[Any]'
) -> 'Select[Any]':
    """Apply the ordering options to the given query and attribute.

    Args:
        query: The query to apply the ordering options to.
        attr: The attribute to apply the ordering options to.

    Returns:
        The query with the ordering options applied.
    """
    expression: UnaryExpression[Any]

    # Add direction clause
    if self.direction == 'asc':
        expression = attr.asc()
    else:
        expression = attr.desc()

    # Add nulls clause
    if self.nulls == 'first':
        expression = expression.nullsfirst()
    elif self.nulls == 'last':
        expression = expression.nullslast()

    return query.order_by(expression)

Filter

Filter(*criteria: FilterDict, **update: FilterValue | str)

Bases: FilterDict

A filter object to be used in a query.

It is used to filter the results of a query based on the given criteria. The filter criteria can be validated from nested dictionaries and list of conditions, or from raw strings formatted as a semicolon separated list of conditions and optionally prefixed with an operator, separated by a tilde character. Additionally, a filter condition can reference another field by prefixing the target field key value with a dollar character.

Attributes:

Name	Type	Description
references		A list of reference attributes used within the filter criteria to resolve the reference values.

Examples:

The expression {'price': 'gt~0;lt~1', 'cost': 'le~$price'} will be parsed into:

>>> {
...     'price': [
...         {'value': 0, 'operator': 'gt'},
...         {'value': 1, 'operator': 'lt'},
...     ],
...     'cost': [
...         {'value': 'price', 'operator': 'le', 'reference': True},
...     ],
... }

The expression {'user.name': 'like~Al*', user.groups*: 'foo'} will be parsed into:

>>> {
...     'user': {
...         'name': [
...             {'value': 'Al*', 'operator': 'like'},
...         ],
...         'groups*': [
...             {'value': 'foo', 'operator': 'eq'},
...         ],
...     },
... }

Initialize the filter object with the given criteria and updates.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

def __init__(
    self, *criteria: FilterDict, **update: FilterValue | str
) -> None:
    """Initialize the filter object with the given criteria and updates."""
    # Merge filter criteria
    result: FilterDict = {}
    for obj in criteria:
        result = self.merge(result, obj)

    # Update filter criteria
    for key, value in update.items():
        if isinstance(value, str):
            try:
                result[key] = self.parse_criteria(value)
            except ValueError:
                result[key] = self.parse_criterion(value)
        else:
            result[key] = value

    result = self.unflatten(result)

    issues, references = self.inspect(result)
    if issues:
        raise ValueError(
            "The provided filter criteria have the following issues:\n"
            "\n".join(f'- {issue}' for issue in issues)
        )

    self.references = references
    self.update(result)

inspect staticmethod

inspect(
    value: FilterValue,
) -> tuple[list[ExpressionIssue], list[str]]

Inspect the filter criteria from the given value.

Parameters:

Name	Type	Description	Default
value	FilterValue	The filter criteria to inspect.	required

Returns:

Type	Description
list[ExpressionIssue]	A tuple with a list of issues with the filter criteria and a list
list[str]	of references used in the filter criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def inspect(value: FilterValue) -> tuple[list[ExpressionIssue], list[str]]:
    """Inspect the filter criteria from the given value.

    Args:
        value: The filter criteria to inspect.

    Returns:
        A tuple with a list of issues with the filter criteria and a list
        of references used in the filter criteria.
    """
    issues: list[ExpressionIssue] = []
    references: list[str] = []

    # Inspect dictionary criteria
    if value and isinstance(value, dict):
        for alias, value in value.items():
            if not re.match(RegexPattern.ALIAS_EXP, alias):
                issues.append(ExpressionIssue(
                    tuple([alias, value]),
                    f"Invalid filter field alias: {alias!r}.",
                ))
                continue
            inner_issues, inner_references = Filter.inspect(value)
            issues.extend(inner_issues)
            references.extend(inner_references)

    # Inspect list criteria
    elif value and isinstance(value, list):
        for condition in value:
            if not isinstance(condition, Condition):
                issues.append(ExpressionIssue(
                    condition,
                    f"Invalid filter condition: {condition!r}.",
                ))
                continue
            if condition.reference:
                references.append(condition.value)

    # Invalid criteria
    else:
        issues.append(ExpressionIssue(
            value,
            f"Invalid filter criteria. Expected a non-empty dictionary or "
            f"list, but got {value!r}.",
        ))

    return issues, list(dict.fromkeys(references))

merge staticmethod

merge(obj: FilterVar, other: FilterVar) -> FilterVar

Copy and merge the given filter criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def merge(obj: FilterVar, other: FilterVar) -> FilterVar:
    """Copy and merge the given filter criteria."""
    # Merge dictionary criteria
    if isinstance(obj, dict):
        if not isinstance(other, dict):
            raise ValueError(
                f"Incompatible filter criteria. Expected a dictionary to "
                f"merge with {obj!r}, but got {other!r}."
            )
        result = obj.copy()
        for key, value in other.items():
            if key in obj:
                result[key] = Filter.merge(obj[key], value)
            else:
                result[key] = value
        return result

    # Merge list criteria
    if isinstance(obj, list):
        if not isinstance(other, list):
            raise ValueError(
                f"Incompatible filter criteria. Expected a list to merge "
                f"with {obj!r}, but got {other!r}."
            )
        return [*obj, *other]

    raise ValueError(
        f"Incompatible filter criteria. Expected a dictionary or a list "
        f"but got {obj!r} and {other!r}."
    )

flatten staticmethod

flatten(obj: FilterDict) -> FilterDict

Flatten a nested dictionary of filter criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def flatten(obj: FilterDict) -> FilterDict:
    """Flatten a nested dictionary of filter criteria."""
    result: FilterDict = {}
    for alias, value in obj.items():
        if isinstance(value, dict):
            value = Filter.flatten(value)
            for key, conditions in value.items():
                result[f'{alias}.{key}'] = conditions
        else:
            result[alias] = value
    return result

unflatten staticmethod

unflatten(obj: FilterDict) -> FilterDict

Unflatten a dictionary of filter criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def unflatten(obj: FilterDict) -> FilterDict:
    """Unflatten a dictionary of filter criteria."""
    result: FilterDict = {}

    for key, value in obj.items():
        # Validate alias
        node = result
        aliases = key.split('.')
        for alias in aliases[:-1]:
            if alias not in node:
                node[alias] = {}
            elif not isinstance(node[alias], dict):
                raise ValueError(
                    f"Incompatible filter criteria for alias {alias!r}. "
                    f"Expected a dictionary, but got {node[alias]!r}."
                )
            node = node[alias]  # type: ignore[assignment]
        alias = aliases[-1]

        # Validate value
        if isinstance(value, dict):
            value = Filter.unflatten(value)
        if alias in node:
            node[alias] = Filter.merge(node[alias], value)
        else:
            node[alias] = value

    return result

parse_criterion staticmethod

parse_criterion(string: str) -> list[Condition]

Parse the filter criterion conditions from the given string.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def parse_criterion(string: str) -> list[Condition]:
    """Parse the filter criterion conditions from the given string."""
    pattern = (
        r'^(?:(?P<operator>[a-z]+)' + Symbol.OPERATOR + ')?'
        r'(?P<value>.+)$'
    )

    # Parse filter criterion
    conditions = []
    for condition in string.split(Symbol.STATEMENT):
        # Validate filter condition
        match = re.match(pattern, condition.strip())
        if not match:
            raise ValueError(f"Invalid filter condition: {condition!r}.")

        # Validate filter operator and value
        operator = Operator.get(match['operator'] or 'eq')
        value = match['value'].split(Symbol.LIST)
        if len(value) == 1:
            value = value[0]
        if not isinstance(value, operator.value.types):
            raise ValueError(
                f"Invalid filter value for {operator!r}: {value!r}."
            )

        conditions.append(Condition(value, operator=operator))

    return conditions

parse_criteria staticmethod

parse_criteria(string: str) -> FilterDict

Parse the filter criteria from the given string.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def parse_criteria(string: str) -> FilterDict:
    """Parse the filter criteria from the given string."""
    return {
        key: Filter.parse_criterion(value)
        for key, value in [
            criterion.split(Symbol.ASSIGNMENT)
            for criterion in string.split(Symbol.AND)
        ]
    }

validate classmethod

validate(obj: Any) -> Self

Validate the filter criteria from the given object.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@classmethod
def validate(cls, obj: Any) -> Self:
    """Validate the filter criteria from the given object."""
    if isinstance(obj, dict):
        return cls(**obj)
    elif isinstance(obj, str):
        return cls(**cls.parse_criteria(obj))
    raise ValueError(f"Invalid filter criteria: {obj!r}")

Sort

Sort(*criteria: SortValue | str)

Bases: SortList

A sort object to be used in a query.

It is used to store the sort criteria to order the results of a query. It can be validated from a raw string formatted as a comma or semicolon separated list of field key as alias fullnames; optionally prefixed with a plus or minus character for respectively ascending or descending order, and suffixed with direction and nulls clauses piped with a colon character.

The direction can be specified as asc or desc and the nulls as first and last. By default, the direction is set to asc and the nulls to None, i.e. the default database provider behavior.

Examples:

The expression price:nf will be parsed into:

>>> [('price', {'direction': 'asc', 'nulls': 'first'})

The expression user.name:asc:nl,-created_at will be parsed into:

>>> [
...     ('user.name', {'direction': 'asc', 'nulls': 'last'}),
...     ('created_at', {'direction': 'desc', 'nulls': None}),
... ]

Initialize the sort object with the given criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

def __init__(self, *criteria: SortValue | str) -> None:
    """Initialize the sort object with the given criteria."""
    result = []
    for obj in criteria:
        if isinstance(obj, str):
            try:
                result.extend(self.parse_criteria(obj))
            except ValueError:
                result.append(self.parse_criterion(obj))
        else:
            result.append(obj)

    issues = self.inspect(result)
    if issues:
        raise ValueError(
            "The provided sort criteria have the following issues:\n"
            "\n".join(f'- {issue}' for issue in issues)
        )

    self.extend(result)

inspect staticmethod

inspect(criteria: SortList) -> list[ExpressionIssue]

Inspect the sort criterion key and options from the given criteria.

Parameters:

Name	Type	Description	Default
criteria	SortList	The sort criteria to inspect.	required

Returns:

Type	Description
list[ExpressionIssue]	A list of issues with the sort criteria.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def inspect(criteria: SortList) -> list[ExpressionIssue]:
    """Inspect the sort criterion key and options from the given criteria.

    Args:
        criteria: The sort criteria to inspect.

    Returns:
        A list of issues with the sort criteria.
    """
    issues: list[ExpressionIssue] = []

    # Inspect list criteria
    for criterion in criteria:
        if not isinstance(criterion, tuple) or len(criterion) != 2:
            issues.append(ExpressionIssue(
                criterion,
                "Invalid sort criterion. Expected a tuple of field key "
                "and options, but got {criterion!r}.",
            ))
            continue

        key, options = criterion
        if not re.match(RegexPattern.NAME_EXP, key):
            issues.append(ExpressionIssue(
                criterion,
                f"Invalid sort field key: {key!r}.",
            ))
        if not isinstance(options, Ordering):
            issues.append(ExpressionIssue(
                criterion,
                f"Invalid sort options: {options!r}.",
            ))

    return issues

parse_criterion staticmethod

parse_criterion(string: str) -> SortValue

Parse the sort criterion key and options from the given string.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def parse_criterion(string: str) -> SortValue:
    """Parse the sort criterion key and options from the given string."""
    pattern = (
        r'^(?P<sign>[+-])?'
        r'(?P<key>' + RegexPattern.NAME_EXP[1:-1] + r')'
        r'(?::(?:(?P<direction>asc|desc)|(?P<nulls>nf|nl)))?$'
    )

    # Parse sort criterion
    match = re.match(pattern, string.strip())
    if not match:
        raise ValueError(f"Invalid sort criterion: {string!r}.")
    sign = -1 if match['sign'] == '-' else 1
    sign *= -1 if match['direction'] == 'desc' else 1

    return (
        match['key'],
        Ordering(
            direction='asc' if sign == 1 else 'desc',
            nulls='first' if match['nulls'] == 'nf'
                else 'last' if match['nulls'] == 'nl'
                else None,
        ),
    )

parse_criteria staticmethod

parse_criteria(string: str) -> SortList

Parse the sort criteria from the given string.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@staticmethod
def parse_criteria(string: str) -> SortList:
    """Parse the sort criteria from the given string."""
    separators = r'[' + Symbol.LIST + Symbol.STATEMENT + r']'
    return [Sort.parse_criterion(c) for c in string.split(separators)]

validate classmethod

validate(obj: Any) -> Self

Validate the sort criteria from the given object.

Source code in .venv/lib/python3.12/site-packages/plateforme/core/expressions.py

@classmethod
def validate(cls, obj: Any) -> Self:
    """Validate the sort criteria from the given object."""
    if isinstance(obj, (list, set, tuple)):
        return cls(*obj)
    elif isinstance(obj, str):
        return cls(*cls.parse_criteria(obj))
    raise ValueError(f"Invalid sort criteria: {obj!r}")