Services

plateforme.core.services

This module provides a collection of classes and utilities to define and interact with services within the Plateforme framework. It includes base classes for creating services, utilities for service configuration and management, and a set of protocols and metaclasses to define service behavior and characteristics.

The BaseService base class provides a base structure for defining services for all services in the Plateforme framework, providing common functionality and enforcing a structure for service implementations.

The BaseServiceWithSpec base class extends the BaseService class to include the declaration of a resource class or specification associated with the service.

The ServiceConfig class provides a configuration class for services, allowing services to define and validate configuration options for individual services.

Service `module-attribute`

Service = TypeVar('Service', bound='BaseService')

A type variable for service classes.

ServiceType `module-attribute`

ServiceType = Type['BaseService']

A type variable for service types.

ServiceFacade

Bases: Protocol

A service facades protocol to manage services.

ServiceWithSpecFacade

Bases: ServiceFacade, Protocol

A service facade protocol with specification support.

ServiceConfigDict

Bases: TypedDict

A service class configuration dictionary.

name `instance-attribute`

name: str

The name of the service. It must adhere to a specific ALIAS pattern as defined in the framework's regular expressions repository. It is inferred from the snake case version of the service name.

include `instance-attribute`

include: list[str] | set[str] | None

The method names to include when binding the service.

exclude `instance-attribute`

exclude: list[str] | set[str] | None

The method names to exclude when binding the service.

include_method `instance-attribute`

include_method: list[APIMethod] | set[APIMethod] | None

The HTTP methods to include when binding the service.

exclude_method `instance-attribute`

exclude_method: list[APIMethod] | set[APIMethod] | None

The HTTP methods to exclude when binding the service.

include_mode `instance-attribute`

include_mode: list[APIMode] | set[APIMode] | None

The method modes to include when binding the service.

exclude_mode `instance-attribute`

exclude_mode: list[APIMode] | set[APIMode] | None

The method modes to exclude when binding the service.

limit `instance-attribute`

limit: int

The default page size limit for service queries. It is used to limit the number of results returned per page when paginating results.

page_size `instance-attribute`

page_size: int

The default page size for service queries. It is used to determine the number of results returned per page when paginating results.

auto_apply_spec `instance-attribute`

auto_apply_spec: bool

Whether to automatically apply the specification when binding the service to a facade owner.

ServiceConfig

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

Bases: ConfigWrapper

A service class 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="service", frozen=True, init=False
)

The configuration owner type set to service. 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.

name `class-attribute` `instance-attribute`

name: str = Deferred

The name of the service. It must adhere to a specific ALIAS pattern as defined in the framework's regular expressions repository. It is inferred from the snake case version of the service name.

include `class-attribute` `instance-attribute`

include: list[str] | set[str] | None = ConfigField(
    default=None
)

The method names to include when binding the service.

exclude `class-attribute` `instance-attribute`

exclude: list[str] | set[str] | None = ConfigField(
    default=None
)

The method names to exclude when binding the service.

include_method `class-attribute` `instance-attribute`

include_method: list[APIMethod] | set[APIMethod] | None = (
    ConfigField(default=None)
)

The HTTP methods to include when binding the service.

exclude_method `class-attribute` `instance-attribute`

exclude_method: list[APIMethod] | set[APIMethod] | None = (
    ConfigField(default=None)
)

The HTTP methods to exclude when binding the service.

include_mode `class-attribute` `instance-attribute`

include_mode: list[APIMode] | set[APIMode] | None = (
    ConfigField(default=None)
)

The method modes to include when binding the service.

exclude_mode `class-attribute` `instance-attribute`

exclude_mode: list[APIMode] | set[APIMode] | None = (
    ConfigField(default=None)
)

The method modes to exclude when binding the service.

limit `class-attribute` `instance-attribute`

limit: int = ConfigField(default=20)

The default page size limit for service queries. It is used to limit the number of results returned per page when paginating results.

page_size `class-attribute` `instance-attribute`

page_size: int = ConfigField(default=20)

The default page size for service queries. It is used to determine the number of results returned per page when paginating results.

auto_apply_spec `class-attribute` `instance-attribute`

auto_apply_spec: bool = ConfigField(default=True)

Whether to automatically apply the specification when binding the service to a facade owner.

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)

post_init

post_init() -> None

Post-initialization steps for the service configuration.

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

def post_init(self) -> None:
    """Post-initialization steps for the service configuration."""
    # Skip post-initialization if the configuration owner is not set
    service = self.__config_owner__
    if service is None:
        return

    self.setdefault('name', to_name_case(service.__name__, ('all', None)))
    if not re.match(RegexPattern.ALIAS, self.name):
        raise ValueError(
            f"Invalid service name {self.name!r} for service "
            f"configuration {service.__qualname__!r}."
        )

    if self.limit < 1:
        raise ValueError(
            f"Invalid page size limit {self.limit!r} for service "
            f"configuration {service.__qualname__!r}."
        )

    if self.page_size < 1:
        raise ValueError(
            f"Invalid page size {self.page_size!r} for service "
            f"configuration {service.__qualname__!r}."
        )

ServiceMeta

Bases: ABCMeta, ConfigurableMeta

A metaclass for service 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_spec

collect_spec(
    bases: tuple[type, ...], namespace: dict[str, Any]
) -> SpecType | None

Collect the specification from the given bases and namespace.

It collects the specification from the given bases and namespace by extracting the specification from the original bases annotation if it is a generic subclass of the base service with specification class.

Parameters:

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

Returns:

Type	Description
`SpecType \| None`	The specification if found, otherwise `None`.

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

def collect_spec(
    cls, bases: tuple[type, ...], namespace: dict[str, Any], /,
) -> SpecType | None:
    """Collect the specification from the given bases and namespace.

    It collects the specification from the given bases and namespace by
    extracting the specification from the original bases annotation if
    it is a generic subclass of the base service with specification class.

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

    Returns:
        The specification if found, otherwise ``None``.
    """
    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, 'BaseServiceWithSpec'):
            continue
        args = typing.get_args(meta_base)
        if len(args) == 1 \
                and isimplclass_lenient(args[0], BaseSpec, max_depth=1):
            return args[0]  # type: ignore
        raise TypeError(
            f"Generic argument for the `BaseServiceWithSpec` class must "
            f"be a subclass of the `BaseSpec` class. Got: {args}."
        )

    return None

BaseService

BaseService(name: str | None = None)

Bases: Configurable[ServiceConfig]

A base class for services.

It provides a base class for creating service objects that can be attached to a service facade owner. Services can be customized through their service_config attribute and optionally integrated with a specification.

When combined with a specification, services are augmented with the specification specific methods, attributes and model schemas. This allows services to have a dynamic behavior based on the provided specification.

See the BaseServiceWithSpec class for more details on how to integrate services with a specification.

Attributes:

Name	Type	Description
`__config__`	`ServiceConfig \| ConfigDict[ServiceConfigDict]`	The configuration class setter for the service.
`__config_spec__`	`SpecType \| None`	The specification class for the service.
`service_config`	`ServiceConfig`	The configuration class for the service.
`service_methods`	`dict[str, FunctionLenientType]`	A dictionary of public methods of the service.
`service_owner`	`ServiceFacade \| ServiceWithSpecFacade \| None`	The service facade owner of the service.

Note

The service class should not be directly instantiated, and it should be subclassed to define a new service class. The service class should define public methods that can be bound to the service facade owner.

Initialize the service.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	The name of the service. If provided, it will update the service name in the service configuration. This can be useful when multiple services have the same name and need to be uniquely identified.	`None`

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

def __init__(self, name: str | None = None) -> None:
    """Initialize the service.

    Args:
        name: The name of the service. If provided, it will update the
            service name in the service configuration. This can be useful
            when multiple services have the same name and need to be
            uniquely identified.
    """
    super().__init__()

    if name is not None:
        self.service_config.name = name

    object.__setattr__(self, 'service_owner', None)

BaseServiceWithSpec

BaseServiceWithSpec(name: str | None = None)

Bases: BaseService, Generic[Spec]

A base class for services with a specification.

It provides a base class for creating service objects that can be attached to a service facade owner. Services can be customized through their service_config attribute.

The provided specification in the generic argument, that much implements the base specification protocol such as resources, augments the service with specific methods, attributes and model schemas. This allows services to have a dynamic behavior.

Attributes:

Name	Type	Description
`service_config`	`ServiceConfig`	The configuration class for the service.
`service_methods`	`dict[str, FunctionLenientType]`	A dictionary of public methods of the service.
`service_owner`	`ServiceFacade \| ServiceWithSpecFacade \| None`	The service facade owner of the service.

Note

The service class should not be directly instantiated, and it should be subclassed to define a new service class. The service class should define public methods that can be bound to the service facade owner.

Initialize the service.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	The name of the service. If provided, it will update the service name in the service configuration. This can be useful when multiple services have the same name and need to be uniquely identified.	`None`

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

def __init__(self, name: str | None = None) -> None:
    """Initialize the service.

    Args:
        name: The name of the service. If provided, it will update the
            service name in the service configuration. This can be useful
            when multiple services have the same name and need to be
            uniquely identified.
    """
    super().__init__(name)

CRUDService

CRUDService(name: str | None = None)

Bases: BaseServiceWithSpec[CRUDSpec]

The CRUD services.

Initialize the service.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	The name of the service. If provided, it will update the service name in the service configuration. This can be useful when multiple services have the same name and need to be uniquely identified.	`None`

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

def __init__(self, name: str | None = None) -> None:
    """Initialize the service.

    Args:
        name: The name of the service. If provided, it will update the
            service name in the service configuration. This can be useful
            when multiple services have the same name and need to be
            uniquely identified.
    """
    super().__init__(name)

ResourceFieldInfo `module-attribute`

ResourceFieldInfo = FieldInfo['BaseResource']

A type alias for a resource field information.

Package

Package(
    name: str,
    module: ModuleType,
    *,
    info: dict[Any, Any] | None = None,
    settings: PackageSettings | None = None,
)

Bases: Representation

A package within the Plateforme framework.

It is initialized with a python module name. It checks for a package configuration to ensure the given module name corresponds to an importable package within the Plateforme framework. Each package is uniquely represented as a singleton object, identified by its module name. This process guarantees the consistent and correct identification of Plateforme packages for further operations.

The package class exposes a catalog attribute that maps the alias and slug names within the package namespace to their respective objects, either a resource type, an association (i.e. a tuple of one or two linked fields), or a service method. It also provides access to the package resources, dependencies, and dependents, as well as the package implementations within the current application context.

Attributes:

Name	Type	Description
`_impls`	`dict[Plateforme \| None, PackageImpl]`	The registered package implementations as a dictionary mapping the application context to the package implementation.
`module`	`ModuleType`	The package module object.
`name`	`str`	The package module name that is unique within the entire runtime environment.
`info`	`dict[Any, Any] \| None`	Additional information about the package.
`catalog`	`Catalog`	A dictionary mapping the alias and slug names within the package namespace to their respective objects, either a resource type, an association, or a service method.
`metadata`	`MetaData`	The package resource metadata.
`registry`	`Registry`	The package resource registry.

