mafw.db.db_filter

The module provides the capability to filter model classes with information taken from the steering file.

Classes

`Filter`(name_, **kwargs)	Class to filter rows from a model.
`FilterCondition`(operation, value[, field_name])	Represents a single filter condition with operation and value(s).
`FilterRegister`([data])	A special dictionary to store all `Filters` in a processors.

class mafw.db.db_filter.Filter(name_: str, **kwargs: Any)[source]

Bases: object

Class to filter rows from a model.

The filter object can be used to generate a where clause to be applied to Model.select().

The construction of a Filter is normally done via a configuration file using the from_conf() class method. The name of the filter is playing a key role in this. If it follows a dot structure like:

ProcessorName.Filter.ModelName

then the corresponding table from the TOML configuration object will be used.

For each processor, there might be many Filters, up to one for each Model used to get the input list. If a processor is joining together three Models when performing the input select, there will be up to three Filters collaborating on making the selection.

The filter configuration can contain the following key, value pair:

key / string pairs, where the key is the name of a field in the corresponding Model

key / numeric pairs

key / arrays

key / dict pairs with ‘op’ and ‘value’ keys for explicit operation specification

All fields from the configuration file will be added to the instance namespace, thus accessible with the dot notation. Moreover, the field names and their filter value will be added to a private dictionary to simplify the generation of the filter SQL code.

The user can use the filter object to store selection criteria. He can construct queries using the filter contents in the same way as he could use processor parameters.

If he wants to automatically generate valid filtering expression, he can use the filter() method. In order for this to work, the Filter object be bound to a Model. Without this binding the Filter will not be able to automatically generate expressions.

For each field in the filter, one condition will be generated according to the following scheme:

Filter field type	Logical operation	Example
Numeric, boolean	==	Field == 3.14
String	GLOB	Field GLOB ‘*ree’
List	IN	Field IN [1, 2, 3]
Dict (explicit)	op from dict	Field BIT_AND 5

All conditions will be joined with a AND logic by default, but this can be changed.

Consider the following example:

class MeasModel(MAFwBaseModel):
    meas_id = AutoField(primary_key=True)
    sample_name = TextField()
    successful = BooleanField()
    flags = IntegerField()


# Traditional simplified usage
flt = Filter(
    'MyProcessor.Filter.MyModel',
    sample_name='sample_00*',
    meas_id=[1, 2, 3],
    successful=True,
)

# New explicit operation usage
flt = Filter(
    'MyProcessor.Filter.MyModel',
    sample_name={'op': 'LIKE', 'value': 'sample_00%'},
    flags={'op': 'BIT_AND', 'value': 5},
    meas_id={'op': 'IN', 'value': [1, 2, 3]},
)

flt.bind(MeasModel)
filtered_query = MeasModel.select().where(flt.filter())

The explicit operation format allows for bitwise operations and other advanced filtering.

TOML Configuration Examples:

[MyProcessor.Filter.MyModel]
sample_name = "sample_00*"  # Traditional GLOB
successful = true           # Traditional equality

# Explicit operations
flags = { op = "BIT_AND", value = 5 }
score = { op = ">=", value = 75.0 }
category = { op = "IN", value = ["A", "B", "C"] }
date_range = { op = "BETWEEN", value = ["2024-01-01", "2024-12-31"] }

Constructor parameters:

Parameters:

name_ (str) – The name of the filter. It should be in dotted format to facilitate the configuration via the steering file. The _ is used to allow the user to have a keyword argument named name.
kwargs – Keyword parameters corresponding to fields and filter values.

Changed in version 1.2.0: The parameter name has been renamed as name_.

Changed in version 1.3.0: From this version also explicit operations are implemented. When more conditions are provided, it is now possible to decide how to join them.

classmethod from_conf(name: str, conf: dict[str, Any], default: dict[str, Any] | None = None) → Self[source]

Builds a Filter object from a steering file dictionary.

If the name is in dotted notation, then this should be corresponding to the table in the configuration file. If a default configuration is provided, this will be used as a starting point for the filter, and it will be updated by the actual configuration in conf.

In normal use, you would provide the specific configuration via the conf parameter and the global filter configuration as default.

See details in the class documentation

Parameters:

name (str) – The name of the filter in dotted notation.
conf (dict) – The configuration dictionary.
default (dict) – Default configuration dictionary

Returns:

A Filter object

Return type:

Filter

