Resources

plateforme.core.resources

This module provides utilities for managing resources within the Plateforme framework's. It integrates with other core components of the framework, such as model validation and serialization, database interaction, and API routing.

The BaseResource class is the foundational base for all resources within the framework. It provides a common interface for defining resources and managing their lifecycle. Resources are used to represent entities within the system, such as users, products, or orders. They are used to define the structure of the data, the business logic, and the API endpoints.

Note

See also the Field function for more information on modeling features.

Resource module-attribute

Resource = TypeVar('Resource', bound='BaseResource')

A type variable for a resource class.

ResourceType module-attribute

ResourceType = Type['BaseResource']

A type alias for a resource class.

ResourceFieldInfo module-attribute

ResourceFieldInfo = FieldInfo['BaseResource']

A type alias for a resource field information.

ResourceIndex

Bases: TypedDict

A configuration dictionary for defining indexes on resource fields.

name instance-attribute

name: str

The name of the index. It can be used to customize the generated index name using the index registy naming convention. Defaults to the snake-cased concatenation of the resource alias and specified field aliases.

aliases instance-attribute

aliases: Required[set[str]]

The required resource field aliases to include in the index. The field aliases must be defined in the underlying resource model.

unique instance-attribute

unique: bool

Whether the index is unique. Defaults to True.

ResourceConfigDict

Bases: BaseModelConfigDict

A resource class configuration dictionary.

alias instance-attribute

alias: str

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 instance-attribute

slug: str

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 instance-attribute

title: str

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 instance-attribute

str_to_lower: bool

Whether to convert strings to lowercase. Defaults to False.

str_to_upper instance-attribute

str_to_upper: bool

Whether to convert strings to uppercase. Defaults to False.

str_strip_whitespace instance-attribute

str_strip_whitespace: bool

Whether to strip whitespace from strings. Defaults to False.

str_min_length instance-attribute

str_min_length: int

The minimum length for strings. Defaults to None.

str_max_length instance-attribute

str_max_length: int | None

The maximum length for strings. Defaults to None.

frozen instance-attribute

frozen: bool

Whether to freeze the configuration. Defaults to False.

populate_by_name instance-attribute

populate_by_name: bool

Whether to populate fields by name. Defaults to False.

use_enum_values instance-attribute

use_enum_values: bool

Whether to use enum values. Defaults to False.

validate_assignment instance-attribute

validate_assignment: bool

Whether to validate assignments. Defaults to False.

arbitrary_types_allowed instance-attribute

arbitrary_types_allowed: bool

Whether to allow arbitrary types. Defaults to False.

from_attributes instance-attribute

from_attributes: bool

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

loc_by_alias instance-attribute

loc_by_alias: bool

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

alias_generator instance-attribute

alias_generator: (
    Callable[[str], str] | AliasGenerator | None
)

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

ignored_types instance-attribute

ignored_types: tuple[type, ...]

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

allow_inf_nan instance-attribute

allow_inf_nan: bool

Whether to allow infinity and NaN. Defaults to True.

json_schema_extra instance-attribute

json_schema_extra: JsonSchemaExtraCallable | None

Dictionary of extra JSON schema properties. Defaults to None.

json_encoders instance-attribute

json_encoders: dict[type[object], JsonEncoder] | None

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

strict instance-attribute

strict: bool

Whether to make the configuration strict. Defaults to False.

revalidate_instances instance-attribute

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

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

ser_json_timedelta instance-attribute

ser_json_timedelta: Literal['iso8601', 'float']

The format of JSON serialized timedeltas. Defaults to iso8601.

ser_json_bytes instance-attribute

ser_json_bytes: Literal['utf8', 'base64']

The encoding of JSON serialized bytes. Defaults to utf8.

ser_json_inf_nan instance-attribute

ser_json_inf_nan: Literal['null', 'constants']

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

validate_default instance-attribute

validate_default: bool

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

validate_return instance-attribute

validate_return: bool

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

protected_namespaces instance-attribute

protected_namespaces: tuple[str, ...]

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 instance-attribute

hide_input_in_errors: bool

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

plugin_settings instance-attribute

plugin_settings: dict[str, object] | None

A dictionary of settings for plugins. Defaults to None.

schema_generator instance-attribute

schema_generator: type[Any] | None

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

json_schema_serialization_defaults_required instance-attribute

json_schema_serialization_defaults_required: bool

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

json_schema_mode_override instance-attribute

json_schema_mode_override: (
    Literal["validation", "serialization"] | 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 instance-attribute

coerce_numbers_to_str: bool

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

regex_engine instance-attribute

regex_engine: Literal['rust-regex', 'python-re']

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

validation_error_cause instance-attribute

validation_error_cause: bool

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 instance-attribute

use_attribute_docstrings: bool

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

cache_strings instance-attribute

cache_strings: bool | Literal['all', 'keys', 'none']

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.

tags instance-attribute

tags: list[str | Enum] | 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 instance-attribute

api_max_depth: int

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 instance-attribute

api_max_selection: int

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 instance-attribute

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

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 instance-attribute

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

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 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 instance-attribute

use_single_table_inheritance: bool

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 instance-attribute

indexes: Sequence[ResourceIndex | set[str]]

A sequence or list of resource indexes configurations that define indexing for resource model fields. An index is defined either as a set of field aliases or 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 sequence.

services instance-attribute

services: Sequence[BaseService | EllipsisType | ServiceType]

A sequence 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 sequence.

specs instance-attribute

specs: Sequence[SpecType]

A sequence 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 sequence.

deprecated instance-attribute

deprecated: bool | None

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

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)

ResourceDict

ResourceDict(__model: BaseModel, *args: Any, **kwargs: Any)

Bases: dict[str, Any]

A custom dictionary class for the resource instance.

Initialize a new resource dictionary with the provided model.

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

def __init__(self, __model: BaseModel, *args: Any, **kwargs: Any) -> None:
    """Initialize a new resource dictionary with the provided model."""
    super().__init__(*args, **kwargs)
    self.model = __model

clear

clear() -> None

Clear the inner dictionary and reset all values.

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

def clear(self) -> None:
    """Clear the inner dictionary and reset all values."""
    self.model.__dict__.clear()
    super().clear()

copy

copy() -> Self

Return a shallow copy of the inner dictionary.

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

def copy(self) -> Self:
    """Return a shallow copy of the inner dictionary."""
    return self.__class__(self.model, super().copy())

keys

keys() -> KeysView[str]

Return the inner dictionary keys.

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

def keys(self) -> KeysView[str]:  # type: ignore[override]
    """Return the inner dictionary keys."""
    return KeysView(self.__dict__)

values

values() -> ValuesView[Any]

Return the inner dictionary values.

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

def values(self) -> ValuesView[Any]:  # type: ignore[override]
    """Return the inner dictionary values."""
    return ValuesView(self.__dict__)

items

items() -> ItemsView[str, Any]

Return the inner dictionary items.

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

def items(self) -> ItemsView[str, Any]:  # type: ignore[override]
    """Return the inner dictionary items."""
    return ItemsView(self.__dict__)

get

get(key: str) -> Any

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

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

Get the value for the specified key.

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

def get(self, key: str, default: Any = Undefined) -> Any:
    """Get the value for the specified key."""
    if default is Undefined:
        return self.__dict__.get(key)
    return self.__dict__.get(key, default)

pop

pop(key: str) -> Any

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

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

Pop the specified key from the inner dictionary.

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

def pop(self, key: str, default: Any = Undefined) -> Any:
    """Pop the specified key from the inner dictionary."""
    # Check if key is in the model instance dictionary
    if key in self.model.__dict__:
        if default is Undefined:
            return self.model.__dict__.pop(key)
        return self.model.__dict__.pop(key, default)
    # Fallback to the resource dictionary
    if default is Undefined:
        return super().pop(key)
    return super().pop(key, default)

setdefault

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

Set the default value for the specified key if not set.

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

def setdefault(  # type: ignore[override]
    self, key: str, default: Any
) -> Any:
    """Set the default value for the specified key if not set."""
    if key not in self:
        self[key] = default
        return default
    return self[key]

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/resources.py

def update(  # type: ignore[override]
    self,
    *args: tuple[str, Any] | Mapping[str, Any],
    **kwargs: Any,
) -> None:
    """Update the config dictionary with new data."""
    # Update args data
    for arg in args:
        if isinstance(arg, tuple):
            self[arg[0]] = arg[1]
        else:
            for key, value in arg.items():
                self[key] = value
    # Update kwargs data
    for key, value in kwargs.items():
        self[key] = value

ResourceNode

ResourceNode(
    resource: type[Resource],
    accessor: ResourceFieldInfo | None = None,
)

Bases: Generic[Resource]

A resource node within the resources graph.

The resource node represents a segment within the resources graph. It is associated with a specific resource and a linked field accessor. The node is used to build resource paths and navigate the resources graph.

Attributes:

Name	Type	Description
resource		The resource associated with the node.
accessor		The linked field accessor associated with the node.

Note

The identifiers generated from the unique indexes of the resource are used to identify a specific resource instance within this node. Thus, it can be used within a resource path to navigate the resources graph.

Initialize a new resource node.

It initializes a new resource node with the provided resource and linked field accessor. It serves as a building block for constructing resource paths and navigating the resources graph.

Parameters:

Name	Type	Description	Default
resource	type[Resource]	The resource associated with the node.	required
accessor	ResourceFieldInfo | None	The linked field accessor associated with the node.	None

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

def __init__(
    self,
    resource: type[Resource],
    accessor: ResourceFieldInfo | None = None,
) -> None:
    """Initialize a new resource node.

    It initializes a new resource node with the provided resource and
    linked field accessor. It serves as a building block for constructing
    resource paths and navigating the resources graph.

    Args:
        resource: The resource associated with the node.
        accessor: The linked field accessor associated with the node.
    """
    # Initialize with default identifier
    self.resource = resource
    self.accessor = accessor

backref property

backref: ResourceFieldInfo | None

The accessor association backref of the resource node.

config property

config: ResourceConfig

The config of the node resource.

segment property

segment: str | None

The accessor segment alias of the resource node.

ResourcePathInfo dataclass

ResourcePathInfo(
    *, path: str, parameters: dict[str, Parameter]
)

A resource path information class.

It provides information about the resource path, including the path string, and the parameters required to resolve the resource path, and the resolver function used to fetch and validate a single resource instance or a collection of resource instances.

path instance-attribute

path: str

The path string for the resource path.

parameters instance-attribute

parameters: dict[str, Parameter]

A dictionary of parameters required to resolve the resource path.

ResourcePath

ResourcePath(root: ResourceType, *path: str)

Bases: Iterable[ResourceNode['BaseResource']]

A resource path within the resources graph.

The resource path represents a sequence of linked fields within the resources graph. It is used to navigate the resources graph and build API paths to interact with the resources.

Attributes:

Name	Type	Description
nodes		A list of ResourceNode objects, each item representing a resource node within the resources graph.

Initialize a new resource path.

It initializes a new resource path with the provided root resource and path segments. The path segments represent a sequence of linked fields within the resources graph, and together with the root resource they define a resource path.

Parameters:

Name	Type	Description	Default
root	ResourceType	The root resource of the API path serving as the starting point in the resources graph.	required
*path	str	Variable length argument list representing the path segments of the resource path. Each segment corresponds to a linked field alias within the resources graph.	()

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

def __init__(self, root: ResourceType, *path: str) -> None:
    """Initialize a new resource path.

    It initializes a new resource path with the provided root resource and
    path segments. The path segments represent a sequence of linked fields
    within the resources graph, and together with the root resource they
    define a resource path.

    Args:
        root: The root resource of the API path serving as the starting
            point in the resources graph.
        *path: Variable length argument list representing the path segments
            of the resource path. Each segment corresponds to a linked
            field alias within the resources graph.
    """
    # Initialize with the root resource
    self.nodes = [ResourceNode(root)]

    # Navigate to the provided path
    self.goto(*path, in_place=True)

is_root property

is_root: bool

Whether the resource path is at the root level.

root property

root: ResourceType

The root resource of the resource path.

target property

target: ResourceType

The target resource of the resource path.

goto

goto(*path: str, in_place: bool = True) -> Self

Navigate to the provided path segments.

It navigates to the provided path segments within the resources graph. Each segment is checked to match a linked field alias within the resource hierarchy. If any segment fails to match a field alias in the resource hierarchy, an error is raised.

Parameters:

Name	Type	Description	Default
*path	str	Variable length argument list representing the path segments to navigate within the resources graph. Each segment corresponds to a linked field alias within the resources graph.	()
in_place	bool	A flag indicating whether to navigate in place or create a new resource path. Defaults to True. If set to False, a new resource path is created and returned, not modifying the current resource path.	True

Raises:

Type	Description
ValueError	If any of the provided segments do not correspond to a linked field alias in the resource hierarchy.

Note

A provided path arg can be a dot-separated string representing multiple segments. For example, the path arg foo.bar is split into two segments foo and bar.

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

def goto(
    self,
    *path: str,
    in_place: bool = True,
) -> Self:
    """Navigate to the provided path segments.

    It navigates to the provided path segments within the resources graph.
    Each segment is checked to match a linked field alias within the
    resource hierarchy. If any segment fails to match a field alias in the
    resource hierarchy, an error is raised.

    Args:
        *path: Variable length argument list representing the path segments
            to navigate within the resources graph. Each segment
            corresponds to a linked field alias within the resources graph.
        in_place: A flag indicating whether to navigate in place or create
            a new resource path. Defaults to ``True``. If set to ``False``,
            a new resource path is created and returned, not modifying the
            current resource path.

    Raises:
        ValueError: If any of the provided segments do not correspond to a
            linked field alias in the resource hierarchy.

    Note:
        A provided path arg can be a dot-separated string representing
        multiple segments. For example, the path arg `foo.bar` is split
        into two segments `foo` and `bar`.
    """
    # Collect all segments from the path arguments
    segments = [
        segment
        for arg in path
        for segment in arg.split('.') if segment
    ]

    # Initialize the resource path object
    obj = self
    if not in_place:
        obj = deepcopy(self)

    # Retrieve the linked field for each provided segment alias. It looks
    # for field aliases within the resource hierarchy and checks whether
    # linked fields are valid.
    resource = obj.target
    for segment in segments:
        # Look for a linked field matching the segment alias
        accessor: ResourceFieldInfo | None = None
        for field in resource.resource_fields.values():
            # Skip non-matching field aliases
            if segment != field.alias:
                continue
            # Check if field is a linked field
            if not field.linked:
                raise ValueError(
                    f"Segment {segment!r} in the resource path does not "
                    f"correspond to a linked field alias. Got: "
                    f"{field!r}."
                )
            # Check if field target is a valid resource type
            if not field.target or isinstance(field.target, str):
                raise ValueError(
                    f"Segment {segment!r} in the resource path does not "
                    f"have a valid target resource type. Got: "
                    f"{field.target!r}."
                )
            # Update resource and accessor
            resource = field.target
            accessor = field
            break

        # Raise an error if no matching field is found
        if accessor is None:
            raise ValueError(
                f"Segment {segment!r} in the resource path does not match "
                f"any field alias within the resource "
                f"{resource.__class__.__qualname__!r}."
            )

        # Initialize the resource node and check for duplicates
        node = ResourceNode(resource, accessor)
        if any(node == other_node for other_node in self.nodes):
            raise ValueError(
                f"Segment {segment!r} in the resource path is a duplicate "
                f"and has already been used in the resource "
                f"{resource.__class__.__qualname__!r}."
            )

        # Append the resource node to the resource path
        obj.nodes.append(node)

    return obj

walk

walk(
    _guard: frozenset[str] = frozenset(),
    /,
    max_depth: int = 2,
    skip_root: bool = False,
) -> Generator[Self, None, None]

Walk the resource graph and yield all possible sub-paths.

It recursively walks the resource graph and yields all possible sub-paths up to the specified maximum depth. The maximum depth is used to limit the recursion depth and avoid infinite loops when walking the resource graph. Already walked paths are skipped to avoid duplicates.

Parameters:

Name	Type	Description	Default
_guard	frozenset[str]	A set of already walked resource association alias. It should not be used directly and is only meant to be used internally for recursion.	frozenset()
max_depth	int	The maximum depth to walk the resource graph. It limits the recursion depth and avoids infinite loops. Defaults to 1.	2
skip_root	bool	A flag indicating whether to skip the root resource path. Defaults to False.	False

Yields:

Type	Description
Self	All possible sub-paths up to the specified maximum depth.

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

def walk(
    self,
    _guard: frozenset[str] = frozenset(),
    /,
    max_depth: int = 2,
    skip_root: bool = False,
) -> Generator[Self, None, None]:
    """Walk the resource graph and yield all possible sub-paths.

    It recursively walks the resource graph and yields all possible
    sub-paths up to the specified maximum depth. The maximum depth is used
    to limit the recursion depth and avoid infinite loops when walking the
    resource graph. Already walked paths are skipped to avoid duplicates.

    Args:
        _guard: A set of already walked resource association alias. It
            should not be used directly and is only meant to be used
            internally for recursion.
        max_depth: The maximum depth to walk the resource graph. It limits
            the recursion depth and avoids infinite loops.
            Defaults to ``1``.
        skip_root: A flag indicating whether to skip the root resource
            path. Defaults to ``False``.

    Yields:
        All possible sub-paths up to the specified maximum depth.
    """
    # Yield the root resource path if not skipped
    if len(_guard) == 0:
        if not skip_root:
            yield self

    # Stop the recursion if the maximum depth is reached
    if len(_guard) >= max_depth - 1:
        return

    # Otherwise, walk the resource graph
    for field in self.target.resource_fields.values():
        # Skip already walked paths
        if field.recursive_guard is None \
                or field.recursive_guard in _guard:
            continue
        # Skip non-linked fields
        if not field.linked:
            continue
        # Skip non-backref fields
        if field.rel_backref is None:
            continue
        # Skip lazy loaded fields
        if field.is_eagerly() is not True:
            continue
        # Get the resource path object
        obj = self.goto(
            field.alias,
            in_place=False,
        )
        # Yield the resource path and walk its sub-paths
        yield obj
        yield from obj.walk(
            _guard | {field.recursive_guard},
            max_depth=max_depth,
        )

ResourceManager

ResourceManager(resource: type[Resource])

Bases: Manager[Resource]

A resource class manager.

It provides a common interface to access and manage the service methods associated with a resource class.

Initialize the resource manager.

It initializes the resource manager with the provided resource class, and collects all resource public instance methods decorated with the route decorator.

Parameters:

Name	Type	Description	Default
resource	type[Resource]	The resource class to manage.	required

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

def __init__(self, resource: type[Resource]) -> None:
    """Initialize the resource manager.

    It initializes the resource manager with the provided resource class,
    and collects all resource public instance methods decorated with the
    route decorator.

    Args:
        resource: The resource class to manage.
    """
    # Set the resource class
    super().__init__(resource)

    # Schedule public resource methods gathering
    self.resource.__state__.schedule(
        self._add_resource_methods, when=Lifecycle.LOADING,
    )

exists async

exists(
    __session: AsyncSession | None = None, /, **kwargs: Any
) -> bool

Check if a resource instance exist based on the given filters.

Parameters:

Name	Type	Description	Default
__session	AsyncSession | None	The async session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None
**kwargs	Any	The filters to apply when querying the resource instances as keyword arguments with the field aliases as keys and the values to filter by as values.	{}

Returns:

Type	Description
bool	True if a resource instance exists that matches the provided
bool	filters, otherwise False.

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

async def exists(
    self,
    __session: AsyncSession | None = None,
    /,
    **kwargs: Any,
) -> bool:
    """Check if a resource instance exist based on the given filters.

    Args:
        __session: The async session to use for the operation. If not
            provided, the session in the current context is used.
            Defaults to ``None``.
        **kwargs: The filters to apply when querying the resource
            instances as keyword arguments with the field aliases as keys
            and the values to filter by as values.

    Returns:
        ``True`` if a resource instance exists that matches the provided
        filters, otherwise ``False``.
    """
    async def query(session: AsyncSession) -> bool:
        query = select(select(self.resource).where(**kwargs).exists())
        buffer = await session.execute(query)
        return buffer.scalar() or False

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

get async

get(
    __session: AsyncSession | None = None, /, **kwargs: Any
) -> Resource | None

Get a single resource instance based on the given filters.

Parameters:

Name	Type	Description	Default
__session	AsyncSession | None	The async session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None
**kwargs	Any	The filters to apply when querying the resource instances as keyword arguments with the field aliases as keys and the values to filter by as values.	{}

Returns:

Type	Description
Resource | None	A single resource instance that matches the provided filters, or
Resource | None	None if no resource instance is found.

Note

This method is equivalent to the get_one method, but it returns None if no resource instance is found for the provided filters instead of raising an exception.

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

async def get(
    self,
    __session: AsyncSession | None = None,
    /,
    **kwargs: Any,
) -> Resource | None:
    """Get a single resource instance based on the given filters.

    Args:
        __session: The async session to use for the operation. If not
            provided, the session in the current context is used.
            Defaults to ``None``.
        **kwargs: The filters to apply when querying the resource
            instances as keyword arguments with the field aliases as keys
            and the values to filter by as values.

    Returns:
        A single resource instance that matches the provided filters, or
        ``None`` if no resource instance is found.

    Note:
        This method is equivalent to the `get_one` method, but it returns
        ``None`` if no resource instance is found for the provided filters
        instead of raising an exception.
    """
    async def query(session: AsyncSession) -> Resource | None:
        query = select(self.resource).filter_by(**kwargs).limit(1)
        buffer = await session.execute(query)
        return buffer.scalar()

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

get_many async

get_many(
    __session: AsyncSession | None = None, /, **kwargs: Any
) -> Sequence[Resource]

Get a collection of resource instances based on the given filters.

Parameters:

Name	Type	Description	Default
__session	AsyncSession | None	The async session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None
**kwargs	Any	The filters to apply when querying the resource instances as keyword arguments with the field aliases as keys and the values to filter by as values.	{}

Returns:

Type	Description
Sequence[Resource]	A collection of resource instances that match the provided filters.

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

async def get_many(
    self,
    __session: AsyncSession | None = None,
    /,
    **kwargs: Any,
) -> Sequence[Resource]:
    """Get a collection of resource instances based on the given filters.

    Args:
        __session: The async session to use for the operation. If not
            provided, the session in the current context is used.
            Defaults to ``None``.
        **kwargs: The filters to apply when querying the resource
            instances as keyword arguments with the field aliases as keys
            and the values to filter by as values.

    Returns:
        A collection of resource instances that match the provided filters.
    """
    async def query(session: AsyncSession) -> Sequence[Resource]:
        query = select(self.resource).filter_by(**kwargs)
        buffer = await session.execute(query)
        return buffer.unique().scalars().all()

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

get_one async

get_one(
    __session: AsyncSession | None = None, /, **kwargs: Any
) -> Resource

Get exactly one resource instance based on the given filters.

It returns exactly one resource instance that matches the provided filters, or raises an exception if no resource instance is found.

Parameters:

Name	Type	Description	Default
__session	AsyncSession | None	The async session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None
**kwargs	Any	The filters to apply when querying the resource instances as keyword arguments with the field aliases as keys and the values to filter by as values.	{}

Raises:

Type	Description
ValueError	If no resource instance is found.

Returns:

Type	Description
Resource	A single resource instance that matches the provided filters.

Note

This method is equivalent to the get method, but it raises an exception if no resource instance is found for the provided filters instead of returning None.

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

async def get_one(
    self,
    __session: AsyncSession | None = None,
    /,
    **kwargs: Any,
) -> Resource:
    """Get exactly one resource instance based on the given filters.

    It returns exactly one resource instance that matches the provided
    filters, or raises an exception if no resource instance is found.

    Args:
        __session: The async session to use for the operation. If not
            provided, the session in the current context is used.
            Defaults to ``None``.
        **kwargs: The filters to apply when querying the resource
            instances as keyword arguments with the field aliases as keys
            and the values to filter by as values.

    Raises:
        ValueError: If no resource instance is found.

    Returns:
        A single resource instance that matches the provided filters.

    Note:
        This method is equivalent to the `get` method, but it raises an
        exception if no resource instance is found for the provided filters
        instead of returning ``None``.
    """
    async def query(session: AsyncSession) -> Resource:
        query = select(self.resource).filter_by(**kwargs).limit(2)
        buffer = await session.execute(query)
        result = buffer.unique().scalars().all()
        if len(result) != 1:
            raise ValueError(
                f"Expected exactly one resource instance matching the "
                f"provided filters. Got: {len(result)} instances with "
                f"filters: {kwargs}."
            )
        return result[0]

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

ResourceState

ResourceState(resource: type[Resource])

Bases: Generic[Resource]

A resource class state.

It provides utilities to manage the state of a resource class, including its lifecycle status and scheduled tasks associated with the resource class build.

Initialize the resource state.

It initializes the resource state with the provided resource class and sets the lifecycle status to RESOLVED.

Parameters:

Name	Type	Description	Default
resource	type[Resource]	The resource class owner of the state.	required

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

def __init__(self, resource: type[Resource]) -> None:
    """Initialize the resource state.

    It initializes the resource state with the provided resource class and
    sets the lifecycle status to ``RESOLVED``.

    Args:
        resource: The resource class owner of the state.
    """
    self.owner = resource

    # Initialize lifecycle status and scheduled tasks
    self.status = Lifecycle.RESOLVED
    self.tasks = {
        Lifecycle.INITIALIZING: [],
        Lifecycle.LOADING: [],
        Lifecycle.BUILDING: [],
        Lifecycle.FINALIZING: [],
        Lifecycle.READY: [],
    }

flush

flush(
    _check_guard: set[str] | None = None,
    _flush_guard: set[str] | None = None,
    *,
    until: SchedulableState | None = None,
) -> bool | None

Advance the resource through its lifecycle states.

This method handles the transition through lifecycle states while managing dependencies and preventing circular references. For each state transition, it ensures all dependencies have reached the required state, executes any scheduled tasks, and propagates the state change to dependent resources.

The flush operation follows these steps: 1. Check and handle any base resources (inheritance-based dependencies) 2. Verify all resource dependencies have reached the required state 3. Execute scheduled tasks for the next state 4. Update the resource state 5. Propagate state changes to dependent resources

Parameters:

Name	Type	Description	Default
_check_guard	set[str] | None	Internal set to prevent circular dependency checks. Used to track which resources have been processed during dependency validation. Defaults to an empty set.	None
_flush_guard	set[str] | None	Internal set to prevent recursive flush operations. Used to detect and break circular flush attempts between interdependent resources. Defaults to an empty set.	None
until	SchedulableState | None	The target lifecycle state to reach. The flush operation will advance the resource state up to this point but not beyond. If not specified, attempts to reach the READY state. Defaults to None.	None

Returns:

Type	Description
bool | None	False if dependencies prevent state advancement, None if no
bool | None	state change is needed (already at target state or detected
bool | None	recursion), or recursively continues flushing until target state is
bool | None	reached. Otherwise, returns True to indicate a successful state
bool | None	advancement.

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

def flush(
    self,
    _check_guard: set[str] | None = None,
    _flush_guard: set[str] | None = None,
    *,
    until: SchedulableState | None = None,
) -> bool | None:
    """Advance the resource through its lifecycle states.

    This method handles the transition through lifecycle states while
    managing dependencies and preventing circular references. For each
    state transition, it ensures all dependencies have reached the required
    state, executes any scheduled tasks, and propagates the state change to
    dependent resources.

    The flush operation follows these steps:
    1. Check and handle any base resources (inheritance-based dependencies)
    2. Verify all resource dependencies have reached the required state
    3. Execute scheduled tasks for the next state
    4. Update the resource state
    5. Propagate state changes to dependent resources

    Args:
        _check_guard: Internal set to prevent circular dependency checks.
            Used to track which resources have been processed during
            dependency validation. Defaults to an empty set.
        _flush_guard: Internal set to prevent recursive flush operations.
            Used to detect and break circular flush attempts between
            interdependent resources. Defaults to an empty set.
        until: The target lifecycle state to reach. The flush operation
            will advance the resource state up to this point but not
            beyond. If not specified, attempts to reach the ``READY``
            state. Defaults to ``None``.

    Returns:
        ``False`` if dependencies prevent state advancement, ``None`` if no
        state change is needed (already at target state or detected
        recursion), or recursively continues flushing until target state is
        reached. Otherwise, returns ``True`` to indicate a successful state
        advancement.
    """
    # Initialize recursion guards
    _check_guard = set() if _check_guard is None else _check_guard
    _flush_guard = set() if _flush_guard is None else _flush_guard

    name = get_object_name(self.owner, fullname=True)
    if name in _flush_guard:
        return None
    _flush_guard.add(name)

    # Validate flush lifecycle status
    until = until or Lifecycle.READY
    if self.status >= until:
        return None

    next_status = self.status.next()

    # Helper function to flush bases
    def flush_bases() -> bool:
        for base in self.owner.__bases__:
            if not is_resource(base) or is_abstract(base):
                continue
            if base.__state__.flush(
                _check_guard,
                _flush_guard,
                until=next_status,
            ) is False:
                return False
        return True

    # Helper function to check dependencies
    def check_dependencies() -> bool:
        if runtime.get_dependencies(
            self.owner,
            _check_guard,
            kind='resources',
            status=self.status.lower(),
            max_depth=None,
        ):
            return False
        return True

    # Check for pending dependencies
    if not flush_bases() or not check_dependencies():
        return False

    # Process and clear scheduled tasks
    for scheduled_task in self.tasks[next_status]:
        scheduled_task(self.owner)
    self.tasks[next_status].clear()

    self.status = next_status

    logger.debug(
        f"rsc:{self.owner.resource_config.alias} -> {self.status}"
    )

    # Recursively flush dependents
    for dependent in runtime.get_dependents(self.owner, kind='resources'):
        dependent.__state__.flush(
            _check_guard.copy(),
            _flush_guard,
            until=next_status,
        )

    # Recursively flush children
    for sub in get_subclasses(self.owner):
        assert is_resource(sub)
        sub.__state__.flush(
            _check_guard.copy(),
            _flush_guard,
            until=next_status,
        )

    return self.flush(until=until)

schedule

schedule(
    *tasks: Task,
    _guard: set[str] | None = None,
    when: SchedulableState | None = None,
    when_reached: Literal[
        "fallback", "run", "skip"
    ] = "run",
    when_future: Literal["await", "skip"] = "await",
    propagate: bool = False,
) -> None

Schedule tasks for execution.

It schedules tasks for immediate or future execution based on resource lifecycle states. Each task can target a specific state, with configurable behavior for handling already reached states using the when_reached option, and future states using the when_future option. Both strategies can be used concurrently to handle different scenarios. Tasks scheduled for the same lifecycle state execute sequentially in the order they were added.

Parameters:

Name	Type	Description	Default
*tasks	Task	The tasks to execute or schedule for the resource state. Multiple tasks scheduled for the same state are executed sequentially in the order they were added.	()
_guard	set[str] | None	Internal set to prevent recursive schedule propagation operations. Used to detect and break circular schedule attempts between interdependent resources. Defaults to an empty set.	None
when	SchedulableState | None	Default scheduling state for the provided tasks. If a task doesn't specify its own lifecycle state to schedule for, this value is used. When None, tasks are scheduled for the next resource lifecycle state. Defaults to None.	None
when_reached	Literal['fallback', 'run', 'skip']	The strategy to apply when a task is scheduled for a state that has already been reached. It can be set to one of the following values: - 'fallback': Fallback to the previous state and schedule the task for execution, - 'run': Run the task immediately, - 'skip': Skip the task and do not execute it. Defaults to 'run'.	'run'
when_future	Literal['await', 'skip']	The strategy to apply when a task is scheduled for a future state. It can be set to one of the following values: - 'await': Await the state transition and execute the task when the state is reached, - 'skip': Skip the task and do not execute it. Defaults to 'await'.	'await'
propagate	bool	Whether to propagate and apply the same scheduled tasks to all dependent and child resources. Defaults to False.	False

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