Initialize a package.

Note

The import_package method should be used to retrieve a package instance based on the provided module name.

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

def __init__(
    self,
    name: str,
    module: ModuleType,
    *,
    info: dict[Any, Any] | None = None,
    settings: PackageSettings | None = None,
) -> None:
    """Initialize a package.

    Note:
        The `import_package` method should be used to retrieve a package
        instance based on the provided module name.
    """
    object.__setattr__(self, '_impls', {})
    object.__setattr__(self, 'module', module)
    object.__setattr__(self, 'name', name)
    object.__setattr__(self, 'info', info)

    self._add_impl(None, settings=settings, auto_generated=True)

    catalog = Catalog(self)
    metadata = MetaData(
        schema=self.impl.settings.namespace,
        schema_factory=lambda: self.impl.namespace.alias,
        info=info,
    )
    registry = Registry(metadata=metadata)

    object.__setattr__(self, 'catalog', catalog)
    object.__setattr__(self, 'metadata', metadata)
    object.__setattr__(self, 'registry', registry)

impl `property`

impl: PackageImpl

The current package implementation based on the app context.

A read-only property that returns the package implementation based on the application context. If no context is available, it returns the package default implementation. Otherwise, it raises an error if the package implementation is not available in the current application context.

ResourceConfig

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

Bases: ModelConfig

A resource class 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()

stub `class-attribute` `instance-attribute`

stub: bool = ConfigField(
    default=False, frozen=True, init=False
)

Whether the model is a stub model. This is set to True when a collected bare model has a final stub no-operation statement. Defaults to False.

alias `class-attribute` `instance-attribute`

alias: str = Deferred

The alias name of the model. It must adhere to a specific ALIAS pattern as defined in the framework's regular expressions repository. It is inferred from the snake case version of the resolved model identifier.

slug `class-attribute` `instance-attribute`

slug: str = Deferred

The slug name of the model. It must adhere to a specific SLUG pattern as defined in the framework's regular expressions repository. It is inferred from the pluralized kebab case version of the model identifier.

title `class-attribute` `instance-attribute`

title: Annotated[str, pydantic] = Deferred

The human-readable name of the model. It must adhere to a specific TITLE pattern as defined in the framework's regular expressions repository. It is inferred from the titleized version of the model identifier.

str_to_lower `class-attribute` `instance-attribute`

str_to_lower: Annotated[bool, pydantic] = False

Whether to convert strings to lowercase. Defaults to False.

str_to_upper `class-attribute` `instance-attribute`

str_to_upper: Annotated[bool, pydantic] = False

Whether to convert strings to uppercase. Defaults to False.

str_strip_whitespace `class-attribute` `instance-attribute`

str_strip_whitespace: Annotated[bool, pydantic] = False

Whether to strip whitespace from strings. Defaults to False.

str_min_length `class-attribute` `instance-attribute`

str_min_length: Annotated[int, pydantic] = 0

The minimum length for strings. Defaults to None.

str_max_length `class-attribute` `instance-attribute`

str_max_length: Annotated[int | None, pydantic] = None

The maximum length for strings. Defaults to None.

frozen `class-attribute` `instance-attribute`

frozen: Annotated[bool, pydantic] = False

Whether to freeze the configuration. Defaults to False.

populate_by_name `class-attribute` `instance-attribute`

populate_by_name: Annotated[bool, pydantic] = False

Whether to populate fields by name. Defaults to False.

use_enum_values `class-attribute` `instance-attribute`

use_enum_values: Annotated[bool, pydantic] = False

Whether to use enum values. Defaults to False.

validate_assignment `class-attribute` `instance-attribute`

validate_assignment: Annotated[bool, pydantic] = False

Whether to validate assignments. Defaults to False.

arbitrary_types_allowed `class-attribute` `instance-attribute`

arbitrary_types_allowed: Annotated[bool, pydantic] = False

Whether to allow arbitrary types. Defaults to False.

from_attributes `class-attribute` `instance-attribute`

from_attributes: Annotated[bool, pydantic] = False

Whether to set attributes from the configuration. Defaults to False.

loc_by_alias `class-attribute` `instance-attribute`

loc_by_alias: Annotated[bool, pydantic] = True

Whether to use the alias for error locs. Defaults to True.

alias_generator `class-attribute` `instance-attribute`

alias_generator: Annotated[
    Callable[[str], str] | AliasGenerator | None, pydantic
] = to_name_case

A callable or alias generator to create aliases for the model. Defaults to None.

ignored_types `class-attribute` `instance-attribute`

ignored_types: Annotated[tuple[type, ...], pydantic] = ()

A tuple of types to ignore. Defaults to an empty tuple.

allow_inf_nan `class-attribute` `instance-attribute`

allow_inf_nan: Annotated[bool, pydantic] = True

Whether to allow infinity and NaN. Defaults to True.

json_schema_extra `class-attribute` `instance-attribute`

json_schema_extra: Annotated[
    JsonSchemaExtraCallable | None, pydantic
] = None

Dictionary of extra JSON schema properties. Defaults to None.

json_encoders `class-attribute` `instance-attribute`

json_encoders: Annotated[
    dict[type[object], JsonEncoder] | None, pydantic
] = None

A dictionary of custom JSON encoders for specific types. Defaults to None.

strict `class-attribute` `instance-attribute`

strict: Annotated[bool, pydantic] = False

Whether to make the configuration strict. Defaults to False.

revalidate_instances `class-attribute` `instance-attribute`

revalidate_instances: Annotated[
    Literal["always", "never", "subclass-instances"],
    pydantic,
] = "never"

When and how to revalidate models and dataclasses during validation. Defaults to never.

ser_json_timedelta `class-attribute` `instance-attribute`

ser_json_timedelta: Annotated[
    Literal["iso8601", "float"], pydantic
] = "iso8601"

The format of JSON serialized timedeltas. Defaults to iso8601.

ser_json_bytes `class-attribute` `instance-attribute`

ser_json_bytes: Annotated[
    Literal["utf8", "base64"], pydantic
] = "utf8"

The encoding of JSON serialized bytes. Defaults to utf8.

ser_json_inf_nan `class-attribute` `instance-attribute`

ser_json_inf_nan: Annotated[
    Literal["null", "constants"], pydantic
] = "null"

The encoding of JSON serialized infinity and NaN float values. Accepts the string values of 'null' and 'constants'. Defaults to 'null'.

validate_default `class-attribute` `instance-attribute`

validate_default: Annotated[bool, pydantic] = False

Whether to validate default values during validation. Defaults to False.

validate_return `class-attribute` `instance-attribute`

validate_return: Annotated[bool, pydantic] = False

Whether to validate return values during validation. Defaults to False.

protected_namespaces `class-attribute` `instance-attribute`

protected_namespaces: Annotated[
    tuple[str, ...], pydantic
] = Deferred

A tuple of strings that prevent models to have field which conflict with them. The provided namespaces are added to the internally forbidden and protected ones, model_, resource_, and service_. Defaults to an empty tuple.

hide_input_in_errors `class-attribute` `instance-attribute`

hide_input_in_errors: Annotated[bool, pydantic] = False

Whether to hide inputs when printing errors. Defaults to False.

plugin_settings `class-attribute` `instance-attribute`

plugin_settings: Annotated[
    dict[str, object] | None, pydantic
] = None

A dictionary of settings for plugins. Defaults to None.

schema_generator `class-attribute` `instance-attribute`

schema_generator: Annotated[type[Any] | None, pydantic] = (
    None
)

A custom core schema generator class to use when generating JSON schemas. Defaults to None.

json_schema_serialization_defaults_required `class-attribute` `instance-attribute`

json_schema_serialization_defaults_required: Annotated[
    bool, pydantic
] = False

Whether fields with default values should be marked as required in the serialization schema. Defaults to False.

json_schema_mode_override `class-attribute` `instance-attribute`

json_schema_mode_override: Annotated[
    Literal["validation", "serialization"] | None, pydantic
] = None

If not None, the specified mode will be used to generate the JSON schema regardless of what mode was passed to the function call. Defaults to None.

coerce_numbers_to_str `class-attribute` `instance-attribute`

coerce_numbers_to_str: Annotated[bool, pydantic] = False

If True, enables automatic coercion of any Number type to str in lax (non-strict) mode. Defaults to False.

regex_engine `class-attribute` `instance-attribute`

regex_engine: Annotated[
    Literal["rust-regex", "python-re"], pydantic
] = "rust-regex"

The regex engine to use for pattern validation. Defaults to rust-regex.

validation_error_cause `class-attribute` `instance-attribute`

validation_error_cause: Annotated[bool, pydantic] = False

If True, Python exceptions that were part of a validation failure will be shown as an exception group as a cause. Can be useful for debugging. Defaults to False.

use_attribute_docstrings `class-attribute` `instance-attribute`

use_attribute_docstrings: Annotated[bool, pydantic] = False

Whether docstrings of attributes should be used for field descriptions. Defaults to False.

cache_strings `class-attribute` `instance-attribute`

cache_strings: Annotated[
    bool | Literal["all", "keys", "none"], pydantic
] = True

Whether to cache strings to avoid constructing new Python objects. Enabling this setting should significantly improve validation performance while increasing memory usage slightly. - True or 'all' (default): Cache all strings - 'keys': Cache only dictionary keys - False or 'none': No caching Defaults to True.

type_ `class-attribute` `instance-attribute`

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

The configuration owner type set to resource. 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.

extra `class-attribute` `instance-attribute`

extra: Annotated[
    Literal["allow", "ignore", "forbid"], pydantic
] = ConfigField(default="forbid", frozen=True, init=False)

Extra values are not allowed within a resource instance. This attribute is protected and will initialize to its default value forbid.

defer_build `class-attribute` `instance-attribute`

defer_build: Annotated[bool, pydantic] = ConfigField(
    default=True, frozen=True, init=False
)

Defer building is not allowed for resource instances. This attribute is protected and will initialize to its default value True.

tags `class-attribute` `instance-attribute`

tags: list[str | Enum] | None = None

A list of tags to associate with the resource. Tags are used to group resources and provide additional metadata. It will be added to the generated OpenAPI, visible at /docs. If not provided, the resource slug will be used as the default tag. Defaults to None.

api_max_depth `class-attribute` `instance-attribute`

api_max_depth: int = 2

The maximum depth to walk through the resource path to collect manager methods from resource dependencies. It is used to generate API routes. Defaults to 2.

api_max_selection `class-attribute` `instance-attribute`

api_max_selection: int = 20

The limit of resources to return for the API route selections. It is used when generating the API routes for resources within the application to avoid too many resources being returned. Defaults to 20.

id_strategy `class-attribute` `instance-attribute`

id_strategy: Literal['auto', 'manual', 'hybrid'] = 'auto'