static _create_condition_from_value(value: Any, field_name: str) → FilterCondition[source]

Create a FilterCondition based on value type (backward compatibility).

Parameters:

value – The filter value
field_name – The field name

Returns:

A FilterCondition

bind(model: type[Model] | None = None) → None[source]

Connects a filter to a Model class.

If no model is provided, the method will try to bind a class from with global dictionary with a name matching the model name used during the Filter configuration. It only works when the Model is defined as global.

Parameters:: model (Model, Optional) – Model to be bound. Defaults to None

filter(join_with: str = 'AND') → Expression | bool[source]

Generates a filtering expression joining all filtering fields.

See details in the class documentation

Changed in version 1.3.0: Add the possibility to specify a join_with function

Parameters:

join_with (str) – How to join conditions (‘AND’ or ‘OR’). Defaults to ‘AND’.

Returns:

The filtering expression.

Return type:

peewee.Expression | bool

Raises:

TypeError – when the field value type is not supported.
ValueError – when join_with is not ‘AND’ or ‘OR’.

get_condition(key: str) → FilterCondition[source]

Gets a filter condition by field name.

Added in version 1.3.0.

Parameters:: key – The field name
Returns:: The FilterCondition
Raises:: KeyError – if the field does not exist

get_field(key: str) → Any[source]

Gets a field by name.

Parameters:: key (str) – The name of the field.
Returns:: The value of the field.
Return type:: Any
Raises:: KeyError – if the requested field does not exist.

set_condition(key: str, operation: LogicalOp | str, value: Any) → None[source]

Sets a filter condition explicitly.

Added in version 1.3.0.

Parameters:

key – The field name
operation – The logical operation
value – The filter value

set_field(key: str, value: Any) → None[source]

Sets the value of a field by name

Parameters:

key (str) – The name of the field.
value (Any) – The value of the field.

property field_names: list[str]: The list of field names.

property is_bound: bool: Returns true if the Filter has been bound to a Model

class mafw.db.db_filter.FilterCondition(operation: LogicalOp | str, value: Any, field_name: str | None = None)[source]

Bases: object

Represents a single filter condition with operation and value(s).

This allows for explicit specification of logical operations beyond the automatic type-based determination.

Added in version 1.3.0: Before only implicit types of conditions were allowed.

Initialize a filter condition.

Parameters:

operation – The logical operation to apply
value – The value(s) to compare against
field_name – Optional field name for error reporting

to_expression(model_field: Field) → Expression[source]

Convert this condition to a Peewee expression.

Parameters:: model_field – The Peewee model field
Returns:: The expression

class mafw.db.db_filter.FilterRegister(data: dict[str, Filter] | None = None, /, **kwargs: Any)[source]

Bases: UserDict[str, Filter]

A special dictionary to store all Filters in a processors.

It contains a publicly accessible dictionary with the configuration of each Filter using the Model name as keyword.

It contains a private dictionary with the global filter configuration as well. The global filter is not directly accessible, but only some of its members will be exposed via properties. In particular, the new_only flag that is relevant only at the Processor level can be accessed directly using the new_only. If not specified in the configuration file, the new_only is by default True.

Constructor parameters:

Parameters:

data (dict) – Initial data
kwargs – Keywords arguments

bind_all(models: list[type[Model]] | dict[str, type[Model]]) → None[source]

Binds all filters to their models.

The models list or dictionary should contain a valid model for all the Filter in the registry. In the case of a dictionary, the key value should be the model name.

If the user provides a model for which there is no corresponding filter in the register, then a new filter for that model is created using the GlobalFilter default.

This can happen if the user did not provide a specific table for the Processor/Model, but simply put all filtering conditions in the GlobalFilter table. Even though, this behaviour is allowed and working, it may result in unexpected results. Also listing more than needed models in the input list can be dangerous because they will anyhow use the default filters.

Parameters:: models (list[type(Model)] | dict[str,type(Model)]) – List or dictionary of a databank of Models from which the Filter can be bound.

filter_all(join_with: str = 'AND')[source]

Generates a where clause joining all filters.

If one Filter is not bound, then True is returned.

Changed in version 1.3.0: Add the join_with parameter to decide if to AND or OR all filters in the register.

Parameters:: join_with (str) – How to join filters (‘AND’ or ‘OR’). Defaults to ‘AND’.

property new_only: bool

The new only flag.

Returns:: True, if only new items, not already in the output database table must be processed.
Return type:: bool