def schedule(
    self,
    *tasks: Task,
    _guard: set[str] | None = None,
    when: SchedulableState | None = None,
    when_reached: Literal['fallback', 'run', 'skip'] = 'run',
    when_future: Literal['await', 'skip'] = 'await',
    propagate: bool = False,
) -> None:
    """Schedule tasks for execution.

    It schedules tasks for immediate or future execution based on resource
    lifecycle states. Each task can target a specific state, with
    configurable behavior for handling already reached states using the
    `when_reached` option, and future states using the `when_future`
    option. Both strategies can be used concurrently to handle different
    scenarios. Tasks scheduled for the same lifecycle state execute
    sequentially in the order they were added.

    Args:
        *tasks: The tasks to execute or schedule for the resource state.
            Multiple tasks scheduled for the same state are executed
            sequentially in the order they were added.
        _guard: Internal set to prevent recursive schedule propagation
            operations. Used to detect and break circular schedule attempts
            between interdependent resources. Defaults to an empty set.
        when: Default scheduling state for the provided tasks. If a task
            doesn't specify its own lifecycle state to schedule for, this
            value is used. When ``None``, tasks are scheduled for the next
            resource lifecycle state. Defaults to ``None``.
        when_reached: The strategy to apply when a task is scheduled for a
            state that has already been reached. It can be set to one of
            the following values:
            - ``'fallback'``: Fallback to the previous state and schedule
                the task for execution,
            - ``'run'``: Run the task immediately,
            - ``'skip'``: Skip the task and do not execute it.
            Defaults to ``'run'``.
        when_future: The strategy to apply when a task is scheduled for a
            future state. It can be set to one of the following values:
            - ``'await'``: Await the state transition and execute the task
                when the state is reached,
            - ``'skip'``: Skip the task and do not execute it.
            Defaults to ``'await'``.
        propagate: Whether to propagate and apply the same scheduled tasks
            to all dependent and child resources. Defaults to ``False``.
    """
    # Initialize recursion guard
    _guard = set() if _guard is None else _guard

    name = get_object_name(self.owner, fullname=True)
    if name in _guard:
        return
    _guard.add(name)

    # Validate default scheduling state
    when = when or self.status.next()

    # Collect and sort tasks based on lifecycle state
    scheduled_tasks: list[tuple[Action, SchedulableState]]
    if len(_guard) == 1:
        scheduled_tasks = []
        for task in tasks:
            if isinstance(task, tuple):
                task_action, task_when = task
            else:
                task_action = task
                task_when = when or self.status  # type: ignore
            if not isinstance(task_action, Action):
                task_action = Action(task_action)
            scheduled_tasks.append((task_action, task_when))
        scheduled_tasks = sorted(scheduled_tasks, key=lambda task: task[1])
    else:
        scheduled_tasks = tasks  # type: ignore

    # Schedule or execute tasks based on trigger state
    for task_action, task_when in scheduled_tasks:
        # Handle reached state
        if task_when <= self.status:
            if when_reached == 'skip':
                continue
            if when_reached == 'run':
                task_action(self.owner)
                continue
            # Fallback
            self.status = min(self.status.previous(), task_when)
            self.tasks[task_when].append(task_action)

        # Handle future state
        else:
            if when_future == 'skip':
                continue
            # Await
            self.tasks[task_when].append(task_action)

    # Skip propagation if not required
    if not propagate:
        return

    # Recursively schedule dependents
    for dependent in runtime.get_dependents(self.owner, kind='resources'):
        dependent.__state__.schedule(
            *scheduled_tasks,
            _guard=_guard,
            when=when,
            when_reached=when_reached,
            when_future=when_future,
            propagate=propagate,
        )

    # Recursively schedule children
    for sub in get_subclasses(self.owner):
        assert is_resource(sub)
        sub.__state__.schedule(
            *scheduled_tasks,
            _guard=_guard,
            when=when,
            when_reached=when_reached,
            when_future=when_future,
            propagate=propagate,
        )

ResourceMeta

ResourceMeta(
    name: str,
    bases: tuple[type, ...],
    namespace: dict[str, Any],
    /,
    **kwargs: Any,
)

Bases: ABCMeta, ConfigurableMeta, DeclarativeMeta

Meta class for the base resource class.

Initialize the resource meta class.

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

def __init__(
    cls,
    name: str,
    bases: tuple[type, ...],
    namespace: dict[str, Any],
    /,
    **kwargs: Any,
) -> None:
    """Initialize the resource meta class."""
    # Initialize resource package
    _init_package(cls)

    # Initialize resource declarative model
    _init_declarative_model(cls, bases)
    _init_declarative_fields(cls, bases)

    # Initialize resource services
    _init_specs_and_services(cls, bases)

    # Skip if abstract
    if is_abstract(cls):
        return

    # Initialize SQLAlchemy declarative meta and add the created
    # instrumented attributes to the resource attributes dictionary.
    super().__init__(name, bases, namespace, **kwargs)
    for key, value in list(cls.__dict__.items()):
        # Skip dunder attributes
        if key.startswith('__') and key.endswith('__'):
            continue
        if isinstance(value, InstrumentedAttribute):
            cls.resource_attributes[key] = value

    # Initialize resource identifiers and schemas
    _init_identifiers_and_indexes(cls)
    _init_schemas(cls)

    # Add resource to its package
    cls.resource_package._add_resource(cls)  # type: ignore

    # Schedule initialization build tasks
    cls.__state__.schedule(
        Action(_build_base_model, bound=True),
        Action(_init_declarative_composite_indexes, bound=True),
        when=Lifecycle.INITIALIZING,
    )

    # Schedule finalization build tasks
    cls.__state__.schedule(
        Action(_build_schemas_and_adapter, bound=True),
        when=Lifecycle.READY,
    )

    logger.debug(f"rsc:{cls.resource_config.alias} -> created")

    cls.__state__.flush()

objects property

objects: ResourceManager[BaseResource]

The resource manager that provides service methods on objects.

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

BaseResource

BaseResource(
    __model: BaseModel | None = None,
    __context: dict[str, Any] | None = None,
    /,
    **data: Any,
)

Bases: Representation, Configurable[ResourceConfig]

Base class for all resources.

It exposes the base class for all resources within the Plateforme framework. A resource owns a data model and provides a set of services to interact with.

FIXME: Fix MyPy stub schema models typing. FIXME: Fix MyPy SQLAlchemy instrumented attributes typing.

Attributes:

Name	Type	Description
__config__	ResourceConfig | ConfigDict[ResourceConfigDict]	The configuration class setter for the resource.
__config_services__	tuple[BaseService, ...]	The services configured for the resource.
__config_specs__	tuple[SpecType, ...]	The specifications applied to the resource.
__manager__	tuple[SpecType, ...]	The resource manager used to access and manage the service methods associated with the resource class.
__state__	tuple[SpecType, ...]	The resource state used to manage the resource lifecycle operations, such as creating, initializing, and building the resource class and its dependencies. Additionally, it is used to manage the tasks needed to complete the resource class building process.
__tablename__	str | None	The name of the table in the database.
__table_args__	Any	The arguments for the table in the database.
__mapper_args__	Any	The arguments for the mapper in the database.
_sa_class_manager	ClassManager[Self]	The SQLAlchemy class manager of the resource.
_sa_instance_state	InstanceState[Self]	The SQLAlchemy instance state of the resource used for ORM operations.
_sa_registry	Registry	The SQLAlchemy registry of the resource. It is handled dynamically by the Plateforme application.
__class_vars__	Registry	The names of classvars defined on the model.
__private_attributes__	Registry	Metadata about the private attributes of the model.
__signature__	Registry	The signature for instantiating the model.
__pydantic_complete__	bool	Whether resource model building is completed, or if there are still undefined fields.
__pydantic_validated__	bool	Whether the resource model has been validated or directly constructed.
__pydantic_core_schema__	bool	The pydantic-core schema used to build the SchemaValidator and SchemaSerializer.
__pydantic_model_schema__	CoreSchema	The pydantic-core schema used to build the underlying resource model.
__pydantic_custom_init__	CoreSchema	Whether the resource model has a custom __init__ function.
__pydantic_decorators__	CoreSchema	Metadata containing the decorators defined on the resource model.
__pydantic_generic_metadata__	CoreSchema	Metadata for generic resource models, it contains data used for a similar purpose to __args__, __origin__, and __parameters__ in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__	CoreSchema	Parent namespace of the resource model, used for automatic rebuilding of resource models.
__pydantic_post_init__	None | Literal['resource_post_init']	The name of the post-init method for the resource model, if defined.
__pydantic_root_model__	None | Literal['resource_post_init']	Whether the resource model is an implementation of the RootModel class.
__pydantic_adapter__	TypeAdapterList[BaseResource]	The pydantic TypeAdapterList used to validate and serialize collections of instances of the resource model.
__pydantic_serializer__	TypeAdapterList[BaseResource]	The pydantic-core SchemaSerializer used to dump instances of the resource model.
__pydantic_validator__	TypeAdapterList[BaseResource]	The pydantic-core SchemaValidator used to validate instances of the resource model.
__pydantic_extra__	TypeAdapterList[BaseResource]	An instance attribute with the values of extra fields from validation when resource_config['extra'] == 'allow'.
__pydantic_fields_set__	TypeAdapterList[BaseResource]	An instance attribute with the names of fields explicitly set.
__pydantic_private__	TypeAdapterList[BaseResource]	Instance attribute with the values of private attributes set on the resource model instance.

Initialize a resource instance.

It initializes a resource instance by parsing and validating input data from the data keyword arguments.

Parameters:

Name	Type	Description	Default
__model	BaseModel | None	The model instance to use to create the resource instance. This lets the resource instance be created with an existing model instance within the __new__ method. It is ignored here.	None
__context	dict[str, Any] | None	The context dictionary to use to initialize the resource instance. It is passed to the post-init method.	None
**data	Any	The input data to initialize the resource instance if no model instance is provided. Note that self is explicitly positional-only to allow a same-named field to be defined in the resource model.	{}

Note

The mangled arguments are injected by SQLAlchemy internal behavior within the keyword arguments. Hence, they need to be parsed and removed from the data dictionary before initializing the resource instance.

The validation can be strictly enforced or skipped using the validation context manager validation_manager defined in the schema core module. This can be necessary to allow the resource to be initialized with an already validated model instance and avoid triggering the Pydantic validation logic.

When a resource is created directly by SQLAlchemy, while querying the database for instance, the __init__ method is not called and the data is set directly on the model instance without validation.

Raises:

Type	Description
PlateformeError	If the resource instance could not be initialized from the provided input data. For instance, if the input data is invalid and raises a ValidationError.
AttributeError	If an attribute is not an instrumented attribute of the resource instance while it is defined as a model field.

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