The identifier strategy to use for the resource. It defines how the identifier is generated for the resource and can be set to one of the following values: - auto: Enforce automatic generation of the resource identifier. - manual: Enforce manual specification of the resource identifier. - hybrid: Allow both automatic and manual generation of the resource identifier. Defaults to auto.

id_type `class-attribute` `instance-attribute`

id_type: Literal['integer', 'uuid'] = 'integer'

The identifier type to use for the resource. It defines the data type of the resource identifier and can be set to one of the following values: - integer: Use an integer data type engine for the resource identifier. - uuid: Use a UUID data type engine for the resource identifier. Defaults to integer.

mapper_args `class-attribute` `instance-attribute`

mapper_args: dict[str, Any] = {}

A dictionary of additional arguments to pass to the resource declarative mapper during the resource configuration, i.e. the dictionary of arguments to add to the __mapper_args__ attribute of the resource class. Defaults to an empty dictionary.

use_single_table_inheritance `class-attribute` `instance-attribute`

use_single_table_inheritance: bool = False

It defines the inheritance design pattern to use for database ORM. Defaults to False, using joined table inheritance with separate tables for parent and child classes linked by a foreign key. This approach is generally preferred for its normalization benefits. Setting this to True enables single table inheritance, creating a single table for both parent and child classes with a discriminator column.

Note

Single table inheritance can lead to sparse tables and may impact performance for large and complex hierarchies.

indexes `class-attribute` `instance-attribute`

indexes: tuple[ResourceIndex, ...] = ()

A tuple of resource indexes configurations that define indexing for resource model fields. An index is defined as a dictionary with the following keys: - aliases: A tuple of strings representing the resource field aliases to include in the index. - unique: Whether the index is unique. Defaults to True. Defaults to an empty tuple.

services `class-attribute` `instance-attribute`

services: tuple[
    BaseService | EllipsisType | ServiceType, ...
] = ()

A tuple of services to bind to the resource. The services are used to define the business logic and data access methods for the resource. The services can be defined as instances of BaseService or as service types. The ellipsis ... can be used to insert all services from the parent resource. Defaults to an empty tuple.

specs `class-attribute` `instance-attribute`

specs: tuple[SpecType, ...] = ()

A tuple of specifications to apply to the resource. The specifications are used to define additional resource configurations, schemas, and behaviors. All the specifications from the parent resource are merged with the provided configuration, .i.e. a child resource must inherit all the specifications from the parent resource. Defaults to an empty tuple.

deprecated `class-attribute` `instance-attribute`

deprecated: bool | None = None

A flag indicating whether the resource is deprecated. Defaults to None.

id_autoincrement `property`

id_autoincrement: bool | Literal['auto']

Whether the resource identifier is autoincremented.

id_engine `property`

id_engine: type[IntegerEngine | UuidEngine[UUID4]]

The identifier engine to use for the resource.

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)

post_init

post_init() -> None

Post-initialization steps for the resource configuration.

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

def post_init(self) -> None:
    """Post-initialization steps for the resource configuration."""
    # Skip post-initialization if the configuration owner is not set
    resource = self.__config_owner__
    if resource is None:
        return

    # Model post-initialization
    super().post_init()

    # Resource post-initialization
    fields = resource.resource_fields
    aliases = [field.alias for field in fields.values()]
    indexes: list[ResourceIndex] = []

    # Collect configuration indexes
    for index in self.indexes:
        # Validate index type
        if isinstance(index, (list, set)):
            index = {'aliases': set(index)}
        elif not isinstance(index, dict):
            raise PlateformeError(
                f"The resource {resource.__qualname__!r} has an invalid "
                f"index configuration for entry {index!r}. An index must "
                f"be a dictionary.",
                code='resource-invalid-config',
            )

        # Validate index field aliases length
        if len(index['aliases']) < 2:
            raise PlateformeError(
                f"The resource {resource.__qualname__!r} has an invalid "
                f"index configuration for entry {index!r}. Composite "
                f"indexes defined in the resource configuration must have "
                f"at least two field aliases. Use ``indexed=True`` or "
                f"``unique=True`` for single field indexes.",
                code='resource-invalid-config',
            )

        # Validate index field aliases existence
        for alias in index['aliases']:
            if alias == 'id':
                raise PlateformeError(
                    f"The resource {resource.__qualname__!r} has an "
                    f"invalid index configuration for entry {index!r}. "
                    f"The field alias {alias!r} is reserved and cannot be "
                    f"used in indexes.",
                    code='resource-invalid-config',
                )
            if alias not in aliases:
                raise PlateformeError(
                    f"The resource {resource.__qualname__!r} has an "
                    f"invalid index configuration for entry {index!r}. "
                    f"The field alias {alias!r} is not defined in the "
                    f"resource model.",
                    code='resource-invalid-config',
                )

        # Set alias default name and unique flag
        default_name = self.alias + '_' + '_'.join(index['aliases'])
        index.setdefault('name', default_name)
        index.setdefault('unique', True)

        indexes.append(index)

    # Check for duplicate indexes
    indexes_check = set()
    for index in indexes:
        index_check = frozenset(index['aliases'])
        if index_check in indexes_check:
            raise PlateformeError(
                f"The resource {self.__config_owner__.__qualname__!r} has "
                f"a duplicate index configuration for {index_check!r}.",
                code='resource-invalid-config',
            )
        indexes_check.add(index_check)

    # Update indexes
    if indexes:
        self.indexes = tuple(indexes)

create `async`

create(
    session: AsyncSessionDep,
    __selection: KeyList[CRUDSpec] = Selection(),
    payload: Create | Sequence[Create] = Payload(
        apply_selection=True,
        title="Payload",
        description="The payload to create instances of the associated\n                resource, either a single instance or a sequence of instances.\n                The payload must adhere to the resource schema for the create\n                operation.",
    ),
    return_result: bool = Body(
        default=True,
        title="Return result",
        description="Whether to return the created result. If set to\n                `False`, the transaction will be executed without returning\n                the result.",
    ),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title="Dry-run",
        description="Whether to run the transaction in dry-run mode. If\n                set to `True`, the transaction will be rolled back after\n                execution.",
    ),
) -> Any | None

