Pythonic Method of Querying Complex Nested Data Objects

Question

I'm a few weeks into my journey with Python. My background is more in C# and SQL. I'm working with a vendor API and for various reasons, I need to use Python. One aspect of Python that has been a challenge for me is not having LINQ. My understanding is that constructs such as list comprehensions and filters are the rough equivalents.

In this specific case, I need to query the response of the API to determine which escalation policies reference a target user. The data structure is fairly complex, but the relevant aspects for this question are that there is a list of escalation policies, each escalation policy has one or more escalation rules, each escalation rule has one or more targets that reference a user, and each user can be in any number of escalation rules including being in multiple escalation rules for the same escalation policy.

Stated in English, I need to find the escalation policies for which there is any escalation rule for which there is any target that references the the specified user.

To focus on the relevant details, I created the following test data set that only has the necessary fields:

from collections import namedtuple

EscalationPolicy = namedtuple('EscalationPolicy','Id EscalationRules')
EscalationRule = namedtuple('EscalationRule','Id Targets')
Target = namedtuple('Target', 'Id UserId')

escalation_policies = [
    EscalationPolicy('ep1_2',[
        EscalationRule('er1_2-1',[
            Target('t1','u1')
        ])
        ,EscalationRule('er1_2-2',[
            Target('t2','u2')
        ])
    ])
    ,EscalationPolicy('ep2_3',[
        EscalationRule('er2_3-1',[
            Target('t3,','u2')
        ])
        ,EscalationRule('er2_3-2',[
            Target('t4','u3')
        ])
    ])
    ,EscalationPolicy('ep1_3',[
        EscalationRule('er1_3-1',[
            Target('t5','u1')])
        ,EscalationRule('er1_3-2',[
            Target('t6','u3')
        ])
    ])
    ,EscalationPolicy('ep1_2_3',[
        EscalationRule('er1_2_3-1',[
            Target('t7','u1')
        ])
        ,EscalationRule('er1_2_3-2',[
            Target('t8','u2')
        ])
        ,EscalationRule('er1_2_3-3',[
            Target('t9','u3')
        ])
    ])
    ,EscalationPolicy('ep12',[
        EscalationRule('ep12-1',[
            Target('t10','u1')
            ,Target('t11','u2')
        ])
    ])
    ,EscalationPolicy('ep23',[
        EscalationRule('ep23-1',[ 
            Target('t12','u2')
            ,Target('t13','u3')])
    ])
    ,EscalationPolicy('ep12_23',[
        EscalationRule('ep12_23-1',[
            Target('t14','u1')
            ,Target('t15','u2')
        ])
        ,EscalationRule('ep12_23-2',[ 
            Target('t16','u2')
            ,Target('t17','u3')])
    ])
    ,EscalationPolicy('ep123',[
        EscalationRule('ep123-1',[
            Target('t18','u1')
            ,Target('t19','u2')
            ,Target('t20','u3')])
    ])         
]

Using the roughly equivalent data structure in C#, a workable query in LINQ would be:

var targetUserId = "u1";
var targetEscalationPolicies = escalationPolicies
   .Where(ep => ep.EscalationRules
      .Any(er => er.Targets.Select(t => t.UserId).Contains(targetUserId)))
   .ToList();

What I came up with in Python is:

target_user_id = 'u1'
target_escalation_policies = [ep for ep in escalation_policies if any(t.UserId == target_user_id for er in ep.EscalationRules for t in er.Targets)]

It works. My question is whether this a best practice and are there any other recommended approaches?

Thanks!

Answer 1

Here's an example of how you could leverage some very simplistic OOP to abstract away the logic of searching by user at the interface level:

from collections import defaultdict
from dataclasses import dataclass
from typing import List


@dataclass
class Target:
    target: str
    user: str


@dataclass
class EscalationRule:
    label: str
    targets: List[Target]


class EscalationPolicy:
    def __init__(self, label: str, rules: List[EscalationRule]) -> None:
        self.label = label
        self.rules = rules
        self.users_targeted = defaultdict(list)
        for rule in self.rules:
            self.update_users_targeted(rule)

    def add_rule(self, rule: EscalationRule) -> None:
        self.rules.append(rule)
        self.update_users_targeted(rule)

    def update_users_targeted(self, rule: EscalationRule) -> None:
        for target in rule.targets:
            self.users_targeted[target.user].append(rule)


class Procedure:
    def __init__(self, label: str, policies: List[EscalationPolicy]) -> None:
        self.label = label
        self.policies = policies
    
    def search_policies_by_user(self, user: str) -> List[EscalationRule]:
        rules = []
        for policy in self.policies:
            rules.extend(policy.users_targeted.get(user, []))
        return rules

Using your list, escalation_policies, I was able to do this:

In [3]: procedure = Procedure("procedure_1", escalation_policies)

In [4]: procedure.search_policies_by_user('u1')
Out[4]:
[EscalationRule(label='er1_2-1', targets=[Target(target='t1', user='u1')]),
 EscalationRule(label='er1_3-1', targets=[Target(target='t5', user='u1')]),
 EscalationRule(label='er1_2_3-1', targets=[Target(target='t7', user='u1')]),
 EscalationRule(label='ep12-1', targets=[Target(target='t10', user='u1'), Target(target='t11', user='u2')]),
 EscalationRule(label='ep12_23-1', targets=[Target(target='t14', user='u1'), Target(target='t15', user='u2')]),
 EscalationRule(label='ep123-1', targets=[Target(target='t18', user='u1'), Target(target='t19', user='u2'), Target(target='t20', user='u3')])]

Obviously this is a lot of boilerplate just to abstract away that somewhat gnarly list comprehension. This would only be a good approach if you had a lot more functionality you wanted to add (like adding escalation rules to existing policies, or adding new policies to a "procedure" -- I couldn't think of a better name for List[EscalationPolicy] ¯\_(?)_/¯ ).

The good news is that the interface didn't change at all. I was able to create these objects with your provided escalation_policies list, with no alterations.