mex.backend.merged package¶
Submodules¶
mex.backend.merged.helpers module¶
- mex.backend.merged.helpers._apply_additive_rule(merged_dict: dict[str, Any], mergeable_fields: list[str], rule: AdditiveAccessPlatform | AdditiveActivity | AdditiveContactPoint | AdditiveDistribution | AdditiveOrganization | AdditiveOrganizationalUnit | AdditivePerson | AdditivePrimarySource | AdditiveResource | AdditiveVariable | AdditiveVariableGroup) None ¶
Merge the values from an additive rule into a merged_dict.
- Parameters:
merged_dict – Mapping from field names to lists of merged values
mergeable_fields – List of mergeable field names
rule – Additive rule with values to be added
- mex.backend.merged.helpers._apply_subtractive_rule(merged_dict: dict[str, Any], mergeable_fields: list[str], rule: SubtractiveAccessPlatform | SubtractiveActivity | SubtractiveContactPoint | SubtractiveDistribution | SubtractiveOrganization | SubtractiveOrganizationalUnit | SubtractivePerson | SubtractivePrimarySource | SubtractiveResource | SubtractiveVariable | SubtractiveVariableGroup) None ¶
Prune values of a subtractive rule from a merged_dict.
- Parameters:
merged_dict – Mapping from field names to lists of merged values
mergeable_fields – List of mergeable field names
rule – Subtractive rule with values to remove
- mex.backend.merged.helpers._merge_extracted_items_and_apply_preventive_rule(merged_dict: dict[str, Any], mergeable_fields: list[str], extracted_items: list[ExtractedAccessPlatform | ExtractedActivity | ExtractedContactPoint | ExtractedDistribution | ExtractedOrganization | ExtractedOrganizationalUnit | ExtractedPerson | ExtractedPrimarySource | ExtractedResource | ExtractedVariable | ExtractedVariableGroup], rule: PreventiveAccessPlatform | PreventiveActivity | PreventiveContactPoint | PreventiveDistribution | PreventiveOrganization | PreventiveOrganizationalUnit | PreventivePerson | PreventivePrimarySource | PreventiveResource | PreventiveVariable | PreventiveVariableGroup | None) None ¶
Merge a list of extracted items while applying a preventive rule.
Collect unique values from the extracted items and write them into merged_dict, unless the primary source of the extracted item was blocked by the rule.
- Parameters:
merged_dict – Mapping from field names to lists of merged values
mergeable_fields – List of mergeable field names
extracted_items – List of extracted items
rule – Preventive rules with primary source identifiers, can be None
- mex.backend.merged.helpers.create_merged_item(identifier: Identifier, extracted_items: list[ExtractedAccessPlatform | ExtractedActivity | ExtractedContactPoint | ExtractedDistribution | ExtractedOrganization | ExtractedOrganizationalUnit | ExtractedPerson | ExtractedPrimarySource | ExtractedResource | ExtractedVariable | ExtractedVariableGroup], rule_set: AccessPlatformRuleSetRequest | ActivityRuleSetRequest | ContactPointRuleSetRequest | DistributionRuleSetRequest | OrganizationRuleSetRequest | OrganizationalUnitRuleSetRequest | PersonRuleSetRequest | PrimarySourceRuleSetRequest | ResourceRuleSetRequest | VariableRuleSetRequest | VariableGroupRuleSetRequest | AccessPlatformRuleSetResponse | ActivityRuleSetResponse | ContactPointRuleSetResponse | DistributionRuleSetResponse | OrganizationRuleSetResponse | OrganizationalUnitRuleSetResponse | PersonRuleSetResponse | PrimarySourceRuleSetResponse | ResourceRuleSetResponse | VariableRuleSetResponse | VariableGroupRuleSetResponse | None) MergedAccessPlatform | MergedActivity | MergedContactPoint | MergedDistribution | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedVariable | MergedVariableGroup ¶
Merge a list of extracted items with a set of rules.
- Parameters:
identifier – Identifier the finished merged item should have
extracted_items – List of extracted items, can be empty
rule_set – Rule set, with potentially empty rules
- Returns:
Instance of a merged item
- mex.backend.merged.helpers.search_merged_items_in_graph(query_string: str | None = None, stable_target_id: str | None = None, entity_type: list[str] | None = None, skip: int = 0, limit: int = 100) MergedItemSearch ¶
Search for merged items.
- Parameters:
query_string – Full text search query term
stable_target_id – Optional stable target ID filter
entity_type – Optional entity type filter
skip – How many items to skip for pagination
limit – How many items to return at most
- Returns:
MergedItemSearch instance
mex.backend.merged.main module¶
- mex.backend.merged.main.search_merged_items(q: Annotated[str, Query(PydanticUndefined)] = '', identifier: Identifier | None = None, entityType: Annotated[Sequence[MergedType], Query(PydanticUndefined)] = [], skip: Annotated[int, Query(PydanticUndefined)] = 0, limit: Annotated[int, Query(PydanticUndefined)] = 10) MergedItemSearch ¶
Search for merged items by query text or by type and id.
mex.backend.merged.models module¶
- class mex.backend.merged.models.MergedItemSearch(*, items: list[Annotated[MergedAccessPlatform | MergedActivity | MergedContactPoint | MergedDistribution | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedVariable | MergedVariableGroup, FieldInfo(annotation=NoneType, required=True, discriminator='entityType')]], total: int)¶
Bases:
BaseModel
Response body for the merged item search endpoint.
- items: list[Annotated[MergedAccessPlatform | MergedActivity | MergedContactPoint | MergedDistribution | MergedOrganization | MergedOrganizationalUnit | MergedPerson | MergedPrimarySource | MergedResource | MergedVariable | MergedVariableGroup, FieldInfo(annotation=NoneType, required=True, discriminator='entityType')]]¶
- model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_max_length': 100000, 'str_min_length': 1, 'str_strip_whitespace': True, 'use_enum_values': True, 'validate_assignment': True, 'validate_default': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_fields: ClassVar[Dict[str, FieldInfo]] = {'items': FieldInfo(annotation=list[Annotated[Union[MergedAccessPlatform, MergedActivity, MergedContactPoint, MergedDistribution, MergedOrganization, MergedOrganizationalUnit, MergedPerson, MergedPrimarySource, MergedResource, MergedVariable, MergedVariableGroup], FieldInfo(annotation=NoneType, required=True, discriminator='entityType')]], required=True), 'total': FieldInfo(annotation=int, required=True)}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- total: int¶