Create a single or collection of resource instances.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the transaction.	required
`__selection`	`KeyList[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`payload`	`Create \| Sequence[Create]`	The payload to create instances of the associated resource, either a single instance or a sequence of instances.	`Payload(apply_selection=True, title='Payload', description='The payload to create instances of the associated\n resource, either a single instance or a sequence of instances.\n The payload must adhere to the resource schema for the create\n operation.')`
`return_result`	`bool`	Whether to return the created result.	Body(default=True, title='Return result', description='Whether to return the created result. If set to\n `False`, the transaction will be executed without returning\n the result.')
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`dry_run`	`bool`	Whether to run the transaction in dry-run mode.	Body(default=False, title='Dry-run', description='Whether to run the transaction in dry-run mode. If\n set to `True`, the transaction will be rolled back after\n execution.')

Returns:

Type	Description
`Any \| None`	The created resource instances or `None` if return result is
`Any \| None`	disabled, i.e., `return_result=False`.

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

@route.post(
    path='',
    response_model=list[CRUDSpec.Read] | None,
    response_model_serialization=False,
)
async def create(
    self,
    session: AsyncSessionDep,
    __selection: KeyList[CRUDSpec] = Selection(),
    payload: CRUDSpec.Create | Sequence[CRUDSpec.Create] = Payload(
        apply_selection=True,
        title='Payload',
        description="""The payload to create instances of the associated
            resource, either a single instance or a sequence of instances.
            The payload must adhere to the resource schema for the create
            operation.""",
    ),
    return_result: bool = Body(
        default=True,
        title='Return result',
        description="""Whether to return the created result. If set to
            `False`, the transaction will be executed without returning
            the result.""",
    ),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title='Dry-run',
        description="""Whether to run the transaction in dry-run mode. If
            set to `True`, the transaction will be rolled back after
            execution.""",
    ),
) -> Any | None:
    """Create a single or collection of resource instances.

    Args:
        session: The async session dependency to use for the transaction.
        __selection: The key selection dependency to resolve the query.
        payload: The payload to create instances of the associated
            resource, either a single instance or a sequence of instances.
        return_result: Whether to return the created result.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        dry_run: Whether to run the transaction in dry-run mode.

    Returns:
        The created resource instances or ``None`` if return result is
        disabled, i.e., ``return_result=False``.
    """
    async with session.bulk() as bulk:
        # Validate payload
        result = self.resource_adapter.validate_python(
            payload, from_attributes=True
        )
        # Resolve references
        await bulk.resolve(
            raise_errors=True,
            scope='references',
            strategy='hydrate' if return_result else 'bind',
        )
        # Add instances to the session
        session.add_all(result)

    # Commit or rollback the transaction
    if dry_run:
        await session.rollback()
    else:
        if return_result:
            session.sync_session.expire_on_commit = False
        await session.commit()

    # Handle return result
    if not return_result:
        return None
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )

read_one `async`

read_one(
    session: AsyncSessionDep,
    selection: Key[CRUDSpec] = Selection(),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
) -> Any

Read a resource instance.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the query.	required
`selection`	`Key[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`

Returns:

Type	Description
`Any`	The resource instance from the selection.

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

@route.get(
    path='',
    response_model=CRUDSpec.Read,
    response_model_serialization=False,
)
async def read_one(
    self,
    session: AsyncSessionDep,
    selection: Key[CRUDSpec] = Selection(),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
) -> Any:
    """Read a resource instance.

    Args:
        session: The async session dependency to use for the query.
        selection: The key selection dependency to resolve the query.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.

    Returns:
        The resource instance from the selection.
    """
    result = await selection.resolve(session)

    # Handle result
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )[0]

read_many `async`

read_many(
    session: AsyncSessionDep,
    selection: KeyList[CRUDSpec] = Selection(),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    filter: Annotated[
        Filter | None, Depends(filter_dependency)
    ] = Body(
        default=None,
        title="Filter criteria",
        description="The filter criteria to apply to the query. It is\n                specified in the request body as a dictionary of field aliases\n                with their corresponding filter criteria. Additionally, it can\n                be specified directly in the query parameters using a dot\n                notation for field aliases and tilde `~` character for filter\n                criteria. For example, to filter results where the `user.name`\n                field starts with `Al`, the filter criteria can be specified as\n                `.user.name=like~Al*` in the query parameters.",
        examples=[
            ".user.name=like~Al*",
            ".price=gt~1000",
            ".user.name=like~Al*&.price=gt~1000",
            {
                "user": {
                    "name": {
                        "operator": "like",
                        "value": "Al*",
                    }
                },
                "price": {"operator": "gt", "value": 1000},
            },
        ],
    ),
    sort: Sort | None = Query(
        default=None,
        title="Sort criteria",
        description="The sort criteria to apply to the query. It is\n                specified in the query parameters as a comma-separated list of\n                field aliases with an optional prefix of a minus sign `-` for\n                descending order. For example, to sort results by the\n                `user.name` field in descending order, the sort criteria can be\n                specified as `user.name,-price` in the query parameters.\n                Additionally, the direction and nulls position can be piped\n                using a colon `:` character. For example, to sort results by\n                the `price` field in ascending order with nulls first, the sort\n                criteria can be specified as `price:asc:nf` in the query\n                parameters.",
        examples=["user.name,-price", "price:asc:nf"],
    ),
    limit: int | None = Query(
        default=None,
        gt=0,
        title="Limit",
        description="The maximum number of results to return. It is\n                specified in the query parameters as an integer value. Defaults\n                to the service configuration default page size if not\n                specified.",
        examples=[10, 20, 50],
    ),
    offset: int | None = Query(
        default=None,
        ge=0,
        title="Offset",
        description="The number of results to skip before returning the\n                results. It is specified in the query parameters as an integer\n                value. Defaults to `0` if not specified.",
    ),
    page: int | None = Query(
        default=None,
        gt=0,
        title="Page",
        description="The page number to return results for. It is\n                specified in the query parameters as an integer value. Defaults\n                to the first page if not specified.",
    ),
    page_size: int | None = Query(
        default=None,
        gt=0,
        title="Page size",
        description="The number of results to return per page. It is\n                specified in the query parameters as an integer value. The\n                maximum page size is defined by the service configuration.\n                Defaults to the service configuration default page size if not\n                specified.",
    ),
) -> Any

Read a collection of resource instances.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the query.	required
`selection`	`KeyList[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`filter`	`Annotated[Filter \| None, Depends(filter_dependency)]`	The filter criteria to apply to the query.	Body(default=None, title='Filter criteria', description='The filter criteria to apply to the query. It is\n specified in the request body as a dictionary of field aliases\n with their corresponding filter criteria. Additionally, it can\n be specified directly in the query parameters using a dot\n notation for field aliases and tilde `~` character for filter\n criteria. For example, to filter results where the `user.name`\n field starts with `Al`, the filter criteria can be specified as\n `.user.name=like~Al` in the query parameters.', examples=['.user.name=like~Al', '.price=gt~1000', '.user.name=like~Al&.price=gt~1000', {'user': {'name': {'operator': 'like', 'value': 'Al'}}, 'price': {'operator': 'gt', 'value': 1000}}])
`sort`	`Sort \| None`	The sort criteria to apply to the query.	Query(default=None, title='Sort criteria', description='The sort criteria to apply to the query. It is\n specified in the query parameters as a comma-separated list of\n field aliases with an optional prefix of a minus sign `-` for\n descending order. For example, to sort results by the\n `user.name` field in descending order, the sort criteria can be\n specified as `user.name,-price` in the query parameters.\n Additionally, the direction and nulls position can be piped\n using a colon `:` character. For example, to sort results by\n the `price` field in ascending order with nulls first, the sort\n criteria can be specified as `price:asc:nf` in the query\n parameters.', examples=['user.name,-price', 'price:asc:nf'])
`limit`	`int \| None`	The maximum number of results to return.	`Query(default=None, gt=0, title='Limit', description='The maximum number of results to return. It is\n specified in the query parameters as an integer value. Defaults\n to the service configuration default page size if not\n specified.', examples=[10, 20, 50])`
`offset`	`int \| None`	The number of results to skip before returning the results.	Query(default=None, ge=0, title='Offset', description='The number of results to skip before returning the\n results. It is specified in the query parameters as an integer\n value. Defaults to `0` if not specified.')
`page`	`int \| None`	The page number to return results for.	`Query(default=None, gt=0, title='Page', description='The page number to return results for. It is\n specified in the query parameters as an integer value. Defaults\n to the first page if not specified.')`
`page_size`	`int \| None`	The number of results to return per page.	`Query(default=None, gt=0, title='Page size', description='The number of results to return per page. It is\n specified in the query parameters as an integer value. The\n maximum page size is defined by the service configuration.\n Defaults to the service configuration default page size if not\n specified.')`

Returns:

Type	Description
`Any`	The collection of resource instances from the selection.

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

@route.get(
    path='',
    response_model=list[CRUDSpec.Read],
    response_model_serialization=False,
)
async def read_many(
    self,
    session: AsyncSessionDep,
    selection: KeyList[CRUDSpec] = Selection(),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    filter: Annotated[Filter | None, Depends(filter_dependency)] = Body(
        default=None,
        title='Filter criteria',
        description="""The filter criteria to apply to the query. It is
            specified in the request body as a dictionary of field aliases
            with their corresponding filter criteria. Additionally, it can
            be specified directly in the query parameters using a dot
            notation for field aliases and tilde `~` character for filter
            criteria. For example, to filter results where the `user.name`
            field starts with `Al`, the filter criteria can be specified as
            `.user.name=like~Al*` in the query parameters.""",
        examples=[
            '.user.name=like~Al*',
            '.price=gt~1000',
            '.user.name=like~Al*&.price=gt~1000',
            {
                'user': {'name': {'operator': 'like', 'value': 'Al*'}},
                'price': {'operator': 'gt', 'value': 1000},
            },
        ],
    ),
    sort: Sort | None = Query(
        default=None,
        title='Sort criteria',
        description="""The sort criteria to apply to the query. It is
            specified in the query parameters as a comma-separated list of
            field aliases with an optional prefix of a minus sign `-` for
            descending order. For example, to sort results by the
            `user.name` field in descending order, the sort criteria can be
            specified as `user.name,-price` in the query parameters.
            Additionally, the direction and nulls position can be piped
            using a colon `:` character. For example, to sort results by
            the `price` field in ascending order with nulls first, the sort
            criteria can be specified as `price:asc:nf` in the query
            parameters.""",
        examples=['user.name,-price', 'price:asc:nf'],
    ),
    limit: int | None = Query(
        default=None,
        gt=0,
        title='Limit',
        description="""The maximum number of results to return. It is
            specified in the query parameters as an integer value. Defaults
            to the service configuration default page size if not
            specified.""",
        examples=[10, 20, 50],
    ),
    offset: int | None = Query(
        default=None,
        ge=0,
        title='Offset',
        description="""The number of results to skip before returning the
            results. It is specified in the query parameters as an integer
            value. Defaults to `0` if not specified.""",
    ),
    page: int | None = Query(
        default=None,
        gt=0,
        title='Page',
        description="""The page number to return results for. It is
            specified in the query parameters as an integer value. Defaults
            to the first page if not specified.""",
    ),
    page_size: int | None = Query(
        default=None,
        gt=0,
        title='Page size',
        description="""The number of results to return per page. It is
            specified in the query parameters as an integer value. The
            maximum page size is defined by the service configuration.
            Defaults to the service configuration default page size if not
            specified.""",
    ),
) -> Any:
    """Read a collection of resource instances.

    Args:
        session: The async session dependency to use for the query.
        selection: The key selection dependency to resolve the query.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        filter: The filter criteria to apply to the query.
        sort: The sort criteria to apply to the query.
        limit: The maximum number of results to return.
        offset: The number of results to skip before returning the results.
        page: The page number to return results for.
        page_size: The number of results to return per page.

    Returns:
        The collection of resource instances from the selection.
    """
    # Resolve parameters
    limit = (
        limit
        or page_size
        or self.service_config.get(
            'page_size', self.resource_config.api_max_selection
        )
    )
    limit = min(limit, self.resource_config.api_max_selection)
    offset = offset or 0
    page = page or 1
    page_size = page_size or self.service_config.page_size

    # Build query
    query = selection.build_query(
        filter_criteria=filter,
        sort_criteria=sort,
    )
    query = query.offset((page - 1) * page_size + offset)
    query = query.limit(limit)

    # Execute query and return results
    buffer = await session.execute(query)
    result = buffer.unique().scalars().all()

    # Handle result
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )

update_one `async`

update_one(
    session: AsyncSessionDep,
    selection: Key[CRUDSpec] = Selection(),
    payload: Update = Payload(
        apply_selection=False,
        title="Payload",
        description="The payload to update the instance of the associated\n                resource. The payload must adhere to the resource schema for\n                the update operation.",
    ),
    return_result: bool = Body(
        default=True,
        title="Return result",
        description="Whether to return the updated result. If set to\n                `False`, the transaction will be executed without returning\n                the result.",
    ),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title="Dry-run",
        description="Whether to run the transaction in dry-run mode. If\n                set to `True`, the transaction will be rolled back after\n                execution.",
    ),
) -> Any | None