def __init__(
    self,
    __model: BaseModel | None = None,
    __context: dict[str, Any] | None = None,
    /,
    **data: Any,
) -> None:
    """Initialize a resource instance.

    It initializes a resource instance by parsing and validating input data
    from the `data` keyword arguments.

    Args:
        __model: The model instance to use to create the resource instance.
            This lets the resource instance be created with an existing
            model instance within the `__new__` method. It is ignored here.
        __context: The context dictionary to use to initialize the resource
            instance. It is passed to the post-init method.
        **data: The input data to initialize the resource instance if no
            model instance is provided. Note that ``self`` is explicitly
            positional-only to allow a same-named field to be defined in
            the resource model.

    Note:
        The mangled arguments are injected by SQLAlchemy internal behavior
        within the keyword arguments. Hence, they need to be parsed and
        removed from the data dictionary before initializing the resource
        instance.

        The validation can be strictly enforced or skipped using the
        validation context manager `validation_manager` defined in the
        schema core module. This can be necessary to allow the resource to
        be initialized with an already validated model instance and avoid
        triggering the Pydantic validation logic.

        When a resource is created directly by SQLAlchemy, while querying
        the database for instance, the `__init__` method is not called and
        the data is set directly on the model instance without validation.

    Raises:
        PlateformeError: If the resource instance could not be initialized
            from the provided input data. For instance, if the input data
            is invalid and raises a `ValidationError`.
        AttributeError: If an attribute is not an instrumented attribute of
            the resource instance while it is defined as a model field.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    # Store and clean up mangled arguments from the data dictionary. This
    # is necessary to avoid SQLAlchemy from populating the model instance
    # with the mangled attributes.
    mangling_prefix = f'_{BaseResource.__name__}'
    __model = data.pop(f'{mangling_prefix}__model', __model)
    __context = data.pop(f'{mangling_prefix}__context', __context)

    # Retrieve validation mode
    validation_mode = VALIDATION_CONTEXT.get()

    # Retrieve resource model
    model = self.resource_model

    # Validate input data if validation mode is not disabled. There are
    # scenarios where the resource is partially initialized by SQLAlchemy
    # from a database query with data already validated. In this case, the
    # data is set directly on the model instance without triggering any
    # Pydantic validation logic.
    if validation_mode != ValidationMode.DISABLED:
        # Merge initial dictionary with data
        model_dict = { **model.model_dump(exclude_unset=True), **data }
        strict = True if validation_mode == ValidationMode.STRICT \
            else False
        try:
            model.__pydantic_validator__.validate_python(
                model_dict,
                strict=strict,
                context=__context,
                self_instance=model,
            )
        except Exception as error:
            raise PlateformeError(
                f"Failed to initialize resource "
                f"{self.__class__.__qualname__!r}.",
                code='resource-initalization-failed',
            ) from error

    # Call post-init method if defined
    if self.__pydantic_post_init__:
        self.resource_post_init(__context)
        # Update private attributes with values set
        if hasattr(model, '__pydantic_private__') \
                and model.__pydantic_private__ is not None:
            for key, value in data.items():
                if key not in model.__private_attributes__:
                    continue
                model.__pydantic_private__[key] = value

    # Copy and replace the model instance dictionary so that SQLAlchemy can
    # track the changes made to the model instance fields.
    model_dict = model.__dict__.copy()
    object.__setattr__(model, '__dict__', {})
    # Set instrumented attributes for populated model instance fields to
    # trigger SQLAlchemy history tracking of events and updates.
    for key, value in model_dict.items():
        # Skip dunders attributes
        if key.startswith('__') and key.endswith('__'):
            continue
        # Skip deferred attributes
        if value is Deferred:
            continue
        if is_instrumented(self, key):  # type: ignore[no-untyped-call]
            set_instrumented_value(self, key, value)
        else:
            model.__dict__[key] = value

resource_extra property

resource_extra: dict[str, Any] | None

Get extra fields set during validation on this resource instance.

Returns:

Type	Description
dict[str, Any] | None	A dictionary of extra fields, or None if
dict[str, Any] | None	resource_config.extra is not set to "allow".

resource_fields_set property

resource_fields_set: set[str]

Get fields that have been explicitly set on this resource instance.

Returns:

Type	Description
set[str]	A set of strings representing the fields that have been set,
set[str]	i.e. that were not filled from defaults.

resource_adapter

resource_adapter() -> TypeAdapterList[BaseResource]

Get the resource type adapter.

Returns:

Type	Description
TypeAdapterList[BaseResource]	The resource type adapter used to validate and serialize instances
TypeAdapterList[BaseResource]	of the resource model.

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

@classproperty
def resource_adapter(cls) -> TypeAdapterList['BaseResource']:
    """Get the resource type adapter.

    Returns:
        The resource type adapter used to validate and serialize instances
        of the resource model.
    """
    if not hasattr(cls, '__pydantic_adapter__'):
        raise AttributeError(
            "The resource type adapter is not defined. This may be due to "
            "the resource not being fully built or an error occurred "
            "during resource construction."
        )
    return cls.__pydantic_adapter__

resource_computed_fields

resource_computed_fields() -> dict[str, ComputedFieldInfo]

Get computed fields defined on this resource class.

Returns:

Type	Description
dict[str, ComputedFieldInfo]	A metadata dictionary of computed field names and their
dict[str, ComputedFieldInfo]	corresponding ComputedFieldInfo objects.

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

@classproperty
def resource_computed_fields(cls) -> dict[str, ComputedFieldInfo]:
    """Get computed fields defined on this resource class.

    Returns:
        A metadata dictionary of computed field names and their
        corresponding `ComputedFieldInfo` objects.
    """
    return cls.Model.model_computed_fields

resource_fields

resource_fields() -> dict[str, ResourceFieldInfo]

Get fields defined on this resource class.

Returns:

Type	Description
dict[str, ResourceFieldInfo]	A metadata dictionary of field names and their corresponding
dict[str, ResourceFieldInfo]	FieldInfo objects.

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

@classproperty
def resource_fields(cls) -> dict[str, ResourceFieldInfo]:
    """Get fields defined on this resource class.

    Returns:
        A metadata dictionary of field names and their corresponding
        `FieldInfo` objects.
    """
    return cls.Model.model_fields  # type: ignore[return-value]

resource_path

resource_path() -> str

Get the resource API route path.

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

@classproperty
def resource_path(cls) -> str:
    """Get the resource API route path."""
    return f'{cls.resource_package.impl.path}/{cls.resource_config.slug}'

resource_add async

resource_add(
    *, session: AsyncSession | None = None
) -> None

Add the resource instance to the database.

It places an object into the current AsyncSession.

Objects that are in the transient state when passed to the session.add method will move to the pending state, until the next flush, at which point they will move to the persistent state.

Objects that are in the detached state when passed to the session.add method will move to the persistent state directly.

If the transaction used by the AsyncSession is rolled back, objects which were transient when they were passed to session.add will be moved back to the transient state, and will no longer be present within this AsyncSession.

Parameters:

Name	Type	Description	Default
session	AsyncSession | None	The session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None

Raises:

Type	Description
PlateformeError	If the resource instance could not be added to the database.

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

async def resource_add(
    self, *, session: AsyncSession | None = None
) -> None:
    """Add the resource instance to the database.

    It places an object into the current `AsyncSession`.

    Objects that are in the ``transient`` state when passed to the
    `session.add` method will move to the ``pending`` state, until the next
    flush, at which point they will move to the ``persistent`` state.

    Objects that are in the ``detached`` state when passed to the
    `session.add` method will move to the ``persistent`` state directly.

    If the transaction used by the `AsyncSession` is rolled back, objects
    which were transient when they were passed to `session.add` will be
    moved back to the ``transient`` state, and will no longer be present
    within this `AsyncSession`.

    Args:
        session: The session to use for the operation. If not provided, the
            session in the current context is used.
            Defaults to ``None``.

    Raises:
        PlateformeError: If the resource instance could not be added to the
            database.
    """
    try:
        if session:
            session.add(self)
            return
        async with async_session_manager(on_missing='raise') as session:
            session.add(self)
    except Exception as error:
        raise PlateformeError(
            f"Failed to add resource {self!r} to the database.",
            code='resource-add-failed',
        ) from error

resource_construct classmethod

resource_construct(
    _fields_set: set[str] | None = None,
    _model: BaseModel | None = None,
    **data: Any,
) -> Resource

Creates a new instance of the resource class with validated data.

Creates a new resource setting underlying model __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. It behaves as if resource_config.extra = 'allow' was set since it adds all passed values.

Parameters:

Name	Type	Description	Default
_fields_set	set[str] | None	The set of field names accepted by the resource instance.	None
_model	BaseModel | None	The model instance to use to create the resource instance. If not provided, a new model instance is created.	None
**data	Any	Trusted or pre-validated input data to initialize the resource. It is used to set the __dict__ attribute of the underlying model. If a model instance is provided, the data is merged with the existing model data.	{}

Returns:

Type	Description
Resource	A new instance of the resource class with validated data.

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

@classmethod
def resource_construct(
    cls: type[Resource],
    _fields_set: set[str] | None = None,
    _model: BaseModel | None = None,
    **data: Any,
) -> Resource:
    """Creates a new instance of the resource class with validated data.

    Creates a new resource setting underlying model `__dict__` and
    `__pydantic_fields_set__` from trusted or pre-validated data. Default
    values are respected, but no other validation is performed. It behaves
    as if `resource_config.extra = 'allow'` was set since it adds all
    passed values.

    Args:
        _fields_set: The set of field names accepted by the resource
            instance.
        _model: The model instance to use to create the resource instance.
            If not provided, a new model instance is created.
        **data: Trusted or pre-validated input data to initialize the
            resource. It is used to set the `__dict__` attribute of the
            underlying model. If a model instance is provided, the data is
            merged with the existing model data.

    Returns:
        A new instance of the resource class with validated data.
    """
    # Retrieve the model instance from the provided model and data
    if _model is None:
        model = cls.Model.model_construct(_fields_set, **data)
    elif data:
        model = cls.Model.model_construct(
            _fields_set, **_model.model_dump(), **data
        )
    else:
        model = _model

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model)

resource_copy

resource_copy(
    *,
    update: dict[str, Any] | None = None,
    deep: bool = False,
) -> Resource

Returns a copy of the resource.

Parameters:

Name	Type	Description	Default
update	dict[str, Any] | None	Values to add/modify within the resource. Note that if assignment validation is not set to True, the integrity of the data is not validated when updating the resource. Data should be trusted or pre-validated in this case. Defaults to None.	None
deep	bool	Set to True to make a deep copy of the model.	False

Returns:

Type	Description
Resource	A new copy of the model instance with the updated values.

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

def resource_copy(
    self: Resource,
    *,
    update: dict[str, Any] | None = None,
    deep: bool = False,
) -> Resource:
    """Returns a copy of the resource.

    Args:
        update: Values to add/modify within the resource. Note that if
            assignment validation is not set to ``True``, the integrity of
            the data is not validated when updating the resource. Data
            should be trusted or pre-validated in this case.
            Defaults to ``None``.
        deep: Set to ``True`` to make a deep copy of the model.

    Returns:
        A new copy of the model instance with the updated values.
    """
    # Copy the model instance
    model = self.resource_model.model_copy(update=update, deep=deep)

    # Return a new resource instance without triggering validation
    cls = self.__class__
    with validation_manager(mode='disabled'):
        return cls(model)

resource_delete async

resource_delete(
    *, session: AsyncSession | None = None
) -> None

Delete the resource instance from the database.

It marks an instance as deleted.

The object is assumed to be either persistent or detached when passed; after the method is called, the object will remain in the persistent state until the next flush proceeds. During this time, the object will also be a member of the session.deleted collection.

When the next flush proceeds, the object will move to the deleted state, indicating a DELETE statement was emitted for its row within the current transaction. When the transaction is successfully committed, the deleted object is moved to the detached state and is no longer present within this AsyncSession.

Parameters:

Name	Type	Description	Default
session	AsyncSession | None	The session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None

Raises:

Type	Description
PlateformeError	If the resource instance could not be deleted from the database.

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

async def resource_delete(
    self, *, session: AsyncSession | None = None,
) -> None:
    """Delete the resource instance from the database.

    It marks an instance as deleted.

    The object is assumed to be either ``persistent`` or ``detached`` when
    passed; after the method is called, the object will remain in the
    ``persistent`` state until the next flush proceeds. During this time,
    the object will also be a member of the `session.deleted` collection.

    When the next flush proceeds, the object will move to the ``deleted``
    state, indicating a ``DELETE`` statement was emitted for its row within
    the current transaction. When the transaction is successfully
    committed, the deleted object is moved to the ``detached`` state and is
    no longer present within this `AsyncSession`.

    Args:
        session: The session to use for the operation. If not provided, the
            session in the current context is used.
            Defaults to ``None``.

    Raises:
        PlateformeError: If the resource instance could not be deleted from
            the database.
    """
    try:
        if session:
            await session.delete(self)
            return
        async with async_session_manager(on_missing='raise') as session:
            await session.delete(self)
    except Exception as error:
        raise PlateformeError(
            f"Failed to delete resource {self!r} from the database.",
            code='resource-delete-failed',
        ) from error

resource_dump

resource_dump(
    *,
    mode: Literal["json", "python", "raw"] | str = "python",
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> dict[str, Any]

Generate a dictionary representation of the resource.

It is used to dump the resource instance to a dictionary representation of the resource, optionally specifying which fields to include or exclude.

Parameters:

Name	Type	Description	Default
mode	Literal['json', 'python', 'raw'] | str	The mode in which to_python should run: - If mode is json, the output will only contain JSON serializable types. - If mode is python, the output may contain non JSON serializable Python objects. - If mode is raw, the output will contain raw values. Defaults to python.	'python'
include	IncEx | None	A list of fields to include in the output. Defaults to None.	None
exclude	IncEx | None	A list of fields to exclude from the output. Defaults to None.	None
by_alias	bool	Whether to use the field's alias in the dictionary key if defined. Defaults to False.	False
exclude_unset	bool	Whether to exclude fields that have not been explicitly set. Defaults to False.	False
exclude_defaults	bool	Whether to exclude fields that are set to their default value. Defaults to False.	False
exclude_none	bool	Whether to exclude fields that have a value of None. Defaults to False.	False
round_trip	bool	If True, dumped values should be valid as input for non-idempotent types such as Json[T]. Defaults to False.	False
warnings	bool	Whether to log warnings when invalid fields are encountered. Defaults to True.	True

Returns:

Type	Description
dict[str, Any]	A dictionary representation of the resource.

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

def resource_dump(
    self,
    *,
    mode: Literal['json', 'python', 'raw'] | str = 'python',
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> dict[str, Any]:
    """Generate a dictionary representation of the resource.

    It is used to dump the resource instance to a dictionary representation
    of the resource, optionally specifying which fields to include or
    exclude.

    Args:
        mode: The mode in which `to_python` should run:
            - If mode is ``json``, the output will only contain JSON
                serializable types.
            - If mode is ``python``, the output may contain non JSON
                serializable Python objects.
            - If mode is ``raw``, the output will contain raw values.
            Defaults to ``python``.
        include: A list of fields to include in the output.
            Defaults to ``None``.
        exclude: A list of fields to exclude from the output.
            Defaults to ``None``.
        by_alias: Whether to use the field's alias in the dictionary key if
            defined. Defaults to ``False``.
        exclude_unset: Whether to exclude fields that have not been
            explicitly set. Defaults to ``False``.
        exclude_defaults: Whether to exclude fields that are set to their
            default value. Defaults to ``False``.
        exclude_none: Whether to exclude fields that have a value of
            ``None``. Defaults to ``False``.
        round_trip: If ``True``, dumped values should be valid as input for
            non-idempotent types such as `Json[T]`. Defaults to ``False``.
        warnings: Whether to log warnings when invalid fields are
            encountered. Defaults to ``True``.

    Returns:
        A dictionary representation of the resource.
    """
    return self.resource_model.model_dump(
        mode=mode,
        include=include,
        exclude=exclude,
        by_alias=by_alias,
        exclude_unset=exclude_unset,
        exclude_defaults=exclude_defaults,
        exclude_none=exclude_none,
        round_trip=round_trip,
        warnings=warnings,
    )

resource_dump_json

resource_dump_json(
    *,
    indent: int | None = None,
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> str

Generate a JSON representation of the resource.

It is used to dump the resource instance to a JSON representation of the resource using Pydantic's to_json method, optionally specifying which fields to include or exclude.

Parameters:

Name	Type	Description	Default
indent	int | None	Indentation to use in the JSON output. If None is passed, the output will be compact. Defaults to None.	None
include	IncEx | None	A list of fields to include in the JSON output. Defaults to None.	None
exclude	IncEx | None	A list of fields to exclude from the JSON output. Defaults to None.	None
by_alias	bool	Whether to use the field's alias in the dictionary key if defined. Defaults to False.	False
exclude_unset	bool	Whether to exclude fields that have not been explicitly set. Defaults to False.	False
exclude_defaults	bool	Whether to exclude fields that are set to their default value. Defaults to False.	False
exclude_none	bool	Whether to exclude fields that have a value of None. Defaults to False.	False
round_trip	bool	If True, dumped values should be valid as input for non-idempotent types such as Json[T]. Defaults to False.	False
warnings	bool	Whether to log warnings when invalid fields are encountered. Defaults to True.	True

Returns:

Type	Description
str	A JSON string representation of the resource.

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

def resource_dump_json(
    self,
    *,
    indent: int | None = None,
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> str:
    """Generate a JSON representation of the resource.

    It is used to dump the resource instance to a JSON representation of
    the resource using Pydantic's `to_json` method, optionally specifying
    which fields to include or exclude.

    Args:
        indent: Indentation to use in the JSON output. If ``None`` is
            passed, the output will be compact. Defaults to ``None``.
        include: A list of fields to include in the JSON output.
            Defaults to ``None``.
        exclude: A list of fields to exclude from the JSON output.
            Defaults to ``None``.
        by_alias: Whether to use the field's alias in the dictionary key if
            defined. Defaults to ``False``.
        exclude_unset: Whether to exclude fields that have not been
            explicitly set. Defaults to ``False``.
        exclude_defaults: Whether to exclude fields that are set to their
            default value. Defaults to ``False``.
        exclude_none: Whether to exclude fields that have a value of
            ``None``. Defaults to ``False``.
        round_trip: If ``True``, dumped values should be valid as input for
            non-idempotent types such as `Json[T]`. Defaults to ``False``.
        warnings: Whether to log warnings when invalid fields are
            encountered. Defaults to ``True``.

    Returns:
        A JSON string representation of the resource.
    """
    return self.resource_model.model_dump_json(
        indent=indent,
        include=include,
        exclude=exclude,
        by_alias=by_alias,
        exclude_unset=exclude_unset,
        exclude_defaults=exclude_defaults,
        exclude_none=exclude_none,
        round_trip=round_trip,
        warnings=warnings,
    )

resource_json_schema classmethod

resource_json_schema(
    by_alias: bool = True,
    ref_template: str = DEFAULT_REF_TEMPLATE,
    schema_generator: type[
        GenerateJsonSchema
    ] = GenerateJsonSchema,
    mode: JsonSchemaMode = "validation",
    source: JsonSchemaSource = "key",
) -> dict[str, Any]

Generates a JSON schema for a resource class.

Parameters:

Name	Type	Description	Default
by_alias	bool	Whether to use field aliases when generating the schema, i.e. if True, fields will be serialized according to their alias, otherwise according to their attribute name. Defaults to True.	True
ref_template	str	The template format string used when generating reference names. Defaults to DEFAULT_REF_TEMPLATE.	DEFAULT_REF_TEMPLATE
schema_generator	type[GenerateJsonSchema]	The class to use for generating the JSON Schema.	GenerateJsonSchema
mode	JsonSchemaMode	The mode to use for generating the JSON Schema. It can be either validation or serialization where respectively the schema is generated for validating data or serializing data. Defaults to validation.	'validation'
source	JsonSchemaSource	The source type to use for generating the resources JSON schema. It can be either key , model, or both where the latter accepts, when applicable, integer and string values for key identifiers in addition to the standard model schema generation. Defaults to key.	'key'

Returns:

Type	Description
dict[str, Any]	The generated JSON schema of the resource class.

Note

The schema generator class can be overridden to customize the logic used to generate the JSON schema. This can be done by subclassing the GenerateJsonSchema class and passing the subclass as the schema_generator argument.

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

@classmethod
def resource_json_schema(
    cls,
    by_alias: bool = True,
    ref_template: str = DEFAULT_REF_TEMPLATE,
    schema_generator: type[GenerateJsonSchema] = GenerateJsonSchema,
    mode: JsonSchemaMode = 'validation',
    source: JsonSchemaSource = 'key',
) -> dict[str, Any]:
    """Generates a JSON schema for a resource class.

    Args:
        by_alias: Whether to use field aliases when generating the schema,
            i.e. if ``True``, fields will be serialized according to their
            alias, otherwise according to their attribute name.
            Defaults to ``True``.
        ref_template: The template format string used when generating
            reference names. Defaults to ``DEFAULT_REF_TEMPLATE``.
        schema_generator: The class to use for generating the JSON Schema.
        mode: The mode to use for generating the JSON Schema. It can be
            either ``validation`` or ``serialization`` where respectively
            the schema is generated for validating data or serializing
            data. Defaults to ``validation``.
        source: The source type to use for generating the resources JSON
            schema. It can be either ``key`` , ``model``, or ``both`` where
            the latter accepts, when applicable, integer and string values
            for key identifiers in addition to the standard model schema
            generation. Defaults to ``key``.

    Returns:
        The generated JSON schema of the resource class.

    Note:
        The schema generator class can be overridden to customize the
        logic used to generate the JSON schema. This can be done by
        subclassing the `GenerateJsonSchema` class and passing the subclass
        as the `schema_generator` argument.
    """
    return cls.Model.model_json_schema(
        by_alias=by_alias,
        ref_template=ref_template,
        schema_generator=schema_generator,
        mode=mode,
        source=source,
    )

resource_keychain

resource_keychain() -> tuple[Key[Self], ...]

Collect all keys set on the resource instance.

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

def resource_keychain(self) -> tuple[Key[Self], ...]:
    """Collect all keys set on the resource instance."""
    keychain: tuple[Key[Self], ...] = ()

    for identifier in self.resource_identifiers:
        assignments = {
            field_alias: getattr(self, field_alias)
            for field_alias in identifier
        }
        key = Key.validate(
            assignments,
            resource=self.__class__,
            validate_assignment=False,
        )
        keychain += (key,)

    return keychain

resource_merge async

resource_merge(
    *,
    load: bool = True,
    options: Sequence[ORMOption] | None = None,
    session: AsyncSession | None = None,
) -> Resource

Merge the resource instance with the database.

It copies the state of the resource instance into a corresponding instance in the database within the current AsyncSession.

First, it examines the primary key attributes if available, otherwise the resource identifier keychain, and attempts to reconcile it with an instance of the same identifier in the session. If not found locally, it attempts to load the object from the database based on the identifier, and if none can be located, creates a new instance in the session. The state of each attribute on the source instance is then copied to the target instance. The resulting target instance is then returned by the resource_merge method; the original source instance is left unmodified, and un-associated with the AsyncSession if not already.

This operation cascades to associated instances if the association is mapped with cascade='merge'.

Parameters:

Name	Type	Description	Default
load	bool	Whether to load the resource instance from the database if not found in the session. Defaults to True. When False, the method switches into a "high performance" mode which causes it to forego emitting history events as well as all database access. This flag is used for cases such as transferring graphs of objects into an AsyncSession from a second level cache, or to transfer just-loaded objects into the AsyncSession owned by a worker thread or process without re-querying the database. The load=False use case adds the caveat that the given object has to be in a "clean" state, that is, has no pending changes to be flushed - even if the incoming object is detached from any AsyncSession. This is so that when the merge operation populates local attributes and cascades to related objects and collections, the values can be "stamped" onto the target object as is, without generating any history or attribute events, and without the need to reconcile the incoming data with any existing related objects or collections that might not be loaded. The resulting objects from load=False are always produced as "clean", so it is only appropriate that the given objects should be "clean" as well, else this suggests a mis-use of the method.	True
options	Sequence[ORMOption] | None	Optional sequence of loader options which will be applied to the session.get method when the merge operation loads the existing version of the object from the database.	None
session	AsyncSession | None	The session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None

Returns:

Type	Description
Resource	The merged resource instance.

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

async def resource_merge(
    self: Resource,
    *,
    load: bool = True,
    options: Sequence[ORMOption] | None = None,
    session: AsyncSession | None = None,
) -> Resource:
    """Merge the resource instance with the database.

    It copies the state of the resource instance into a corresponding
    instance in the database within the current `AsyncSession`.

    First, it examines the primary key attributes if available, otherwise
    the resource identifier keychain, and attempts to reconcile it with an
    instance of the same identifier in the session. If not found locally,
    it attempts to load the object from the database based on the
    identifier, and if none can be located, creates a new instance in the
    session. The state of each attribute on the source instance is then
    copied to the target instance. The resulting target instance is then
    returned by the `resource_merge` method; the original source instance
    is left unmodified, and un-associated with the `AsyncSession` if not
    already.

    This operation cascades to associated instances if the association is
    mapped with ``cascade='merge'``.

    Args:
        load: Whether to load the resource instance from the database if
            not found in the session. Defaults to ``True``.

            When ``False``, the method switches into a "high performance"
            mode which causes it to forego emitting history events as well
            as all database access. This flag is used for cases such as
            transferring graphs of objects into an `AsyncSession` from a
            second level cache, or to transfer just-loaded objects into the
            `AsyncSession` owned by a worker thread or process without
            re-querying the database.

            The ``load=False`` use case adds the caveat that the given
            object has to be in a "clean" state, that is, has no pending
            changes to be flushed - even if the incoming object is detached
            from any `AsyncSession`. This is so that when the merge
            operation populates local attributes and cascades to related
            objects and collections, the values can be "stamped" onto the
            target object as is, without generating any history or
            attribute events, and without the need to reconcile the
            incoming data with any existing related objects or collections
            that might not be loaded. The resulting objects from
            ``load=False`` are always produced as "clean", so it is only
            appropriate that the given objects should be "clean" as well,
            else this suggests a mis-use of the method.
        options: Optional sequence of loader options which will be applied
            to the `session.get` method when the merge operation loads the
            existing version of the object from the database.
        session: The session to use for the operation. If not provided, the
            session in the current context is used.
            Defaults to ``None``.

    Returns:
        The merged resource instance.
    """
    try:
        if session:
            return await session.merge(self, load=load, options=options)
        async with async_session_manager(on_missing='raise') as session:
            return await session.merge(self, load=load, options=options)
    except Exception as error:
        raise PlateformeError(
            f"Failed to merge resource {self!r} within the "
            f"database.",
            code='resource-merge-failed',
        ) from error

resource_parametrized_name classmethod

resource_parametrized_name(
    params: tuple[type[Any], ...],
) -> str

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic Base resopurces.

Parameters:

Name	Type	Description	Default
params	tuple[type[Any], ...]	Tuple of types of the class. Given a generic class Resource with 2 type variables and a concrete resource Resource[str, int], the value (str, int) would be passed to params.	required

Returns:

Type	Description
str	String representing the new class where params are passed to
str	cls as type variables.

Raises:

Type	Description
TypeError	Raised when trying to generate concrete names for non-generic resources.

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

@classmethod
def resource_parametrized_name(cls, params: tuple[type[Any], ...]) -> str:
    """Compute the class name for parametrizations of generic classes.

    This method can be overridden to achieve a custom naming scheme for
    generic Base resopurces.

    Args:
        params: Tuple of types of the class. Given a generic class
            `Resource` with 2 type variables and a concrete resource
            `Resource[str, int]`, the value ``(str, int)`` would be passed
            to `params`.

    Returns:
        String representing the new class where `params` are passed to
        `cls` as type variables.

    Raises:
        TypeError: Raised when trying to generate concrete names for
            non-generic resources.
    """
    if not issubclass(cls, Generic):  # type: ignore[arg-type]
        raise TypeError(
            "Concrete names should only be generated for generic "
            "resources."
        )
    return cls.Model.model_parametrized_name(params)

resource_post_init

resource_post_init(
    __context: dict[str, Any] | None = None,
) -> None

Post-initialization method for the resource class.

Override this method to perform additional initialization after the __init__ and resource_construct methods have been called. This is useful in scenarios where it is necessary to perform additional initialization steps after the resource has been fully initialized.

Parameters:

Name	Type	Description	Default
__context	dict[str, Any] | None	The context object passed to the resource instance.	None

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

def resource_post_init(
    self, __context: dict[str, Any] | None = None
) -> None:
    """Post-initialization method for the resource class.

    Override this method to perform additional initialization after the
    `__init__` and `resource_construct` methods have been called. This is
    useful in scenarios where it is necessary to perform additional
    initialization steps after the resource has been fully initialized.

    Args:
        __context: The context object passed to the resource instance.
    """
    ...

resource_rebuild classmethod

resource_rebuild(
    *,
    force: bool = False,
    raise_errors: bool = True,
    _parent_namespace_depth: int = 2,
    _types_namespace: dict[str, Any] | None = None,
) -> bool | None

Rebuild the pydantic-core schema for the resource.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

Name	Type	Description	Default
force	bool	Whether to force the rebuilding of the resource schema. Defaults to False.	False
raise_errors	bool	Whether to raise errors or fail silently. Defaults to True.	True
_parent_namespace_depth	int	The depth level of the parent namespace. Defaults to 2.	2
_types_namespace	dict[str, Any] | None	The types namespace. Defaults to None.	None

Raises:

Type	Description
PlateformeError	If the resource adapter or schema could not be rebuilt and raise_errors is set to True.
PydanticUndefinedAnnotation	If PydanticUndefinedAnnotation occurs in__get_pydantic_core_schema__ and raise_errors is set to True.

Returns:

Type	Description
bool | None	Returns None if the schema is already "complete" and rebuilding
bool | None	was not required. If rebuilding was required, returns True if
bool | None	rebuilding was successful, otherwise False if an error
bool | None	occurred and raise_errors is set to False.

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

@classmethod
def resource_rebuild(
    cls,
    *,
    force: bool = False,
    raise_errors: bool = True,
    _parent_namespace_depth: int = 2,
    _types_namespace: dict[str, Any] | None = None,
) -> bool | None:
    """Rebuild the pydantic-core schema for the resource.

    This may be necessary when one of the annotations is a `ForwardRef`
    which could not be resolved during the initial attempt to build the
    schema, and automatic rebuilding fails.

    Args:
        force: Whether to force the rebuilding of the resource schema.
            Defaults to ``False``.
        raise_errors: Whether to raise errors or fail silently.
            Defaults to ``True``.
        _parent_namespace_depth: The depth level of the parent namespace.
            Defaults to 2.
        _types_namespace: The types namespace. Defaults to ``None``.

    Raises:
        PlateformeError: If the resource adapter or schema could not be
            rebuilt and `raise_errors` is set to ``True``.
        PydanticUndefinedAnnotation: If `PydanticUndefinedAnnotation`
            occurs in`__get_pydantic_core_schema__` and `raise_errors` is
            set to ``True``.

    Returns:
        Returns ``None`` if the schema is already "complete" and rebuilding
        was not required. If rebuilding was required, returns ``True`` if
        rebuilding was successful, otherwise ``False`` if an error
        occurred and `raise_errors` is set to ``False``.
    """
    build_status = None

    # Rebuild resource model
    build_status = cls.Model.model_rebuild(
        force=force,
        raise_errors=raise_errors,
        _parent_namespace_depth=_parent_namespace_depth,
        _types_namespace=_types_namespace,
    )

    # Rebuild resource schema
    if build_status:
        build_status = cls._rebuild_schemas(
            raise_errors=raise_errors,
            _parent_namespace_depth=_parent_namespace_depth,
            _types_namespace=_types_namespace,
        )

    # Rebuild resource adapter
    if build_status:
        build_status = cls._rebuild_adapter(raise_errors=raise_errors)

    return build_status

resource_revalidate

resource_revalidate(
    *,
    force: bool = False,
    raise_errors: bool = True,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> bool | None

Revalidate the resource instance.

It revalidates the resource instance in place, enforcing the types strictly if specified. If the resource instance has already been validated, it will not be revalidated unless the force argument is set to True.

Parameters:

Name	Type	Description	Default
force	bool	Whether to force the revalidation of the resource instance. Defaults to False.	False
raise_errors	bool	Whether to raise errors or fail silently. Defaults to True.	True
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Raises:

Type	Description
ValidationError	If the resource instance could not be validated and raise_errors is set to True.

Returns:

Type	Description
bool | None	Returns None if the resource instance is already "validated"
bool | None	and revalidation was not required. If validation was required,
bool | None	returns True if validation was successful, otherwise False
bool | None	if an error occurred and raise_errors is set to False.

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

def resource_revalidate(
    self,
    *,
    force: bool = False,
    raise_errors: bool = True,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> bool | None:
    """Revalidate the resource instance.

    It revalidates the resource instance in place, enforcing the types
    strictly if specified. If the resource instance has already been
    validated, it will not be revalidated unless the `force` argument is
    set to ``True``.

    Args:
        force: Whether to force the revalidation of the resource instance.
            Defaults to ``False``.
        raise_errors: Whether to raise errors or fail silently.
            Defaults to ``True``.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Raises:
        ValidationError: If the resource instance could not be validated
            and `raise_errors` is set to ``True``.

    Returns:
        Returns ``None`` if the resource instance is already "validated"
        and revalidation was not required. If validation was required,
        returns ``True`` if validation was successful, otherwise ``False``
        if an error occurred and `raise_errors` is set to ``False``.
    """
    return self.resource_model.model_revalidate(
        force=force,
        raise_errors=raise_errors,
        strict=strict,
        context=context,
    )

resource_update

resource_update(
    obj: Any,
    *,
    update: dict[str, Any] | None = None,
    from_attributes: bool | None = None,
) -> None

Update the resource with the given object and update dictionary.

Parameters:

Name	Type	Description	Default
obj	Any	The object to update the resource with. It can be a dictionary or an object with attributes (if from_attributes is set to True). If it is a dictionary, the keys must match the resource field names if extra fields are not allowed.	required
update	dict[str, Any] | None	Values to add/modify within the resource. Note that if assignment validation is not set to True, the integrity of the data is not validated when updating the resource. Data should be trusted or pre-validated in this case. Defaults to None.	None
from_attributes	bool | None	Whether to extract data from object attributes. Defaults to None.	None

Raises:

Type	Description
ValidationError	If the object could not be validated.
ValueError	If strict or context are set when validate_assignment is set to False.

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

def resource_update(
    self,
    obj: Any,
    *,
    update: dict[str, Any] | None = None,
    from_attributes: bool | None = None,
) -> None:
    """Update the resource with the given object and update dictionary.

    Args:
        obj: The object to update the resource with. It can be a dictionary
            or an object with attributes (if `from_attributes` is set to
            ``True``). If it is a dictionary, the keys must match the
            resource field names if extra fields are not allowed.
        update: Values to add/modify within the resource. Note that if
            assignment validation is not set to ``True``, the integrity of
            the data is not validated when updating the resource. Data
            should be trusted or pre-validated in this case.
            Defaults to ``None``.
        from_attributes: Whether to extract data from object attributes.
            Defaults to ``None``.

    Raises:
        ValidationError: If the object could not be validated.
        ValueError: If `strict` or `context` are set when
            `validate_assignment` is set to ``False``.
    """
    # Collect update
    update = (update or {}).copy()
    if from_attributes:
        for field_name in self.resource_fields:
            if hasattr(obj, field_name):
                update.setdefault(field_name, getattr(obj, field_name))
    elif isinstance(obj, dict):
        update = {**obj, **update}

    # Process update
    for key, value in update.items():
        setattr(self, key, value)

resource_validate classmethod

resource_validate(
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource

Validate the given object against the resource.

It validates the provided object against the resource model and returns the validated resource instance. The validation can be strictly enforced or skipped using the strict argument. The data can be extracted from the object attributes using the from_attributes argument. Additional context can be passed to the validator using the context argument.

Parameters:

Name	Type	Description	Default
obj	Any	The object to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
from_attributes	bool | None	Whether to extract data from the object attributes.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Resource	A validated resource instance.

Raises:

Type	Description
ValidationError	If the object could not be validated.

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

@classmethod
def resource_validate(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource:
    """Validate the given object against the resource.

    It validates the provided object against the resource model and returns
    the validated resource instance. The validation can be strictly
    enforced or skipped using the `strict` argument. The data can be
    extracted from the object attributes using the `from_attributes`
    argument. Additional context can be passed to the validator using the
    `context` argument.

    Args:
        obj: The object to validate.
        strict: Whether to enforce types strictly.
        from_attributes: Whether to extract data from the object
            attributes.
        context: Extra variables to pass to the validator.

    Returns:
        A validated resource instance.

    Raises:
        ValidationError: If the object could not be validated.
    """
    # Validate a new model instance from the provided object
    model = cls.Model.model_validate(
        obj,
        strict=strict,
        from_attributes=from_attributes,
        context=context,
    )

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model, context)

resource_validate_many classmethod

resource_validate_many(
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]

Validate the given object collection against the resource.

It validates the provided object collection against the resource model and returns the validated resource instances. The validation can be strictly enforced or skipped using the strict argument. The data can be extracted from the object items attributes using the from_attributes argument. Additional context can be passed to the validator using the context argument.

Parameters:

Name	Type	Description	Default
obj	Any	The object collection to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
from_attributes	bool | None	Whether to extract data from the object collection items attributes.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Sequence[Resource]	A validated collection of resource instances.

Raises:

Type	Description
ValidationError	If the object collection could not be validated.

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

@classmethod
def resource_validate_many(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]:
    """Validate the given object collection against the resource.

    It validates the provided object collection against the resource model
    and returns the validated resource instances. The validation can be
    strictly enforced or skipped using the `strict` argument. The data can
    be extracted from the object items attributes using the
    `from_attributes` argument. Additional context can be passed to the
    validator using the `context` argument.

    Args:
        obj: The object collection to validate.
        strict: Whether to enforce types strictly.
        from_attributes: Whether to extract data from the object
            collection items attributes.
        context: Extra variables to pass to the validator.

    Returns:
        A validated collection of resource instances.

    Raises:
        ValidationError: If the object collection could not be validated.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    return cls.resource_adapter.validate_python(  # type: ignore
        obj,
        strict=strict,
        from_attributes=from_attributes,
        context=context,
    )