Update a resource instance.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the transaction.	required
`selection`	`Key[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`payload`	`Update`	The payload to update the associated resource instance.	`Payload(apply_selection=False, title='Payload', description='The payload to update the instance of the associated\n resource. The payload must adhere to the resource schema for\n the update operation.')`
`return_result`	`bool`	Whether to return the updated result.	Body(default=True, title='Return result', description='Whether to return the updated result. If set to\n `False`, the transaction will be executed without returning\n the result.')
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`dry_run`	`bool`	Whether to run the transaction in dry-run mode.	Body(default=False, title='Dry-run', description='Whether to run the transaction in dry-run mode. If\n set to `True`, the transaction will be rolled back after\n execution.')

Returns:

Type	Description
`Any \| None`	The updated resource instance or `None` if return result is
`Any \| None`	disabled, i.e., `return_result=False`.

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

@route.patch(
    path='',
    response_model=CRUDSpec.Read,
    response_model_serialization=False,
)
async def update_one(
    self,
    session: AsyncSessionDep,
    selection: Key[CRUDSpec] = Selection(),
    payload: CRUDSpec.Update = Payload(
        apply_selection=False,
        title='Payload',
        description="""The payload to update the instance of the associated
            resource. The payload must adhere to the resource schema for
            the update operation.""",
    ),
    return_result: bool = Body(
        default=True,
        title='Return result',
        description="""Whether to return the updated result. If set to
            `False`, the transaction will be executed without returning
            the result.""",
    ),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title='Dry-run',
        description="""Whether to run the transaction in dry-run mode. If
            set to `True`, the transaction will be rolled back after
            execution.""",
    ),
) -> Any | None:
    """Update a resource instance.

    Args:
        session: The async session dependency to use for the transaction.
        selection: The key selection dependency to resolve the query.
        payload: The payload to update the associated resource instance.
        return_result: Whether to return the updated result.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        dry_run: Whether to run the transaction in dry-run mode.

    Returns:
        The updated resource instance or ``None`` if return result is
        disabled, i.e., ``return_result=False``.
    """
    # Prepare update
    update = payload.model_dump(mode='raw', exclude_unset=True)

    # Execute query and update result
    result_source = await selection.resolve(session)
    result_dirty = {**result_source.resource_dump(mode='raw'), **update}

    async with session.bulk() as bulk:
        # Validate payload
        result_staged = self.resource_adapter.validate_python(
            result_dirty, from_attributes=True
        )
        # Resolve references
        await bulk.resolve(
            raise_errors=True,
            scope='references',
            strategy='hydrate' if return_result else 'bind',
        )

    # Merge staged instances into the session
    result = []
    for instance_staged in result_staged:
        instance = await session.merge(instance_staged)
        result.append(instance)

    # Commit or rollback the transaction
    if dry_run:
        await session.rollback()
    else:
        if return_result:
            session.sync_session.expire_on_commit = False
        await session.commit()

    # Handle return result
    if not return_result:
        return None
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )[0]

update_many `async`

update_many(
    session: AsyncSessionDep,
    selection: KeyList[CRUDSpec] = Selection(),
    payload: Update = Payload(
        apply_selection=False,
        title="Payload",
        description="The payload to update instances of the associated\n                resource. The payload must adhere to the resource schema for\n                the update operation.",
    ),
    return_result: bool = Body(
        default=True,
        title="Return result",
        description="Whether to return the updated result. If set to\n                `False`, the transaction will be executed without returning\n                the result.",
    ),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    filter: Annotated[
        Filter | None, Depends(filter_dependency)
    ] = Body(
        default=None,
        title="Filter criteria",
        description="The filter criteria to apply to the query. It is\n                specified in the request body as a dictionary of field aliases\n                with their corresponding filter criteria. Additionally, it can\n                be specified directly in the query parameters using a dot\n                notation for field aliases and tilde `~` character for filter\n                criteria. For example, to filter results where the `user.name`\n                field starts with `Al`, the filter criteria can be specified as\n                `.user.name=like~Al*` in the query parameters.",
        examples=[
            ".user.name=like~Al*",
            ".price=gt~1000",
            ".user.name=like~Al*&.price=gt~1000",
            {
                "user": {
                    "name": {
                        "operator": "like",
                        "value": "Al*",
                    }
                },
                "price": {"operator": "gt", "value": 1000},
            },
        ],
    ),
    sort: Sort | None = Query(
        default=None,
        title="Sort criteria",
        description="The sort criteria to apply to the query. It is\n                specified in the query parameters as a comma-separated list of\n                field aliases with an optional prefix of a minus sign `-` for\n                descending order. For example, to sort results by the\n                `user.name` field in descending order, the sort criteria can be\n                specified as `user.name,-price` in the query parameters.\n                Additionally, the direction and nulls position can be piped\n                using a colon `:` character. For example, to sort results by\n                the `price` field in ascending order with nulls first, the sort\n                criteria can be specified as `price:asc:nf` in the query\n                parameters.",
        examples=["user.name,-price", "price:asc:nf"],
    ),
    limit: int | None = Query(
        default=None,
        gt=0,
        title="Limit",
        description="The maximum number of results to return. It is\n                specified in the query parameters as an integer value. Defaults\n                to the service configuration default page size if not\n                specified.",
        examples=[10, 20, 50],
    ),
    offset: int | None = Query(
        default=None,
        ge=0,
        title="Offset",
        description="The number of results to skip before returning the\n                results. It is specified in the query parameters as an integer\n                value. Defaults to `0` if not specified.",
    ),
    page: int | None = Query(
        default=None,
        gt=0,
        title="Page",
        description="The page number to return results for. It is\n                specified in the query parameters as an integer value. Defaults\n                to the first page if not specified.",
    ),
    page_size: int | None = Query(
        default=None,
        gt=0,
        title="Page size",
        description="The number of results to return per page. It is\n                specified in the query parameters as an integer value. The\n                maximum page size is defined by the service configuration.\n                Defaults to the service configuration default page size if not\n                specified.",
    ),
    dry_run: bool = Body(
        default=False,
        title="Dry-run",
        description="Whether to run the transaction in dry-run mode. If\n                set to `True`, the transaction will be rolled back after\n                execution.",
    ),
) -> Any | None

Update a collection of resource instances.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the transaction.	required
`selection`	`KeyList[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`payload`	`Update`	The payload to update instances of the associated resource.	`Payload(apply_selection=False, title='Payload', description='The payload to update instances of the associated\n resource. The payload must adhere to the resource schema for\n the update operation.')`
`return_result`	`bool`	Whether to return the updated result.	Body(default=True, title='Return result', description='Whether to return the updated result. If set to\n `False`, the transaction will be executed without returning\n the result.')
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`filter`	`Annotated[Filter \| None, Depends(filter_dependency)]`	The filter criteria to apply to the query.	Body(default=None, title='Filter criteria', description='The filter criteria to apply to the query. It is\n specified in the request body as a dictionary of field aliases\n with their corresponding filter criteria. Additionally, it can\n be specified directly in the query parameters using a dot\n notation for field aliases and tilde `~` character for filter\n criteria. For example, to filter results where the `user.name`\n field starts with `Al`, the filter criteria can be specified as\n `.user.name=like~Al` in the query parameters.', examples=['.user.name=like~Al', '.price=gt~1000', '.user.name=like~Al&.price=gt~1000', {'user': {'name': {'operator': 'like', 'value': 'Al'}}, 'price': {'operator': 'gt', 'value': 1000}}])
`sort`	`Sort \| None`	The sort criteria to apply to the query.	Query(default=None, title='Sort criteria', description='The sort criteria to apply to the query. It is\n specified in the query parameters as a comma-separated list of\n field aliases with an optional prefix of a minus sign `-` for\n descending order. For example, to sort results by the\n `user.name` field in descending order, the sort criteria can be\n specified as `user.name,-price` in the query parameters.\n Additionally, the direction and nulls position can be piped\n using a colon `:` character. For example, to sort results by\n the `price` field in ascending order with nulls first, the sort\n criteria can be specified as `price:asc:nf` in the query\n parameters.', examples=['user.name,-price', 'price:asc:nf'])
`limit`	`int \| None`	The maximum number of results to return.	`Query(default=None, gt=0, title='Limit', description='The maximum number of results to return. It is\n specified in the query parameters as an integer value. Defaults\n to the service configuration default page size if not\n specified.', examples=[10, 20, 50])`
`offset`	`int \| None`	The number of results to skip before returning the results.	Query(default=None, ge=0, title='Offset', description='The number of results to skip before returning the\n results. It is specified in the query parameters as an integer\n value. Defaults to `0` if not specified.')
`page`	`int \| None`	The page number to return results for.	`Query(default=None, gt=0, title='Page', description='The page number to return results for. It is\n specified in the query parameters as an integer value. Defaults\n to the first page if not specified.')`
`page_size`	`int \| None`	The number of results to return per page.	`Query(default=None, gt=0, title='Page size', description='The number of results to return per page. It is\n specified in the query parameters as an integer value. The\n maximum page size is defined by the service configuration.\n Defaults to the service configuration default page size if not\n specified.')`
`dry_run`	`bool`	Whether to run the transaction in dry-run mode.	Body(default=False, title='Dry-run', description='Whether to run the transaction in dry-run mode. If\n set to `True`, the transaction will be rolled back after\n execution.')

Returns:

Type	Description
`Any \| None`	The updated resource instances or `None` if return result is
`Any \| None`	disabled, i.e., `return_result=False`.

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

@route.patch(
    path='',
    response_model=list[CRUDSpec.Read],
    response_model_serialization=False,
)
async def update_many(
    self,
    session: AsyncSessionDep,
    selection: KeyList[CRUDSpec] = Selection(),
    payload: CRUDSpec.Update = Payload(
        apply_selection=False,
        title='Payload',
        description="""The payload to update instances of the associated
            resource. The payload must adhere to the resource schema for
            the update operation.""",
    ),
    return_result: bool = Body(
        default=True,
        title='Return result',
        description="""Whether to return the updated result. If set to
            `False`, the transaction will be executed without returning
            the result.""",
    ),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    filter: Annotated[Filter | None, Depends(filter_dependency)] = Body(
        default=None,
        title='Filter criteria',
        description="""The filter criteria to apply to the query. It is
            specified in the request body as a dictionary of field aliases
            with their corresponding filter criteria. Additionally, it can
            be specified directly in the query parameters using a dot
            notation for field aliases and tilde `~` character for filter
            criteria. For example, to filter results where the `user.name`
            field starts with `Al`, the filter criteria can be specified as
            `.user.name=like~Al*` in the query parameters.""",
        examples=[
            '.user.name=like~Al*',
            '.price=gt~1000',
            '.user.name=like~Al*&.price=gt~1000',
            {
                'user': {'name': {'operator': 'like', 'value': 'Al*'}},
                'price': {'operator': 'gt', 'value': 1000},
            },
        ],
    ),
    sort: Sort | None = Query(
        default=None,
        title='Sort criteria',
        description="""The sort criteria to apply to the query. It is
            specified in the query parameters as a comma-separated list of
            field aliases with an optional prefix of a minus sign `-` for
            descending order. For example, to sort results by the
            `user.name` field in descending order, the sort criteria can be
            specified as `user.name,-price` in the query parameters.
            Additionally, the direction and nulls position can be piped
            using a colon `:` character. For example, to sort results by
            the `price` field in ascending order with nulls first, the sort
            criteria can be specified as `price:asc:nf` in the query
            parameters.""",
        examples=['user.name,-price', 'price:asc:nf'],
    ),
    limit: int | None = Query(
        default=None,
        gt=0,
        title='Limit',
        description="""The maximum number of results to return. It is
            specified in the query parameters as an integer value. Defaults
            to the service configuration default page size if not
            specified.""",
        examples=[10, 20, 50],
    ),
    offset: int | None = Query(
        default=None,
        ge=0,
        title='Offset',
        description="""The number of results to skip before returning the
            results. It is specified in the query parameters as an integer
            value. Defaults to `0` if not specified.""",
    ),
    page: int | None = Query(
        default=None,
        gt=0,
        title='Page',
        description="""The page number to return results for. It is
            specified in the query parameters as an integer value. Defaults
            to the first page if not specified.""",
    ),
    page_size: int | None = Query(
        default=None,
        gt=0,
        title='Page size',
        description="""The number of results to return per page. It is
            specified in the query parameters as an integer value. The
            maximum page size is defined by the service configuration.
            Defaults to the service configuration default page size if not
            specified.""",
    ),
    dry_run: bool = Body(
        default=False,
        title='Dry-run',
        description="""Whether to run the transaction in dry-run mode. If
            set to `True`, the transaction will be rolled back after
            execution.""",
    ),
) -> Any | None:
    """Update a collection of resource instances.

    Args:
        session: The async session dependency to use for the transaction.
        selection: The key selection dependency to resolve the query.
        payload: The payload to update instances of the associated
            resource.
        return_result: Whether to return the updated result.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        filter: The filter criteria to apply to the query.
        sort: The sort criteria to apply to the query.
        limit: The maximum number of results to return.
        offset: The number of results to skip before returning the results.
        page: The page number to return results for.
        page_size: The number of results to return per page.
        dry_run: Whether to run the transaction in dry-run mode.

    Returns:
        The updated resource instances or ``None`` if return result is
        disabled, i.e., ``return_result=False``.
    """
    # Prepare update
    update = payload.model_dump(mode='raw', exclude_unset=True)

    # Resolve parameters
    limit = (
        limit
        or page_size
        or self.service_config.get(
            'page_size', self.resource_config.api_max_selection
        )
    )
    limit = min(limit, self.resource_config.api_max_selection)
    offset = offset or 0
    page = page or 1
    page_size = page_size or self.service_config.page_size

    # Build query
    query = selection.build_query(
        filter_criteria=filter,
        sort_criteria=sort,
    )
    query = query.offset((page - 1) * page_size + offset)
    query = query.limit(limit)

    # Execute query and update result
    buffer = await session.execute(query)
    result_source = buffer.unique().scalars().all()
    result_dirty: list[dict[str, Any]] = []
    for instance_source in result_source:
        result_dirty.append({
            **instance_source.resource_dump(mode='raw'),
            **update,
        })

    async with session.bulk() as bulk:
        # Validate payload
        result_staged = self.resource_adapter.validate_python(
            result_dirty, from_attributes=True
        )
        # Resolve references
        await bulk.resolve(
            raise_errors=True,
            scope='references',
            strategy='hydrate' if return_result else 'bind',
        )

    # Merge staged instances into the session
    result = []
    for instance_staged in result_staged:
        instance = await session.merge(instance_staged)
        result.append(instance)

    # Commit or rollback the transaction
    if dry_run:
        await session.rollback()
    else:
        if return_result:
            session.sync_session.expire_on_commit = False
        await session.commit()

    # Handle return result
    if not return_result:
        return None
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )

upsert `async`

upsert(
    session: AsyncSessionDep,
    __selection: KeyList[CRUDSpec] = Selection(),
    payload: Upsert | Sequence[Upsert] = Payload(
        apply_selection=True,
        title="Payload",
        description="The payload to upsert instances of the associated\n                resource, either a single instance or a sequence of instances.\n                The payload must adhere to the resource schema for the upsert\n                operation.",
    ),
    on_conflict: Literal["update", "skip"] = Body(
        default="update",
        title="On conflict",
        description="The conflict resolution strategy to apply when\n                upserting instances. It can be either `update` to update the\n                existing instances or `skip` to skip the existing instances.\n                ",
    ),
    return_result: bool = Body(
        default=True,
        title="Return result",
        description="Whether to return the upserted result. If set to\n                `False`, the transaction will be executed without returning\n                the result.",
    ),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title="Dry-run",
        description="Whether to run the transaction in dry-run mode. If\n                set to `True`, the transaction will be rolled back after\n                execution.",
    ),
) -> Any | None

Upsert a single or collection of resource instances.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the transaction.	required
`__selection`	`KeyList[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`payload`	`Upsert \| Sequence[Upsert]`	The payload to upsert instances of the associated resource, either a single instance or a sequence of instances.	`Payload(apply_selection=True, title='Payload', description='The payload to upsert instances of the associated\n resource, either a single instance or a sequence of instances.\n The payload must adhere to the resource schema for the upsert\n operation.')`
`on_conflict`	`Literal['update', 'skip']`	The conflict resolution strategy to apply when upserting instances.	Body(default='update', title='On conflict', description='The conflict resolution strategy to apply when\n upserting instances. It can be either `update` to update the\n existing instances or `skip` to skip the existing instances.\n ')
`return_result`	`bool`	Whether to return the upserted result.	Body(default=True, title='Return result', description='Whether to return the upserted result. If set to\n `False`, the transaction will be executed without returning\n the result.')
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`dry_run`	`bool`	Whether to run the transaction in dry-run mode.	Body(default=False, title='Dry-run', description='Whether to run the transaction in dry-run mode. If\n set to `True`, the transaction will be rolled back after\n execution.')

Returns:

Type	Description
`Any \| None`	The upserted resource instances or `None` if return result is
`Any \| None`	disabled, i.e., `return_result=False`.

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

@route.put(
    path='',
    response_model=CRUDSpec.Read | None,
    response_model_serialization=False,
)
async def upsert(
    self,
    session: AsyncSessionDep,
    __selection: KeyList[CRUDSpec] = Selection(),
    payload: CRUDSpec.Upsert | Sequence[CRUDSpec.Upsert] = Payload(
        apply_selection=True,
        title='Payload',
        description="""The payload to upsert instances of the associated
            resource, either a single instance or a sequence of instances.
            The payload must adhere to the resource schema for the upsert
            operation.""",
    ),
    on_conflict: Literal['update', 'skip'] = Body(
        default='update',
        title='On conflict',
        description="""The conflict resolution strategy to apply when
            upserting instances. It can be either `update` to update the
            existing instances or `skip` to skip the existing instances.
            """,
    ),
    return_result: bool = Body(
        default=True,
        title='Return result',
        description="""Whether to return the upserted result. If set to
            `False`, the transaction will be executed without returning
            the result.""",
    ),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title='Dry-run',
        description="""Whether to run the transaction in dry-run mode. If
            set to `True`, the transaction will be rolled back after
            execution.""",
    ),
) -> Any | None:
    """Upsert a single or collection of resource instances.

    Args:
        session: The async session dependency to use for the transaction.
        __selection: The key selection dependency to resolve the query.
        payload: The payload to upsert instances of the associated
            resource, either a single instance or a sequence of instances.
        on_conflict: The conflict resolution strategy to apply when
            upserting instances.
        return_result: Whether to return the upserted result.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        dry_run: Whether to run the transaction in dry-run mode.

    Returns:
        The upserted resource instances or ``None`` if return result is
        disabled, i.e., ``return_result=False``.
    """
    async with session.bulk() as bulk:
        # Validate payload
        _ = self.resource_adapter.validate_python(
            payload, from_attributes=True
        )
        # Resolve references
        await bulk.resolve(
            raise_errors=True,
            scope='references',
            strategy='hydrate' if return_result else 'bind',
        )
        # Resolve values
        await bulk.resolve(
            raise_errors=False,
            scope='values',
            strategy='hydrate' if return_result else 'bind',
        )
        # Retrieve results
        result_resolved = \
            bulk.get(self.resource, resolved=True, scope='values') \
            if on_conflict == 'update' else []
        result_unresolved = \
            bulk.get(self.resource, resolved=False, scope='values')

    # Flush unresolved transient instances
    session.add_all(result_unresolved)
    await session.flush()

    # Merge resolved instances into the session
    result = result_unresolved
    for instance_staged in result_resolved:
        instance = await session.merge(instance_staged)
        result.append(instance)

    # Commit or rollback the transaction
    if dry_run:
        await session.rollback()
    else:
        if return_result:
            session.sync_session.expire_on_commit = False
        await session.commit()

    # Handle return result
    if not return_result:
        return None
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )

delete_one `async`

delete_one(
    session: AsyncSessionDep,
    selection: Key[CRUDSpec] = Selection(),
    return_result: bool = Body(
        default=False,
        title="Return result",
        description="Whether to return the deleted result. If set to\n                `False`, the transaction will be executed without returning\n                the result.",
    ),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title="Dry-run",
        description="Whether to run the transaction in dry-run mode. If\n                set to `True`, the transaction will be rolled back after\n                execution.",
    ),
) -> Any | None

Delete a resource instance.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the query.	required
`selection`	`Key[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`return_result`	`bool`	Whether to return the deletion result.	Body(default=False, title='Return result', description='Whether to return the deleted result. If set to\n `False`, the transaction will be executed without returning\n the result.')
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`dry_run`	`bool`	Whether to run the transaction in dry-run mode.	Body(default=False, title='Dry-run', description='Whether to run the transaction in dry-run mode. If\n set to `True`, the transaction will be rolled back after\n execution.')

Returns:

Type	Description
`Any \| None`	The deleted resource instance or `None` if return result is
`Any \| None`	disabled, i.e., `return_result=False`.

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

@route.delete(
    path='',
    response_model=CRUDSpec.Read | None,
    response_model_serialization=False,
)
async def delete_one(
    self,
    session: AsyncSessionDep,
    selection: Key[CRUDSpec] = Selection(),
    return_result: bool = Body(
        default=False,
        title='Return result',
        description="""Whether to return the deleted result. If set to
            `False`, the transaction will be executed without returning
            the result.""",
    ),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    dry_run: bool = Body(
        default=False,
        title='Dry-run',
        description="""Whether to run the transaction in dry-run mode. If
            set to `True`, the transaction will be rolled back after
            execution.""",
    ),
) -> Any | None:
    """Delete a resource instance.

    Args:
        session: The async session dependency to use for the query.
        selection: The key selection dependency to resolve the query.
        return_result: Whether to return the deletion result.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        dry_run: Whether to run the transaction in dry-run mode.

    Returns:
        The deleted resource instance or ``None`` if return result is
        disabled, i.e., ``return_result=False``.
    """
    result = await selection.resolve(session)

    # Commit or rollback the transaction
    if dry_run:
        await session.rollback()
    else:
        if return_result:
            session.sync_session.expire_on_commit = False
        await session.delete(result)
        await session.commit()

    # Handle return result
    if not return_result:
        return None
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )[0]

delete_many `async`