resource_validate_json classmethod

resource_validate_json(
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource

Validate the given JSON data against the resource.

Parameters:

Name	Type	Description	Default
json_data	str | bytes | bytearray	The JSON data to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Resource	A validated resource instance.

Raises:

Type	Description
ValueError	If json_data is not a JSON string.
ValidationError	If the object could not be validated.

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

@classmethod
def resource_validate_json(
    cls: type[Resource],
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource:
    """Validate the given JSON data against the resource.

    Args:
        json_data: The JSON data to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated resource instance.

    Raises:
        ValueError: If `json_data` is not a JSON string.
        ValidationError: If the object could not be validated.
    """
    # Validate a new resource instance from the provided data
    model = cls.Model.model_validate_json(
        json_data, strict=strict, context=context
    )

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model, context)

resource_validate_json_many classmethod

resource_validate_json_many(
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]

Validate the given JSON data collection against the resource.

Parameters:

Name	Type	Description	Default
json_data	str | bytes | bytearray	The JSON data collection to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Sequence[Resource]	A validated collection of resource instances.

Raises:

Type	Description
ValueError	If json_data is not a JSON string.
ValidationError	If the object collection could not be validated.

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

@classmethod
def resource_validate_json_many(
    cls: type[Resource],
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]:
    """Validate the given JSON data collection against the resource.

    Args:
        json_data: The JSON data collection to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated collection of resource instances.

    Raises:
        ValueError: If `json_data` is not a JSON string.
        ValidationError: If the object collection could not be validated.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    return cls.resource_adapter.validate_json(  # type: ignore
        json_data, strict=strict, context=context
    )

resource_validate_strings classmethod

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

Validate the given string object against the resource.

Parameters:

Name	Type	Description	Default
obj	Any	The string object to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Resource	A validated resource instance.

Raises:

Type	Description
ValidationError	If the object could not be validated.

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

@classmethod
def resource_validate_strings(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource:
    """Validate the given string object against the resource.

    Args:
        obj: The string object to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated resource instance.

    Raises:
        ValidationError: If the object could not be validated.
    """
    # Validate a new resource instance from the provided string object
    model = cls.Model.model_validate_strings(
        obj, strict=strict, context=context
    )

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model, context)

resource_validate_strings_many classmethod

resource_validate_strings_many(
    obj: Any,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]

Validate the given string object collection against the resource.

Parameters:

Name	Type	Description	Default
obj	Any	The string object collection to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Sequence[Resource]	A validated collection of resource instances.

Raises:

Type	Description
ValidationError	If the object collection could not be validated.

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

@classmethod
def resource_validate_strings_many(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]:
    """Validate the given string object collection against the resource.

    Args:
        obj: The string object collection to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated collection of resource instances.

    Raises:
        ValidationError: If the object collection could not be validated.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    return cls.resource_adapter.validate_strings(  # type: ignore
        obj, strict=strict, context=context
    )

CRUDResource

CRUDResource(
    __model: BaseModel | None = None,
    __context: dict[str, Any] | None = None,
    /,
    **data: Any,
)

Bases: BaseResource

A CRUD resource.

It inherits directly from the BaseResource and adds the CRUDService to its configuration services. This enables all CRUD operations on the resource instances.

Initialize a resource instance.

It initializes a resource instance by parsing and validating input data from the data keyword arguments.

Parameters:

Name	Type	Description	Default
__model	BaseModel | None	The model instance to use to create the resource instance. This lets the resource instance be created with an existing model instance within the __new__ method. It is ignored here.	None
__context	dict[str, Any] | None	The context dictionary to use to initialize the resource instance. It is passed to the post-init method.	None
**data	Any	The input data to initialize the resource instance if no model instance is provided. Note that self is explicitly positional-only to allow a same-named field to be defined in the resource model.	{}

Note

The mangled arguments are injected by SQLAlchemy internal behavior within the keyword arguments. Hence, they need to be parsed and removed from the data dictionary before initializing the resource instance.

The validation can be strictly enforced or skipped using the validation context manager validation_manager defined in the schema core module. This can be necessary to allow the resource to be initialized with an already validated model instance and avoid triggering the Pydantic validation logic.

When a resource is created directly by SQLAlchemy, while querying the database for instance, the __init__ method is not called and the data is set directly on the model instance without validation.

Raises:

Type	Description
PlateformeError	If the resource instance could not be initialized from the provided input data. For instance, if the input data is invalid and raises a ValidationError.
AttributeError	If an attribute is not an instrumented attribute of the resource instance while it is defined as a model field.

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

def __init__(
    self,
    __model: BaseModel | None = None,
    __context: dict[str, Any] | None = None,
    /,
    **data: Any,
) -> None:
    """Initialize a resource instance.

    It initializes a resource instance by parsing and validating input data
    from the `data` keyword arguments.

    Args:
        __model: The model instance to use to create the resource instance.
            This lets the resource instance be created with an existing
            model instance within the `__new__` method. It is ignored here.
        __context: The context dictionary to use to initialize the resource
            instance. It is passed to the post-init method.
        **data: The input data to initialize the resource instance if no
            model instance is provided. Note that ``self`` is explicitly
            positional-only to allow a same-named field to be defined in
            the resource model.

    Note:
        The mangled arguments are injected by SQLAlchemy internal behavior
        within the keyword arguments. Hence, they need to be parsed and
        removed from the data dictionary before initializing the resource
        instance.

        The validation can be strictly enforced or skipped using the
        validation context manager `validation_manager` defined in the
        schema core module. This can be necessary to allow the resource to
        be initialized with an already validated model instance and avoid
        triggering the Pydantic validation logic.

        When a resource is created directly by SQLAlchemy, while querying
        the database for instance, the `__init__` method is not called and
        the data is set directly on the model instance without validation.

    Raises:
        PlateformeError: If the resource instance could not be initialized
            from the provided input data. For instance, if the input data
            is invalid and raises a `ValidationError`.
        AttributeError: If an attribute is not an instrumented attribute of
            the resource instance while it is defined as a model field.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    # Store and clean up mangled arguments from the data dictionary. This
    # is necessary to avoid SQLAlchemy from populating the model instance
    # with the mangled attributes.
    mangling_prefix = f'_{BaseResource.__name__}'
    __model = data.pop(f'{mangling_prefix}__model', __model)
    __context = data.pop(f'{mangling_prefix}__context', __context)

    # Retrieve validation mode
    validation_mode = VALIDATION_CONTEXT.get()

    # Retrieve resource model
    model = self.resource_model

    # Validate input data if validation mode is not disabled. There are
    # scenarios where the resource is partially initialized by SQLAlchemy
    # from a database query with data already validated. In this case, the
    # data is set directly on the model instance without triggering any
    # Pydantic validation logic.
    if validation_mode != ValidationMode.DISABLED:
        # Merge initial dictionary with data
        model_dict = { **model.model_dump(exclude_unset=True), **data }
        strict = True if validation_mode == ValidationMode.STRICT \
            else False
        try:
            model.__pydantic_validator__.validate_python(
                model_dict,
                strict=strict,
                context=__context,
                self_instance=model,
            )
        except Exception as error:
            raise PlateformeError(
                f"Failed to initialize resource "
                f"{self.__class__.__qualname__!r}.",
                code='resource-initalization-failed',
            ) from error

    # Call post-init method if defined
    if self.__pydantic_post_init__:
        self.resource_post_init(__context)
        # Update private attributes with values set
        if hasattr(model, '__pydantic_private__') \
                and model.__pydantic_private__ is not None:
            for key, value in data.items():
                if key not in model.__private_attributes__:
                    continue
                model.__pydantic_private__[key] = value

    # Copy and replace the model instance dictionary so that SQLAlchemy can
    # track the changes made to the model instance fields.
    model_dict = model.__dict__.copy()
    object.__setattr__(model, '__dict__', {})
    # Set instrumented attributes for populated model instance fields to
    # trigger SQLAlchemy history tracking of events and updates.
    for key, value in model_dict.items():
        # Skip dunders attributes
        if key.startswith('__') and key.endswith('__'):
            continue
        # Skip deferred attributes
        if value is Deferred:
            continue
        if is_instrumented(self, key):  # type: ignore[no-untyped-call]
            set_instrumented_value(self, key, value)
        else:
            model.__dict__[key] = value

resource_extra property

resource_extra: dict[str, Any] | None

Get extra fields set during validation on this resource instance.

Returns:

Type	Description
dict[str, Any] | None	A dictionary of extra fields, or None if
dict[str, Any] | None	resource_config.extra is not set to "allow".

resource_fields_set property

resource_fields_set: set[str]

Get fields that have been explicitly set on this resource instance.

Returns:

Type	Description
set[str]	A set of strings representing the fields that have been set,
set[str]	i.e. that were not filled from defaults.

resource_adapter

resource_adapter() -> TypeAdapterList[BaseResource]

Get the resource type adapter.

Returns:

Type	Description
TypeAdapterList[BaseResource]	The resource type adapter used to validate and serialize instances
TypeAdapterList[BaseResource]	of the resource model.

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

@classproperty
def resource_adapter(cls) -> TypeAdapterList['BaseResource']:
    """Get the resource type adapter.

    Returns:
        The resource type adapter used to validate and serialize instances
        of the resource model.
    """
    if not hasattr(cls, '__pydantic_adapter__'):
        raise AttributeError(
            "The resource type adapter is not defined. This may be due to "
            "the resource not being fully built or an error occurred "
            "during resource construction."
        )
    return cls.__pydantic_adapter__

resource_computed_fields

resource_computed_fields() -> dict[str, ComputedFieldInfo]

Get computed fields defined on this resource class.

Returns:

Type	Description
dict[str, ComputedFieldInfo]	A metadata dictionary of computed field names and their
dict[str, ComputedFieldInfo]	corresponding ComputedFieldInfo objects.

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

@classproperty
def resource_computed_fields(cls) -> dict[str, ComputedFieldInfo]:
    """Get computed fields defined on this resource class.

    Returns:
        A metadata dictionary of computed field names and their
        corresponding `ComputedFieldInfo` objects.
    """
    return cls.Model.model_computed_fields

resource_fields

resource_fields() -> dict[str, ResourceFieldInfo]

Get fields defined on this resource class.

Returns:

Type	Description
dict[str, ResourceFieldInfo]	A metadata dictionary of field names and their corresponding
dict[str, ResourceFieldInfo]	FieldInfo objects.

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

@classproperty
def resource_fields(cls) -> dict[str, ResourceFieldInfo]:
    """Get fields defined on this resource class.

    Returns:
        A metadata dictionary of field names and their corresponding
        `FieldInfo` objects.
    """
    return cls.Model.model_fields  # type: ignore[return-value]

resource_path

resource_path() -> str

Get the resource API route path.

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

@classproperty
def resource_path(cls) -> str:
    """Get the resource API route path."""
    return f'{cls.resource_package.impl.path}/{cls.resource_config.slug}'

resource_add async

resource_add(
    *, session: AsyncSession | None = None
) -> None

Add the resource instance to the database.

It places an object into the current AsyncSession.

Objects that are in the transient state when passed to the session.add method will move to the pending state, until the next flush, at which point they will move to the persistent state.

Objects that are in the detached state when passed to the session.add method will move to the persistent state directly.

If the transaction used by the AsyncSession is rolled back, objects which were transient when they were passed to session.add will be moved back to the transient state, and will no longer be present within this AsyncSession.

Parameters:

Name	Type	Description	Default
session	AsyncSession | None	The session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None

Raises:

Type	Description
PlateformeError	If the resource instance could not be added to the database.

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

async def resource_add(
    self, *, session: AsyncSession | None = None
) -> None:
    """Add the resource instance to the database.

    It places an object into the current `AsyncSession`.

    Objects that are in the ``transient`` state when passed to the
    `session.add` method will move to the ``pending`` state, until the next
    flush, at which point they will move to the ``persistent`` state.

    Objects that are in the ``detached`` state when passed to the
    `session.add` method will move to the ``persistent`` state directly.

    If the transaction used by the `AsyncSession` is rolled back, objects
    which were transient when they were passed to `session.add` will be
    moved back to the ``transient`` state, and will no longer be present
    within this `AsyncSession`.

    Args:
        session: The session to use for the operation. If not provided, the
            session in the current context is used.
            Defaults to ``None``.

    Raises:
        PlateformeError: If the resource instance could not be added to the
            database.
    """
    try:
        if session:
            session.add(self)
            return
        async with async_session_manager(on_missing='raise') as session:
            session.add(self)
    except Exception as error:
        raise PlateformeError(
            f"Failed to add resource {self!r} to the database.",
            code='resource-add-failed',
        ) from error

resource_construct classmethod

resource_construct(
    _fields_set: set[str] | None = None,
    _model: BaseModel | None = None,
    **data: Any,
) -> Resource

Creates a new instance of the resource class with validated data.

Creates a new resource setting underlying model __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. It behaves as if resource_config.extra = 'allow' was set since it adds all passed values.

Parameters:

Name	Type	Description	Default
_fields_set	set[str] | None	The set of field names accepted by the resource instance.	None
_model	BaseModel | None	The model instance to use to create the resource instance. If not provided, a new model instance is created.	None
**data	Any	Trusted or pre-validated input data to initialize the resource. It is used to set the __dict__ attribute of the underlying model. If a model instance is provided, the data is merged with the existing model data.	{}

Returns:

Type	Description
Resource	A new instance of the resource class with validated data.

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

@classmethod
def resource_construct(
    cls: type[Resource],
    _fields_set: set[str] | None = None,
    _model: BaseModel | None = None,
    **data: Any,
) -> Resource:
    """Creates a new instance of the resource class with validated data.

    Creates a new resource setting underlying model `__dict__` and
    `__pydantic_fields_set__` from trusted or pre-validated data. Default
    values are respected, but no other validation is performed. It behaves
    as if `resource_config.extra = 'allow'` was set since it adds all
    passed values.

    Args:
        _fields_set: The set of field names accepted by the resource
            instance.
        _model: The model instance to use to create the resource instance.
            If not provided, a new model instance is created.
        **data: Trusted or pre-validated input data to initialize the
            resource. It is used to set the `__dict__` attribute of the
            underlying model. If a model instance is provided, the data is
            merged with the existing model data.

    Returns:
        A new instance of the resource class with validated data.
    """
    # Retrieve the model instance from the provided model and data
    if _model is None:
        model = cls.Model.model_construct(_fields_set, **data)
    elif data:
        model = cls.Model.model_construct(
            _fields_set, **_model.model_dump(), **data
        )
    else:
        model = _model

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model)

resource_copy

resource_copy(
    *,
    update: dict[str, Any] | None = None,
    deep: bool = False,
) -> Resource

Returns a copy of the resource.

Parameters:

Name	Type	Description	Default
update	dict[str, Any] | None	Values to add/modify within the resource. Note that if assignment validation is not set to True, the integrity of the data is not validated when updating the resource. Data should be trusted or pre-validated in this case. Defaults to None.	None
deep	bool	Set to True to make a deep copy of the model.	False

Returns:

Type	Description
Resource	A new copy of the model instance with the updated values.

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

def resource_copy(
    self: Resource,
    *,
    update: dict[str, Any] | None = None,
    deep: bool = False,
) -> Resource:
    """Returns a copy of the resource.

    Args:
        update: Values to add/modify within the resource. Note that if
            assignment validation is not set to ``True``, the integrity of
            the data is not validated when updating the resource. Data
            should be trusted or pre-validated in this case.
            Defaults to ``None``.
        deep: Set to ``True`` to make a deep copy of the model.

    Returns:
        A new copy of the model instance with the updated values.
    """
    # Copy the model instance
    model = self.resource_model.model_copy(update=update, deep=deep)

    # Return a new resource instance without triggering validation
    cls = self.__class__
    with validation_manager(mode='disabled'):
        return cls(model)

resource_delete async

resource_delete(
    *, session: AsyncSession | None = None
) -> None

Delete the resource instance from the database.

It marks an instance as deleted.

The object is assumed to be either persistent or detached when passed; after the method is called, the object will remain in the persistent state until the next flush proceeds. During this time, the object will also be a member of the session.deleted collection.

When the next flush proceeds, the object will move to the deleted state, indicating a DELETE statement was emitted for its row within the current transaction. When the transaction is successfully committed, the deleted object is moved to the detached state and is no longer present within this AsyncSession.

Parameters:

Name	Type	Description	Default
session	AsyncSession | None	The session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None

Raises:

Type	Description
PlateformeError	If the resource instance could not be deleted from the database.

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

async def resource_delete(
    self, *, session: AsyncSession | None = None,
) -> None:
    """Delete the resource instance from the database.

    It marks an instance as deleted.

    The object is assumed to be either ``persistent`` or ``detached`` when
    passed; after the method is called, the object will remain in the
    ``persistent`` state until the next flush proceeds. During this time,
    the object will also be a member of the `session.deleted` collection.

    When the next flush proceeds, the object will move to the ``deleted``
    state, indicating a ``DELETE`` statement was emitted for its row within
    the current transaction. When the transaction is successfully
    committed, the deleted object is moved to the ``detached`` state and is
    no longer present within this `AsyncSession`.

    Args:
        session: The session to use for the operation. If not provided, the
            session in the current context is used.
            Defaults to ``None``.

    Raises:
        PlateformeError: If the resource instance could not be deleted from
            the database.
    """
    try:
        if session:
            await session.delete(self)
            return
        async with async_session_manager(on_missing='raise') as session:
            await session.delete(self)
    except Exception as error:
        raise PlateformeError(
            f"Failed to delete resource {self!r} from the database.",
            code='resource-delete-failed',
        ) from error

resource_dump

resource_dump(
    *,
    mode: Literal["json", "python", "raw"] | str = "python",
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> dict[str, Any]

Generate a dictionary representation of the resource.

It is used to dump the resource instance to a dictionary representation of the resource, optionally specifying which fields to include or exclude.

Parameters:

Name	Type	Description	Default
mode	Literal['json', 'python', 'raw'] | str	The mode in which to_python should run: - If mode is json, the output will only contain JSON serializable types. - If mode is python, the output may contain non JSON serializable Python objects. - If mode is raw, the output will contain raw values. Defaults to python.	'python'
include	IncEx | None	A list of fields to include in the output. Defaults to None.	None
exclude	IncEx | None	A list of fields to exclude from the output. Defaults to None.	None
by_alias	bool	Whether to use the field's alias in the dictionary key if defined. Defaults to False.	False
exclude_unset	bool	Whether to exclude fields that have not been explicitly set. Defaults to False.	False
exclude_defaults	bool	Whether to exclude fields that are set to their default value. Defaults to False.	False
exclude_none	bool	Whether to exclude fields that have a value of None. Defaults to False.	False
round_trip	bool	If True, dumped values should be valid as input for non-idempotent types such as Json[T]. Defaults to False.	False
warnings	bool	Whether to log warnings when invalid fields are encountered. Defaults to True.	True

Returns:

Type	Description
dict[str, Any]	A dictionary representation of the resource.

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

def resource_dump(
    self,
    *,
    mode: Literal['json', 'python', 'raw'] | str = 'python',
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> dict[str, Any]:
    """Generate a dictionary representation of the resource.

    It is used to dump the resource instance to a dictionary representation
    of the resource, optionally specifying which fields to include or
    exclude.

    Args:
        mode: The mode in which `to_python` should run:
            - If mode is ``json``, the output will only contain JSON
                serializable types.
            - If mode is ``python``, the output may contain non JSON
                serializable Python objects.
            - If mode is ``raw``, the output will contain raw values.
            Defaults to ``python``.
        include: A list of fields to include in the output.
            Defaults to ``None``.
        exclude: A list of fields to exclude from the output.
            Defaults to ``None``.
        by_alias: Whether to use the field's alias in the dictionary key if
            defined. Defaults to ``False``.
        exclude_unset: Whether to exclude fields that have not been
            explicitly set. Defaults to ``False``.
        exclude_defaults: Whether to exclude fields that are set to their
            default value. Defaults to ``False``.
        exclude_none: Whether to exclude fields that have a value of
            ``None``. Defaults to ``False``.
        round_trip: If ``True``, dumped values should be valid as input for
            non-idempotent types such as `Json[T]`. Defaults to ``False``.
        warnings: Whether to log warnings when invalid fields are
            encountered. Defaults to ``True``.

    Returns:
        A dictionary representation of the resource.
    """
    return self.resource_model.model_dump(
        mode=mode,
        include=include,
        exclude=exclude,
        by_alias=by_alias,
        exclude_unset=exclude_unset,
        exclude_defaults=exclude_defaults,
        exclude_none=exclude_none,
        round_trip=round_trip,
        warnings=warnings,
    )

resource_dump_json

resource_dump_json(
    *,
    indent: int | None = None,
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> str

Generate a JSON representation of the resource.

It is used to dump the resource instance to a JSON representation of the resource using Pydantic's to_json method, optionally specifying which fields to include or exclude.

Parameters:

Name	Type	Description	Default
indent	int | None	Indentation to use in the JSON output. If None is passed, the output will be compact. Defaults to None.	None
include	IncEx | None	A list of fields to include in the JSON output. Defaults to None.	None
exclude	IncEx | None	A list of fields to exclude from the JSON output. Defaults to None.	None
by_alias	bool	Whether to use the field's alias in the dictionary key if defined. Defaults to False.	False
exclude_unset	bool	Whether to exclude fields that have not been explicitly set. Defaults to False.	False
exclude_defaults	bool	Whether to exclude fields that are set to their default value. Defaults to False.	False
exclude_none	bool	Whether to exclude fields that have a value of None. Defaults to False.	False
round_trip	bool	If True, dumped values should be valid as input for non-idempotent types such as Json[T]. Defaults to False.	False
warnings	bool	Whether to log warnings when invalid fields are encountered. Defaults to True.	True

Returns:

Type	Description
str	A JSON string representation of the resource.

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

def resource_dump_json(
    self,
    *,
    indent: int | None = None,
    include: IncEx | None = None,
    exclude: IncEx | None = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
) -> str:
    """Generate a JSON representation of the resource.

    It is used to dump the resource instance to a JSON representation of
    the resource using Pydantic's `to_json` method, optionally specifying
    which fields to include or exclude.

    Args:
        indent: Indentation to use in the JSON output. If ``None`` is
            passed, the output will be compact. Defaults to ``None``.
        include: A list of fields to include in the JSON output.
            Defaults to ``None``.
        exclude: A list of fields to exclude from the JSON output.
            Defaults to ``None``.
        by_alias: Whether to use the field's alias in the dictionary key if
            defined. Defaults to ``False``.
        exclude_unset: Whether to exclude fields that have not been
            explicitly set. Defaults to ``False``.
        exclude_defaults: Whether to exclude fields that are set to their
            default value. Defaults to ``False``.
        exclude_none: Whether to exclude fields that have a value of
            ``None``. Defaults to ``False``.
        round_trip: If ``True``, dumped values should be valid as input for
            non-idempotent types such as `Json[T]`. Defaults to ``False``.
        warnings: Whether to log warnings when invalid fields are
            encountered. Defaults to ``True``.

    Returns:
        A JSON string representation of the resource.
    """
    return self.resource_model.model_dump_json(
        indent=indent,
        include=include,
        exclude=exclude,
        by_alias=by_alias,
        exclude_unset=exclude_unset,
        exclude_defaults=exclude_defaults,
        exclude_none=exclude_none,
        round_trip=round_trip,
        warnings=warnings,
    )

resource_json_schema classmethod

resource_json_schema(
    by_alias: bool = True,
    ref_template: str = DEFAULT_REF_TEMPLATE,
    schema_generator: type[
        GenerateJsonSchema
    ] = GenerateJsonSchema,
    mode: JsonSchemaMode = "validation",
    source: JsonSchemaSource = "key",
) -> dict[str, Any]

Generates a JSON schema for a resource class.

Parameters:

Name	Type	Description	Default
by_alias	bool	Whether to use field aliases when generating the schema, i.e. if True, fields will be serialized according to their alias, otherwise according to their attribute name. Defaults to True.	True
ref_template	str	The template format string used when generating reference names. Defaults to DEFAULT_REF_TEMPLATE.	DEFAULT_REF_TEMPLATE
schema_generator	type[GenerateJsonSchema]	The class to use for generating the JSON Schema.	GenerateJsonSchema
mode	JsonSchemaMode	The mode to use for generating the JSON Schema. It can be either validation or serialization where respectively the schema is generated for validating data or serializing data. Defaults to validation.	'validation'
source	JsonSchemaSource	The source type to use for generating the resources JSON schema. It can be either key , model, or both where the latter accepts, when applicable, integer and string values for key identifiers in addition to the standard model schema generation. Defaults to key.	'key'

Returns:

Type	Description
dict[str, Any]	The generated JSON schema of the resource class.

Note

The schema generator class can be overridden to customize the logic used to generate the JSON schema. This can be done by subclassing the GenerateJsonSchema class and passing the subclass as the schema_generator argument.

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

@classmethod
def resource_json_schema(
    cls,
    by_alias: bool = True,
    ref_template: str = DEFAULT_REF_TEMPLATE,
    schema_generator: type[GenerateJsonSchema] = GenerateJsonSchema,
    mode: JsonSchemaMode = 'validation',
    source: JsonSchemaSource = 'key',
) -> dict[str, Any]:
    """Generates a JSON schema for a resource class.

    Args:
        by_alias: Whether to use field aliases when generating the schema,
            i.e. if ``True``, fields will be serialized according to their
            alias, otherwise according to their attribute name.
            Defaults to ``True``.
        ref_template: The template format string used when generating
            reference names. Defaults to ``DEFAULT_REF_TEMPLATE``.
        schema_generator: The class to use for generating the JSON Schema.
        mode: The mode to use for generating the JSON Schema. It can be
            either ``validation`` or ``serialization`` where respectively
            the schema is generated for validating data or serializing
            data. Defaults to ``validation``.
        source: The source type to use for generating the resources JSON
            schema. It can be either ``key`` , ``model``, or ``both`` where
            the latter accepts, when applicable, integer and string values
            for key identifiers in addition to the standard model schema
            generation. Defaults to ``key``.

    Returns:
        The generated JSON schema of the resource class.

    Note:
        The schema generator class can be overridden to customize the
        logic used to generate the JSON schema. This can be done by
        subclassing the `GenerateJsonSchema` class and passing the subclass
        as the `schema_generator` argument.
    """
    return cls.Model.model_json_schema(
        by_alias=by_alias,
        ref_template=ref_template,
        schema_generator=schema_generator,
        mode=mode,
        source=source,
    )

resource_keychain

resource_keychain() -> tuple[Key[Self], ...]

Collect all keys set on the resource instance.

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

def resource_keychain(self) -> tuple[Key[Self], ...]:
    """Collect all keys set on the resource instance."""
    keychain: tuple[Key[Self], ...] = ()

    for identifier in self.resource_identifiers:
        assignments = {
            field_alias: getattr(self, field_alias)
            for field_alias in identifier
        }
        key = Key.validate(
            assignments,
            resource=self.__class__,
            validate_assignment=False,
        )
        keychain += (key,)

    return keychain

resource_merge async

resource_merge(
    *,
    load: bool = True,
    options: Sequence[ORMOption] | None = None,
    session: AsyncSession | None = None,
) -> Resource

Merge the resource instance with the database.

It copies the state of the resource instance into a corresponding instance in the database within the current AsyncSession.

First, it examines the primary key attributes if available, otherwise the resource identifier keychain, and attempts to reconcile it with an instance of the same identifier in the session. If not found locally, it attempts to load the object from the database based on the identifier, and if none can be located, creates a new instance in the session. The state of each attribute on the source instance is then copied to the target instance. The resulting target instance is then returned by the resource_merge method; the original source instance is left unmodified, and un-associated with the AsyncSession if not already.

This operation cascades to associated instances if the association is mapped with cascade='merge'.

Parameters:

Name	Type	Description	Default
load	bool	Whether to load the resource instance from the database if not found in the session. Defaults to True. When False, the method switches into a "high performance" mode which causes it to forego emitting history events as well as all database access. This flag is used for cases such as transferring graphs of objects into an AsyncSession from a second level cache, or to transfer just-loaded objects into the AsyncSession owned by a worker thread or process without re-querying the database. The load=False use case adds the caveat that the given object has to be in a "clean" state, that is, has no pending changes to be flushed - even if the incoming object is detached from any AsyncSession. This is so that when the merge operation populates local attributes and cascades to related objects and collections, the values can be "stamped" onto the target object as is, without generating any history or attribute events, and without the need to reconcile the incoming data with any existing related objects or collections that might not be loaded. The resulting objects from load=False are always produced as "clean", so it is only appropriate that the given objects should be "clean" as well, else this suggests a mis-use of the method.	True
options	Sequence[ORMOption] | None	Optional sequence of loader options which will be applied to the session.get method when the merge operation loads the existing version of the object from the database.	None
session	AsyncSession | None	The session to use for the operation. If not provided, the session in the current context is used. Defaults to None.	None

Returns:

Type	Description
Resource	The merged resource instance.

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

async def resource_merge(
    self: Resource,
    *,
    load: bool = True,
    options: Sequence[ORMOption] | None = None,
    session: AsyncSession | None = None,
) -> Resource:
    """Merge the resource instance with the database.

    It copies the state of the resource instance into a corresponding
    instance in the database within the current `AsyncSession`.

    First, it examines the primary key attributes if available, otherwise
    the resource identifier keychain, and attempts to reconcile it with an
    instance of the same identifier in the session. If not found locally,
    it attempts to load the object from the database based on the
    identifier, and if none can be located, creates a new instance in the
    session. The state of each attribute on the source instance is then
    copied to the target instance. The resulting target instance is then
    returned by the `resource_merge` method; the original source instance
    is left unmodified, and un-associated with the `AsyncSession` if not
    already.

    This operation cascades to associated instances if the association is
    mapped with ``cascade='merge'``.

    Args:
        load: Whether to load the resource instance from the database if
            not found in the session. Defaults to ``True``.

            When ``False``, the method switches into a "high performance"
            mode which causes it to forego emitting history events as well
            as all database access. This flag is used for cases such as
            transferring graphs of objects into an `AsyncSession` from a
            second level cache, or to transfer just-loaded objects into the
            `AsyncSession` owned by a worker thread or process without
            re-querying the database.

            The ``load=False`` use case adds the caveat that the given
            object has to be in a "clean" state, that is, has no pending
            changes to be flushed - even if the incoming object is detached
            from any `AsyncSession`. This is so that when the merge
            operation populates local attributes and cascades to related
            objects and collections, the values can be "stamped" onto the
            target object as is, without generating any history or
            attribute events, and without the need to reconcile the
            incoming data with any existing related objects or collections
            that might not be loaded. The resulting objects from
            ``load=False`` are always produced as "clean", so it is only
            appropriate that the given objects should be "clean" as well,
            else this suggests a mis-use of the method.
        options: Optional sequence of loader options which will be applied
            to the `session.get` method when the merge operation loads the
            existing version of the object from the database.
        session: The session to use for the operation. If not provided, the
            session in the current context is used.
            Defaults to ``None``.

    Returns:
        The merged resource instance.
    """
    try:
        if session:
            return await session.merge(self, load=load, options=options)
        async with async_session_manager(on_missing='raise') as session:
            return await session.merge(self, load=load, options=options)
    except Exception as error:
        raise PlateformeError(
            f"Failed to merge resource {self!r} within the "
            f"database.",
            code='resource-merge-failed',
        ) from error

resource_parametrized_name classmethod

resource_parametrized_name(
    params: tuple[type[Any], ...],
) -> str

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic Base resopurces.

Parameters:

Name	Type	Description	Default
params	tuple[type[Any], ...]	Tuple of types of the class. Given a generic class Resource with 2 type variables and a concrete resource Resource[str, int], the value (str, int) would be passed to params.	required

Returns:

Type	Description
str	String representing the new class where params are passed to
str	cls as type variables.

Raises:

Type	Description
TypeError	Raised when trying to generate concrete names for non-generic resources.

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

@classmethod
def resource_parametrized_name(cls, params: tuple[type[Any], ...]) -> str:
    """Compute the class name for parametrizations of generic classes.

    This method can be overridden to achieve a custom naming scheme for
    generic Base resopurces.

    Args:
        params: Tuple of types of the class. Given a generic class
            `Resource` with 2 type variables and a concrete resource
            `Resource[str, int]`, the value ``(str, int)`` would be passed
            to `params`.

    Returns:
        String representing the new class where `params` are passed to
        `cls` as type variables.

    Raises:
        TypeError: Raised when trying to generate concrete names for
            non-generic resources.
    """
    if not issubclass(cls, Generic):  # type: ignore[arg-type]
        raise TypeError(
            "Concrete names should only be generated for generic "
            "resources."
        )
    return cls.Model.model_parametrized_name(params)

resource_post_init

resource_post_init(
    __context: dict[str, Any] | None = None,
) -> None

Post-initialization method for the resource class.

Override this method to perform additional initialization after the __init__ and resource_construct methods have been called. This is useful in scenarios where it is necessary to perform additional initialization steps after the resource has been fully initialized.

Parameters:

Name	Type	Description	Default
__context	dict[str, Any] | None	The context object passed to the resource instance.	None

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

def resource_post_init(
    self, __context: dict[str, Any] | None = None
) -> None:
    """Post-initialization method for the resource class.

    Override this method to perform additional initialization after the
    `__init__` and `resource_construct` methods have been called. This is
    useful in scenarios where it is necessary to perform additional
    initialization steps after the resource has been fully initialized.

    Args:
        __context: The context object passed to the resource instance.
    """
    ...

resource_rebuild classmethod

resource_rebuild(
    *,
    force: bool = False,
    raise_errors: bool = True,
    _parent_namespace_depth: int = 2,
    _types_namespace: dict[str, Any] | None = None,
) -> bool | None

Rebuild the pydantic-core schema for the resource.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

Name	Type	Description	Default
force	bool	Whether to force the rebuilding of the resource schema. Defaults to False.	False
raise_errors	bool	Whether to raise errors or fail silently. Defaults to True.	True
_parent_namespace_depth	int	The depth level of the parent namespace. Defaults to 2.	2
_types_namespace	dict[str, Any] | None	The types namespace. Defaults to None.	None

Raises:

Type	Description
PlateformeError	If the resource adapter or schema could not be rebuilt and raise_errors is set to True.
PydanticUndefinedAnnotation	If PydanticUndefinedAnnotation occurs in__get_pydantic_core_schema__ and raise_errors is set to True.

Returns:

Type	Description
bool | None	Returns None if the schema is already "complete" and rebuilding
bool | None	was not required. If rebuilding was required, returns True if
bool | None	rebuilding was successful, otherwise False if an error
bool | None	occurred and raise_errors is set to False.

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

@classmethod
def resource_rebuild(
    cls,
    *,
    force: bool = False,
    raise_errors: bool = True,
    _parent_namespace_depth: int = 2,
    _types_namespace: dict[str, Any] | None = None,
) -> bool | None:
    """Rebuild the pydantic-core schema for the resource.

    This may be necessary when one of the annotations is a `ForwardRef`
    which could not be resolved during the initial attempt to build the
    schema, and automatic rebuilding fails.

    Args:
        force: Whether to force the rebuilding of the resource schema.
            Defaults to ``False``.
        raise_errors: Whether to raise errors or fail silently.
            Defaults to ``True``.
        _parent_namespace_depth: The depth level of the parent namespace.
            Defaults to 2.
        _types_namespace: The types namespace. Defaults to ``None``.

    Raises:
        PlateformeError: If the resource adapter or schema could not be
            rebuilt and `raise_errors` is set to ``True``.
        PydanticUndefinedAnnotation: If `PydanticUndefinedAnnotation`
            occurs in`__get_pydantic_core_schema__` and `raise_errors` is
            set to ``True``.

    Returns:
        Returns ``None`` if the schema is already "complete" and rebuilding
        was not required. If rebuilding was required, returns ``True`` if
        rebuilding was successful, otherwise ``False`` if an error
        occurred and `raise_errors` is set to ``False``.
    """
    build_status = None

    # Rebuild resource model
    build_status = cls.Model.model_rebuild(
        force=force,
        raise_errors=raise_errors,
        _parent_namespace_depth=_parent_namespace_depth,
        _types_namespace=_types_namespace,
    )

    # Rebuild resource schema
    if build_status:
        build_status = cls._rebuild_schemas(
            raise_errors=raise_errors,
            _parent_namespace_depth=_parent_namespace_depth,
            _types_namespace=_types_namespace,
        )

    # Rebuild resource adapter
    if build_status:
        build_status = cls._rebuild_adapter(raise_errors=raise_errors)

    return build_status

resource_revalidate

resource_revalidate(
    *,
    force: bool = False,
    raise_errors: bool = True,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> bool | None

Revalidate the resource instance.

It revalidates the resource instance in place, enforcing the types strictly if specified. If the resource instance has already been validated, it will not be revalidated unless the force argument is set to True.

Parameters:

Name	Type	Description	Default
force	bool	Whether to force the revalidation of the resource instance. Defaults to False.	False
raise_errors	bool	Whether to raise errors or fail silently. Defaults to True.	True
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Raises:

Type	Description
ValidationError	If the resource instance could not be validated and raise_errors is set to True.

Returns:

Type	Description
bool | None	Returns None if the resource instance is already "validated"
bool | None	and revalidation was not required. If validation was required,
bool | None	returns True if validation was successful, otherwise False
bool | None	if an error occurred and raise_errors is set to False.

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

def resource_revalidate(
    self,
    *,
    force: bool = False,
    raise_errors: bool = True,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> bool | None:
    """Revalidate the resource instance.

    It revalidates the resource instance in place, enforcing the types
    strictly if specified. If the resource instance has already been
    validated, it will not be revalidated unless the `force` argument is
    set to ``True``.

    Args:
        force: Whether to force the revalidation of the resource instance.
            Defaults to ``False``.
        raise_errors: Whether to raise errors or fail silently.
            Defaults to ``True``.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Raises:
        ValidationError: If the resource instance could not be validated
            and `raise_errors` is set to ``True``.

    Returns:
        Returns ``None`` if the resource instance is already "validated"
        and revalidation was not required. If validation was required,
        returns ``True`` if validation was successful, otherwise ``False``
        if an error occurred and `raise_errors` is set to ``False``.
    """
    return self.resource_model.model_revalidate(
        force=force,
        raise_errors=raise_errors,
        strict=strict,
        context=context,
    )

resource_update

resource_update(
    obj: Any,
    *,
    update: dict[str, Any] | None = None,
    from_attributes: bool | None = None,
) -> None

Update the resource with the given object and update dictionary.

Parameters:

Name	Type	Description	Default
obj	Any	The object to update the resource with. It can be a dictionary or an object with attributes (if from_attributes is set to True). If it is a dictionary, the keys must match the resource field names if extra fields are not allowed.	required
update	dict[str, Any] | None	Values to add/modify within the resource. Note that if assignment validation is not set to True, the integrity of the data is not validated when updating the resource. Data should be trusted or pre-validated in this case. Defaults to None.	None
from_attributes	bool | None	Whether to extract data from object attributes. Defaults to None.	None

Raises:

Type	Description
ValidationError	If the object could not be validated.
ValueError	If strict or context are set when validate_assignment is set to False.

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

def resource_update(
    self,
    obj: Any,
    *,
    update: dict[str, Any] | None = None,
    from_attributes: bool | None = None,
) -> None:
    """Update the resource with the given object and update dictionary.

    Args:
        obj: The object to update the resource with. It can be a dictionary
            or an object with attributes (if `from_attributes` is set to
            ``True``). If it is a dictionary, the keys must match the
            resource field names if extra fields are not allowed.
        update: Values to add/modify within the resource. Note that if
            assignment validation is not set to ``True``, the integrity of
            the data is not validated when updating the resource. Data
            should be trusted or pre-validated in this case.
            Defaults to ``None``.
        from_attributes: Whether to extract data from object attributes.
            Defaults to ``None``.

    Raises:
        ValidationError: If the object could not be validated.
        ValueError: If `strict` or `context` are set when
            `validate_assignment` is set to ``False``.
    """
    # Collect update
    update = (update or {}).copy()
    if from_attributes:
        for field_name in self.resource_fields:
            if hasattr(obj, field_name):
                update.setdefault(field_name, getattr(obj, field_name))
    elif isinstance(obj, dict):
        update = {**obj, **update}

    # Process update
    for key, value in update.items():
        setattr(self, key, value)

resource_validate classmethod

resource_validate(
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource

Validate the given object against the resource.

It validates the provided object against the resource model and returns the validated resource instance. The validation can be strictly enforced or skipped using the strict argument. The data can be extracted from the object attributes using the from_attributes argument. Additional context can be passed to the validator using the context argument.

Parameters:

Name	Type	Description	Default
obj	Any	The object to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
from_attributes	bool | None	Whether to extract data from the object attributes.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Resource	A validated resource instance.

Raises:

Type	Description
ValidationError	If the object could not be validated.

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

@classmethod
def resource_validate(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource:
    """Validate the given object against the resource.

    It validates the provided object against the resource model and returns
    the validated resource instance. The validation can be strictly
    enforced or skipped using the `strict` argument. The data can be
    extracted from the object attributes using the `from_attributes`
    argument. Additional context can be passed to the validator using the
    `context` argument.

    Args:
        obj: The object to validate.
        strict: Whether to enforce types strictly.
        from_attributes: Whether to extract data from the object
            attributes.
        context: Extra variables to pass to the validator.

    Returns:
        A validated resource instance.

    Raises:
        ValidationError: If the object could not be validated.
    """
    # Validate a new model instance from the provided object
    model = cls.Model.model_validate(
        obj,
        strict=strict,
        from_attributes=from_attributes,
        context=context,
    )

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model, context)

resource_validate_many classmethod

resource_validate_many(
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]

Validate the given object collection against the resource.

It validates the provided object collection against the resource model and returns the validated resource instances. The validation can be strictly enforced or skipped using the strict argument. The data can be extracted from the object items attributes using the from_attributes argument. Additional context can be passed to the validator using the context argument.

Parameters:

Name	Type	Description	Default
obj	Any	The object collection to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
from_attributes	bool | None	Whether to extract data from the object collection items attributes.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Sequence[Resource]	A validated collection of resource instances.

Raises:

Type	Description
ValidationError	If the object collection could not be validated.

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

@classmethod
def resource_validate_many(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    from_attributes: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]:
    """Validate the given object collection against the resource.

    It validates the provided object collection against the resource model
    and returns the validated resource instances. The validation can be
    strictly enforced or skipped using the `strict` argument. The data can
    be extracted from the object items attributes using the
    `from_attributes` argument. Additional context can be passed to the
    validator using the `context` argument.

    Args:
        obj: The object collection to validate.
        strict: Whether to enforce types strictly.
        from_attributes: Whether to extract data from the object
            collection items attributes.
        context: Extra variables to pass to the validator.

    Returns:
        A validated collection of resource instances.

    Raises:
        ValidationError: If the object collection could not be validated.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    return cls.resource_adapter.validate_python(  # type: ignore
        obj,
        strict=strict,
        from_attributes=from_attributes,
        context=context,
    )

resource_validate_json classmethod

resource_validate_json(
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource

Validate the given JSON data against the resource.

Parameters:

Name	Type	Description	Default
json_data	str | bytes | bytearray	The JSON data to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Resource	A validated resource instance.

Raises:

Type	Description
ValueError	If json_data is not a JSON string.
ValidationError	If the object could not be validated.

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

@classmethod
def resource_validate_json(
    cls: type[Resource],
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource:
    """Validate the given JSON data against the resource.

    Args:
        json_data: The JSON data to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated resource instance.

    Raises:
        ValueError: If `json_data` is not a JSON string.
        ValidationError: If the object could not be validated.
    """
    # Validate a new resource instance from the provided data
    model = cls.Model.model_validate_json(
        json_data, strict=strict, context=context
    )

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model, context)

resource_validate_json_many classmethod

resource_validate_json_many(
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]

Validate the given JSON data collection against the resource.

Parameters:

Name	Type	Description	Default
json_data	str | bytes | bytearray	The JSON data collection to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Sequence[Resource]	A validated collection of resource instances.

Raises:

Type	Description
ValueError	If json_data is not a JSON string.
ValidationError	If the object collection could not be validated.

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

@classmethod
def resource_validate_json_many(
    cls: type[Resource],
    json_data: str | bytes | bytearray,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]:
    """Validate the given JSON data collection against the resource.

    Args:
        json_data: The JSON data collection to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated collection of resource instances.

    Raises:
        ValueError: If `json_data` is not a JSON string.
        ValidationError: If the object collection could not be validated.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    return cls.resource_adapter.validate_json(  # type: ignore
        json_data, strict=strict, context=context
    )

resource_validate_strings classmethod

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

Validate the given string object against the resource.

Parameters:

Name	Type	Description	Default
obj	Any	The string object to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Resource	A validated resource instance.

Raises:

Type	Description
ValidationError	If the object could not be validated.

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

@classmethod
def resource_validate_strings(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Resource:
    """Validate the given string object against the resource.

    Args:
        obj: The string object to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated resource instance.

    Raises:
        ValidationError: If the object could not be validated.
    """
    # Validate a new resource instance from the provided string object
    model = cls.Model.model_validate_strings(
        obj, strict=strict, context=context
    )

    # Return a new resource instance without triggering validation
    with validation_manager(mode='disabled'):
        return cls(model, context)

resource_validate_strings_many classmethod

resource_validate_strings_many(
    obj: Any,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]

Validate the given string object collection against the resource.

Parameters:

Name	Type	Description	Default
obj	Any	The string object collection to validate.	required
strict	bool | None	Whether to enforce types strictly.	None
context	dict[str, Any] | None	Extra variables to pass to the validator.	None

Returns:

Type	Description
Sequence[Resource]	A validated collection of resource instances.

Raises:

Type	Description
ValidationError	If the object collection could not be validated.

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

@classmethod
def resource_validate_strings_many(
    cls: type[Resource],
    obj: Any,
    *,
    strict: bool | None = None,
    context: dict[str, Any] | None = None,
) -> Sequence[Resource]:
    """Validate the given string object collection against the resource.

    Args:
        obj: The string object collection to validate.
        strict: Whether to enforce types strictly.
        context: Extra variables to pass to the validator.

    Returns:
        A validated collection of resource instances.

    Raises:
        ValidationError: If the object collection could not be validated.
    """
    # Tell pytest to hide this function from tracebacks
    __tracebackhide__ = True

    return cls.resource_adapter.validate_strings(  # type: ignore
        obj, strict=strict, context=context
    )

plateforme.core.mixins

This module provides resource mixin classes for the Plateforme framework.

Archivable

Archivable mixin for resource class.

FUTURE: Implement archivable mixin for resource class.

Auditable

Auditable mixin for resource class.

FIXME: Implement auditable mixin for resource class.

Encrypted

Encrypted mixin for resource class.

FUTURE: Implement encrypted mixin for resource class.