delete_many(
    session: AsyncSessionDep,
    selection: KeyList[CRUDSpec] = Selection(),
    return_result: bool = Body(
        default=False,
        title="Return result",
        description="Whether to return the deleted result. If set to\n                `False`, the transaction will be executed without returning\n                the result.",
    ),
    include: IncEx | None = Body(
        default=None,
        title="Include fields",
        description="The fields to include in the query results.",
        examples=[
            ["user.name", "user.email"],
            {"user": ["name", "email"], "product": False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title="Exclude fields",
        description="The fields to exclude from the query results.",
        examples=[
            ["user.password", "product.price"],
            {"user": ["password"], "product": True},
        ],
    ),
    filter: Annotated[
        Filter | None, Depends(filter_dependency)
    ] = Body(
        default=None,
        title="Filter criteria",
        description="The filter criteria to apply to the query. It is\n                specified in the request body as a dictionary of field aliases\n                with their corresponding filter criteria. Additionally, it can\n                be specified directly in the query parameters using a dot\n                notation for field aliases and tilde `~` character for filter\n                criteria. For example, to filter results where the `user.name`\n                field starts with `Al`, the filter criteria can be specified as\n                `.user.name=like~Al*` in the query parameters.",
        examples=[
            ".user.name=like~Al*",
            ".price=gt~1000",
            ".user.name=like~Al*&.price=gt~1000",
            {
                "user": {
                    "name": {
                        "operator": "like",
                        "value": "Al*",
                    }
                },
                "price": {"operator": "gt", "value": 1000},
            },
        ],
    ),
    sort: Sort | None = Query(
        default=None,
        title="Sort criteria",
        description="The sort criteria to apply to the query. It is\n                specified in the query parameters as a comma-separated list of\n                field aliases with an optional prefix of a minus sign `-` for\n                descending order. For example, to sort results by the\n                `user.name` field in descending order, the sort criteria can be\n                specified as `user.name,-price` in the query parameters.\n                Additionally, the direction and nulls position can be piped\n                using a colon `:` character. For example, to sort results by\n                the `price` field in ascending order with nulls first, the sort\n                criteria can be specified as `price:asc:nf` in the query\n                parameters.",
        examples=["user.name,-price", "price:asc:nf"],
    ),
    limit: int | None = Query(
        default=None,
        gt=0,
        title="Limit",
        description="The maximum number of results to return. It is\n                specified in the query parameters as an integer value. Defaults\n                to the service configuration default page size if not\n                specified.",
        examples=[10, 20, 50],
    ),
    offset: int | None = Query(
        default=None,
        ge=0,
        title="Offset",
        description="The number of results to skip before returning the\n                results. It is specified in the query parameters as an integer\n                value. Defaults to `0` if not specified.",
    ),
    page: int | None = Query(
        default=None,
        gt=0,
        title="Page",
        description="The page number to return results for. It is\n                specified in the query parameters as an integer value. Defaults\n                to the first page if not specified.",
    ),
    page_size: int | None = Query(
        default=None,
        gt=0,
        title="Page size",
        description="The number of results to return per page. It is\n                specified in the query parameters as an integer value. The\n                maximum page size is defined by the service configuration.\n                Defaults to the service configuration default page size if not\n                specified.",
    ),
    dry_run: bool = Body(
        default=False,
        title="Dry-run",
        description="Whether to run the transaction in dry-run mode. If\n                set to `True`, the transaction will be rolled back after\n                execution.",
    ),
) -> Any | None

Delete a collection of resource instances.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSessionDep`	The async session dependency to use for the query.	required
`selection`	`KeyList[CRUDSpec]`	The key selection dependency to resolve the query.	`Selection()`
`return_result`	`bool`	Whether to return the deletion result.	Body(default=False, title='Return result', description='Whether to return the deleted result. If set to\n `False`, the transaction will be executed without returning\n the result.')
`include`	`IncEx \| None`	The fields to include in the query results.	`Body(default=None, title='Include fields', description='The fields to include in the query results.', examples=[['user.name', 'user.email'], {'user': ['name', 'email'], 'product': False}])`
`exclude`	`IncEx \| None`	The fields to exclude from the query results.	`Body(default=None, title='Exclude fields', description='The fields to exclude from the query results.', examples=[['user.password', 'product.price'], {'user': ['password'], 'product': True}])`
`filter`	`Annotated[Filter \| None, Depends(filter_dependency)]`	The filter criteria to apply to the query.	Body(default=None, title='Filter criteria', description='The filter criteria to apply to the query. It is\n specified in the request body as a dictionary of field aliases\n with their corresponding filter criteria. Additionally, it can\n be specified directly in the query parameters using a dot\n notation for field aliases and tilde `~` character for filter\n criteria. For example, to filter results where the `user.name`\n field starts with `Al`, the filter criteria can be specified as\n `.user.name=like~Al` in the query parameters.', examples=['.user.name=like~Al', '.price=gt~1000', '.user.name=like~Al&.price=gt~1000', {'user': {'name': {'operator': 'like', 'value': 'Al'}}, 'price': {'operator': 'gt', 'value': 1000}}])
`sort`	`Sort \| None`	The sort criteria to apply to the query.	Query(default=None, title='Sort criteria', description='The sort criteria to apply to the query. It is\n specified in the query parameters as a comma-separated list of\n field aliases with an optional prefix of a minus sign `-` for\n descending order. For example, to sort results by the\n `user.name` field in descending order, the sort criteria can be\n specified as `user.name,-price` in the query parameters.\n Additionally, the direction and nulls position can be piped\n using a colon `:` character. For example, to sort results by\n the `price` field in ascending order with nulls first, the sort\n criteria can be specified as `price:asc:nf` in the query\n parameters.', examples=['user.name,-price', 'price:asc:nf'])
`limit`	`int \| None`	The maximum number of results to return.	`Query(default=None, gt=0, title='Limit', description='The maximum number of results to return. It is\n specified in the query parameters as an integer value. Defaults\n to the service configuration default page size if not\n specified.', examples=[10, 20, 50])`
`offset`	`int \| None`	The number of results to skip before returning the results.	Query(default=None, ge=0, title='Offset', description='The number of results to skip before returning the\n results. It is specified in the query parameters as an integer\n value. Defaults to `0` if not specified.')
`page`	`int \| None`	The page number to return results for.	`Query(default=None, gt=0, title='Page', description='The page number to return results for. It is\n specified in the query parameters as an integer value. Defaults\n to the first page if not specified.')`
`page_size`	`int \| None`	The number of results to return per page.	`Query(default=None, gt=0, title='Page size', description='The number of results to return per page. It is\n specified in the query parameters as an integer value. The\n maximum page size is defined by the service configuration.\n Defaults to the service configuration default page size if not\n specified.')`
`dry_run`	`bool`	Whether to run the transaction in dry-run mode.	Body(default=False, title='Dry-run', description='Whether to run the transaction in dry-run mode. If\n set to `True`, the transaction will be rolled back after\n execution.')

Returns:

Type	Description
`Any \| None`	The collection of deleted resource instances or `None` if return
`Any \| None`	result is disabled, i.e., `return_result=False`.

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

@route.delete(
    path='',
    response_model=list[CRUDSpec.Read] | None,
    response_model_serialization=False,
)
async def delete_many(
    self,
    session: AsyncSessionDep,
    selection: KeyList[CRUDSpec] = Selection(),
    return_result: bool = Body(
        default=False,
        title='Return result',
        description="""Whether to return the deleted result. If set to
            `False`, the transaction will be executed without returning
            the result.""",
    ),
    include: IncEx | None = Body(
        default=None,
        title='Include fields',
        description="""The fields to include in the query results.""",
        examples=[
            ['user.name', 'user.email'],
            {'user': ['name', 'email'], 'product': False},
        ],
    ),
    exclude: IncEx | None = Body(
        default=None,
        title='Exclude fields',
        description="""The fields to exclude from the query results.""",
        examples=[
            ['user.password', 'product.price'],
            {'user': ['password'], 'product': True},
        ],
    ),
    filter: Annotated[Filter | None, Depends(filter_dependency)] = Body(
        default=None,
        title='Filter criteria',
        description="""The filter criteria to apply to the query. It is
            specified in the request body as a dictionary of field aliases
            with their corresponding filter criteria. Additionally, it can
            be specified directly in the query parameters using a dot
            notation for field aliases and tilde `~` character for filter
            criteria. For example, to filter results where the `user.name`
            field starts with `Al`, the filter criteria can be specified as
            `.user.name=like~Al*` in the query parameters.""",
        examples=[
            '.user.name=like~Al*',
            '.price=gt~1000',
            '.user.name=like~Al*&.price=gt~1000',
            {
                'user': {'name': {'operator': 'like', 'value': 'Al*'}},
                'price': {'operator': 'gt', 'value': 1000},
            },
        ],
    ),
    sort: Sort | None = Query(
        default=None,
        title='Sort criteria',
        description="""The sort criteria to apply to the query. It is
            specified in the query parameters as a comma-separated list of
            field aliases with an optional prefix of a minus sign `-` for
            descending order. For example, to sort results by the
            `user.name` field in descending order, the sort criteria can be
            specified as `user.name,-price` in the query parameters.
            Additionally, the direction and nulls position can be piped
            using a colon `:` character. For example, to sort results by
            the `price` field in ascending order with nulls first, the sort
            criteria can be specified as `price:asc:nf` in the query
            parameters.""",
        examples=['user.name,-price', 'price:asc:nf'],
    ),
    limit: int | None = Query(
        default=None,
        gt=0,
        title='Limit',
        description="""The maximum number of results to return. It is
            specified in the query parameters as an integer value. Defaults
            to the service configuration default page size if not
            specified.""",
        examples=[10, 20, 50],
    ),
    offset: int | None = Query(
        default=None,
        ge=0,
        title='Offset',
        description="""The number of results to skip before returning the
            results. It is specified in the query parameters as an integer
            value. Defaults to `0` if not specified.""",
    ),
    page: int | None = Query(
        default=None,
        gt=0,
        title='Page',
        description="""The page number to return results for. It is
            specified in the query parameters as an integer value. Defaults
            to the first page if not specified.""",
    ),
    page_size: int | None = Query(
        default=None,
        gt=0,
        title='Page size',
        description="""The number of results to return per page. It is
            specified in the query parameters as an integer value. The
            maximum page size is defined by the service configuration.
            Defaults to the service configuration default page size if not
            specified.""",
    ),
    dry_run: bool = Body(
        default=False,
        title='Dry-run',
        description="""Whether to run the transaction in dry-run mode. If
            set to `True`, the transaction will be rolled back after
            execution.""",
    ),
) -> Any | None:
    """Delete a collection of resource instances.

    Args:
        session: The async session dependency to use for the query.
        selection: The key selection dependency to resolve the query.
        return_result: Whether to return the deletion result.
        include: The fields to include in the query results.
        exclude: The fields to exclude from the query results.
        filter: The filter criteria to apply to the query.
        sort: The sort criteria to apply to the query.
        limit: The maximum number of results to return.
        offset: The number of results to skip before returning the results.
        page: The page number to return results for.
        page_size: The number of results to return per page.
        dry_run: Whether to run the transaction in dry-run mode.

    Returns:
        The collection of deleted resource instances or ``None`` if return
        result is disabled, i.e., ``return_result=False``.
    """
    # Resolve parameters
    limit = (
        limit
        or page_size
        or self.service_config.get(
            'page_size', self.resource_config.api_max_selection
        )
    )
    limit = min(limit, self.resource_config.api_max_selection)
    offset = offset or 0
    page = page or 1
    page_size = page_size or self.service_config.page_size

    # Build query
    query = selection.build_query(
        filter_criteria=filter,
        sort_criteria=sort,
    )
    query = query.offset((page - 1) * page_size + offset)
    query = query.limit(limit)

    # Execute query and return results
    buffer = await session.execute(query)
    result = buffer.unique().scalars().all()

    # Delete instances
    for instance in result:
        await session.delete(instance)

    # Commit or rollback the transaction
    if dry_run:
        await session.rollback()
    else:
        if return_result:
            session.sync_session.expire_on_commit = False
        await session.commit()

    # Handle return result
    if not return_result:
        return None

    # Handle result
    type_adapter = self.resource_schemas['read'].model_adapter
    with recursion_manager(mode='omit'):
        values = type_adapter.validate_python(result, from_attributes=True)
        return type_adapter.dump_python(
            values,
            mode='json',
            by_alias=True,
            include=include,
            exclude=exclude,
        )

bind_service

bind_service(
    service: BaseService,
    facade: ServiceFacade | ServiceWithSpecFacade,
) -> None

Bind the service to a service facade.

It binds the service to a service facade, allowing the service to be used within the context of its facade owner.

Parameters:

Name	Type	Description	Default
`service`	`BaseService`	The service instance to bind to the service facade.	required
`facade`	`ServiceFacade \| ServiceWithSpecFacade`	The service facade to bind the service to.	required

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

def bind_service(
    service: BaseService, facade: ServiceFacade | ServiceWithSpecFacade
) -> None:
    """Bind the service to a service facade.

    It binds the service to a service facade, allowing the service to be used
    within the context of its facade owner.

    Args:
        service: The service instance to bind to the service facade.
        facade: The service facade to bind the service to.
    """
    # Validate owner
    owner = service.service_owner
    if owner is facade:
        return
    if owner is not None:
        raise ValueError(
            f"Service instance {service!r} is already bound to a service "
            f"facade {owner.__class__.__qualname__!r}."
        )

    # Validate facade with specification
    spec = service.__config_spec__
    if spec is not None:
        if not isinstance(facade, ServiceWithSpecFacade):
            raise PlateformeError(
                f"Service facade {facade.__class__.__qualname__!r} must "
                f"implement the service with specification facade when a "
                f"service specification is defined.",
                code='services-invalid-config',
            )
        if service.service_config.auto_apply_spec \
                and spec not in getattr(facade, '__config_specs__', ()):
            facade._add_specs(spec)

    # Bind service to the facade
    object.__setattr__(service, 'service_owner', facade)

    # Call post bind hook
    service.__post_bind__(facade)

copy_service

copy_service(service: BaseService) -> BaseService

Copy and unbind the service from a service facade.

It copies the service instance and unbinds it from the service facade, removing the service from the context of its current facade owner.

Parameters:

Name	Type	Description	Default
`service`	`BaseService`	The service instance to unbind from the service facade.	required

Returns:

Type	Description
`BaseService`	The copied service instance.

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

def copy_service(service: BaseService) -> BaseService:
    """Copy and unbind the service from a service facade.

    It copies the service instance and unbinds it from the service facade,
    removing the service from the context of its current facade owner.

    Args:
        service: The service instance to unbind from the service facade.

    Returns:
        The copied service instance.
    """
    service = deepcopy(service)

    # Unbind if service is bound to a facade
    if service.service_owner is not None:
        for name, method in service.service_methods.items():
            setattr(service, name, method)
        object.__setattr__(service, 'service_owner', None)
    return service

load_service

load_service(
    service: BaseService,
) -> dict[str, Callable[..., Any]] | None

Load the service implementation.

It loads the service implementation by validating the service owner and checking whether the service owner implements the service specification protocol. If the service owner is not concrete or does not implement the service specification protocol, it raises an error.

Finally, it wraps the service methods with the appropriate parameter and return annotations based on the provided specification and resource.

Parameters:

Name	Type	Description	Default
`service`	`BaseService`	The service instance to load.	required

Returns:

Type	Description
`dict[str, Callable[..., Any]] \| None`	A dictionary of public methods of the service.

Raises:

Type	Description
`PlateformeError`	If the service owner is not concrete or does not implement the service specification protocol.

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

def load_service(service: BaseService) -> dict[str, Callable[..., Any]] | None:
    """Load the service implementation.

    It loads the service implementation by validating the service owner and
    checking whether the service owner implements the service specification
    protocol. If the service owner is not concrete or does not implement the
    service specification protocol, it raises an error.

    Finally, it wraps the service methods with the appropriate parameter and
    return annotations based on the provided specification and resource.

    Args:
        service: The service instance to load.

    Returns:
        A dictionary of public methods of the service.

    Raises:
        PlateformeError: If the service owner is not concrete or does not
            implement the service specification protocol.
    """
    # Validate owner
    owner = service.service_owner
    if owner is None:
        return None

    # Validate specification and implementation
    spec = service.__config_spec__
    if spec and not isimplclass_lenient(
        owner,
        spec,
        predicate=inspect.isclass,
        resolver=make_getattr_resolver(
            fallback_attr='resource_schemas',
            fallback_condition=inspect.isclass,
            fallback_transformer=lambda s: to_name_case(s, ('all', None)),
        ),
    ):
        raise PlateformeError(
            f"Service owner {owner.__class__.__qualname__!r} must be concrete "
            f"and implement the service specification {spec.__qualname__!r} "
            f"protocol.",
            code='services-invalid-config',
        )

    # Wrap service methods
    if service.service_methods is not None:
        for name, method in service.service_methods.items():
            wrapped_method = _wrap_service_method(service, method)
            setattr(service, name, wrapped_method)

    # Call post load hook
    service.__post_load__(owner)

    return _collect_service_methods(service)

unbind_service

unbind_service(service: BaseService) -> None

Unbind the service from a service facade.

It unbinds the service from a service facade, removing the service from the context of its current facade owner.

Parameters:

Name	Type	Description	Default
`service`	`BaseService`	The service instance to unbind from the service facade.	required

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

def unbind_service(service: BaseService) -> None:
    """Unbind the service from a service facade.

    It unbinds the service from a service facade, removing the service from the
    context of its current facade owner.

    Args:
        service: The service instance to unbind from the service facade.
    """
    # Check if service is bound to a facade
    if service.service_owner is None:
        raise ValueError(
            f"Service instance {service!r} is not bound to a service facade."
        )

    # Unbind service from the facade
    for name, method in service.service_methods.items():
        setattr(service, name, method)
    object.__setattr__(service, 'service_owner', None)

Services

plateforme.core.services

Service module-attribute

ServiceType module-attribute

ServiceFacade

ServiceWithSpecFacade

ServiceConfigDict

name instance-attribute

include instance-attribute

exclude instance-attribute

include_method instance-attribute

exclude_method instance-attribute

include_mode instance-attribute

exclude_mode instance-attribute

limit instance-attribute

page_size instance-attribute

auto_apply_spec instance-attribute

ServiceConfig

type_ class-attribute instance-attribute

name class-attribute instance-attribute

include class-attribute instance-attribute

exclude class-attribute instance-attribute

include_method class-attribute instance-attribute

exclude_method class-attribute instance-attribute

include_mode class-attribute instance-attribute

exclude_mode class-attribute instance-attribute

limit class-attribute instance-attribute

page_size class-attribute instance-attribute

auto_apply_spec class-attribute instance-attribute

create classmethod

from_meta classmethod

validate

context staticmethod

clear

copy

merge

check

entries

keys

values

items

get

pop

getdefault

setdefault

update

post_init

ServiceMeta

collect_config_wrappers

collect_spec

BaseService

BaseServiceWithSpec

CRUDService

ResourceFieldInfo module-attribute

Package

impl property

ResourceConfig

stub class-attribute instance-attribute

alias class-attribute instance-attribute

slug class-attribute instance-attribute

title class-attribute instance-attribute

str_to_lower class-attribute instance-attribute

str_to_upper class-attribute instance-attribute

str_strip_whitespace class-attribute instance-attribute

str_min_length class-attribute instance-attribute

str_max_length class-attribute instance-attribute

frozen class-attribute instance-attribute

populate_by_name class-attribute instance-attribute

use_enum_values class-attribute instance-attribute

validate_assignment class-attribute instance-attribute

arbitrary_types_allowed class-attribute instance-attribute

from_attributes class-attribute instance-attribute

loc_by_alias class-attribute instance-attribute

alias_generator class-attribute instance-attribute

ignored_types class-attribute instance-attribute

allow_inf_nan class-attribute instance-attribute

json_schema_extra class-attribute instance-attribute

json_encoders class-attribute instance-attribute

strict class-attribute instance-attribute

revalidate_instances class-attribute instance-attribute

Service `module-attribute`

ServiceType `module-attribute`

name `instance-attribute`

include `instance-attribute`

exclude `instance-attribute`

include_method `instance-attribute`

exclude_method `instance-attribute`

include_mode `instance-attribute`

exclude_mode `instance-attribute`

limit `instance-attribute`

page_size `instance-attribute`

auto_apply_spec `instance-attribute`

type_ `class-attribute` `instance-attribute`

name `class-attribute` `instance-attribute`

include `class-attribute` `instance-attribute`

exclude `class-attribute` `instance-attribute`

include_method `class-attribute` `instance-attribute`

exclude_method `class-attribute` `instance-attribute`

include_mode `class-attribute` `instance-attribute`

exclude_mode `class-attribute` `instance-attribute`

limit `class-attribute` `instance-attribute`

page_size `class-attribute` `instance-attribute`

auto_apply_spec `class-attribute` `instance-attribute`

create `classmethod`

from_meta `classmethod`

context `staticmethod`

ResourceFieldInfo `module-attribute`

impl `property`

stub `class-attribute` `instance-attribute`

alias `class-attribute` `instance-attribute`

slug `class-attribute` `instance-attribute`

title `class-attribute` `instance-attribute`

str_to_lower `class-attribute` `instance-attribute`

str_to_upper `class-attribute` `instance-attribute`

str_strip_whitespace `class-attribute` `instance-attribute`

str_min_length `class-attribute` `instance-attribute`

str_max_length `class-attribute` `instance-attribute`

frozen `class-attribute` `instance-attribute`

populate_by_name `class-attribute` `instance-attribute`

use_enum_values `class-attribute` `instance-attribute`

validate_assignment `class-attribute` `instance-attribute`

arbitrary_types_allowed `class-attribute` `instance-attribute`

from_attributes `class-attribute` `instance-attribute`

loc_by_alias `class-attribute` `instance-attribute`

alias_generator `class-attribute` `instance-attribute`

ignored_types `class-attribute` `instance-attribute`

allow_inf_nan `class-attribute` `instance-attribute`

json_schema_extra `class-attribute` `instance-attribute`

json_encoders `class-attribute` `instance-attribute`

strict `class-attribute` `instance-attribute`

revalidate_instances `class-attribute` `instance-attribute`

ser_json_timedelta `class-attribute` `instance-attribute`

ser_json_bytes `class-attribute` `instance-attribute`

ser_json_inf_nan `class-attribute` `instance-attribute`

validate_default `class-attribute` `instance-attribute`

validate_return `class-attribute` `instance-attribute`

protected_namespaces `class-attribute` `instance-attribute`

hide_input_in_errors `class-attribute` `instance-attribute`

plugin_settings `class-attribute` `instance-attribute`

schema_generator `class-attribute` `instance-attribute`

json_schema_serialization_defaults_required `class-attribute` `instance-attribute`

json_schema_mode_override `class-attribute` `instance-attribute`

coerce_numbers_to_str `class-attribute` `instance-attribute`

regex_engine `class-attribute` `instance-attribute`

validation_error_cause `class-attribute` `instance-attribute`

use_attribute_docstrings `class-attribute` `instance-attribute`

cache_strings `class-attribute` `instance-attribute`

type_ `class-attribute` `instance-attribute`

extra `class-attribute` `instance-attribute`

defer_build `class-attribute` `instance-attribute`

tags `class-attribute` `instance-attribute`

api_max_depth `class-attribute` `instance-attribute`

api_max_selection `class-attribute` `instance-attribute`

id_strategy `class-attribute` `instance-attribute`

id_type `class-attribute` `instance-attribute`

mapper_args `class-attribute` `instance-attribute`

use_single_table_inheritance `class-attribute` `instance-attribute`

indexes `class-attribute` `instance-attribute`

services `class-attribute` `instance-attribute`

specs `class-attribute` `instance-attribute`