Python instead of jq

Table of Contents

jq

To query JSON file like XML with XPath or XQuery, you can use a command line tool called jq. It's great but you can use Python also.

Python querying JSON

You can load JSON file very easy:

import json
with open(path_to_json_file, 'rt') as f:
    json_data = json.load(f)

but more interesting is that you can query JSON as some DB. The idea of examples below is not to achieve the best performance - it is not the goal - just demonstration of short queries looking as jq, but where the performance is very bad, there are alternative queries, see carefully.

Considering our decoded JSON in this way:

dat = {
    'project': {
        'name': 'Demo',
        'config': {
            'databases': [
                {
                    'name': 'DB1',
                    'primary': {
                        'host': '10.0.0.5',
                        'port': 5430
                    },
                    'replicas': [
                        {
                            'host': '10.0.0.5',
                            'port': 5431
                        },
                        {
                            'host': '10.0.0.5',
                            'port': 5437
                        }
                    ]
                },
                {
                    'name': 'DB2',
                    'primary': {
                        'host': '10.0.0.12',
                        'port': 5433
                    },
                    'replicas': [
                        {
                            'host': '10.0.0.15',
                            'port': 5434
                        }
                    ]
                },
                {
                    'name': 'DB3',
                    'primary': {
                        'host': '10.0.0.5',
                        'port': 5437
                    }
                }                
            ]
        }
    }
}

and let's try some queries:

databases = dat.get('project',{}).get('config',{}).get('databases',[])
q = len(databases)
print(f'NUMBER OF DATABASES: {q}')

q = {d.get('name', 'Unnamed') for d in databases}
print(f'DATABASES: {q}')

q = {d.get('primary', {}).get('port') for d in databases} \
  | {r.get('port') for d in databases for r in d.get('replicas',[])}
print(f'PORTS IN USE (WAY 1): {q}')

q = { *(d.get('primary', {}).get('port') for d in databases),
      *(r.get('port') for d in databases for r in d.get('replicas',[])) }
print(f'PORTS IN USE (WAY 2): {q}')

q = set( [d.get('primary', {}).get('port') for d in databases] +
         [r.get('port') for d in databases for r in d.get('replicas',[])])
print(f'PORTS IN USE (WAY 3): {q}')

q = { (d.get('name', 'Unnamed'), p) for d in databases
      if (p:=d.get('primary', {}).get('port')) and p > 5430 }
print(f'DATABASE NAMES WITH PORTS WHERE PRIMARY PORT > 5430: {q}')
NUMBER OF DATABASES: 3
DATABASES: {'DB3', 'DB2', 'DB1'}
PORTS IN USE (WAY 1): {5430, 5431, 5433, 5434, 5437}
PORTS IN USE (WAY 2): {5430, 5431, 5433, 5434, 5437}
PORTS IN USE (WAY 3): {5430, 5431, 5433, 5434, 5437}
DATABASE NAMES WITH PORTS WHERE PRIMARY PORT > 5430: {('DB2', 5433), ('DB3', 5437)}

Tricks: union of sets with |, unpacking inside a set with * (star unpacking).

The last query (DATABASE NAMES WITH PORTS WHERE PRIMARY PORT > 5430) is this simple set:

\[ \{name'(d) \mid d \in databases, \exists primary(d), \exists port(primary(d)), port > 5439 \} \\ \]

\begin{aligned} name'(d) &= d \mapsto name(d), &\text{if}\ \exists name(d) \\ &= d \mapsto \text{Unnamed}, &\text{else} \end{aligned}

You see the idea - we can use list/set/dict comprehension and dict.get(KEY, DEFAULT) method (as well as different tricks). Let's do something more complex - to query conflicting ports if they exist - they should be on the same host:

from collections import defaultdict

dd = defaultdict(set)

for k, v in (
    [((ph, pp), d.get('name', f'Unnamed{i}'))
     for i, d in enumerate(databases)
     if (prim := d.get('primary', {}))
     and (ph := prim.get('host'))
     and (pp := prim.get('port'))]
    +
    [((rh, rp), d.get('name', f'Unnamed{i}'))
     for i, d in enumerate(databases)
     for repl in d.get('replicas', [])
     if (rh := repl.get('host'))
     and (rp := repl.get('port'))]
): dd[k].add(v)
conflicts = {k:v for k,v in dd.items() if len(v) > 1}
print(f'DATABASES HAVING PORT CONFLICTS (WAY 1): {conflicts}')
DATABASES HAVING PORT CONFLICTS (WAY 1): {('10.0.0.5', 5437): {'DB3', 'DB1'}}

Tricks:

  • nested "for"-s
  • assignment in "if"-s like if (prim := ...)
  • fallback to default values: d.get(.., DEFAULT)
    • we can form even long getters: d.get('dict1', {}).get('dic2',{}).get('arrays', [])
  • use of defaultdict for grouping (GROUP BY in SQL): if you need order and duplicates, use list, else set, but you can you even int
  • + similarly to ∪ in math (sets union), well, for lists, in the case
  • f'Unnamed{i}' numbers unnamed databases, giving them unique names, but be careful: if 2 +-ed lists are not isomorphic, you will hit problems.

Correspondence to mathematics

The Python's query corresponds to the sets:

\begin{gather*} Q_p = \{ \langle \langle h, p \rangle, n \rangle \mid d \in databases, prim = primary'(d), \exists h = host(prim), \exists p = port(prim), n = name'(d) \} \\ \\ Q_r = \{ \langle \langle h, p \rangle, n \rangle \mid d \in databases, repl = replicas'(d), \exists h = host(repl), \exists p = port(repl), n = name'(d) \} \\ \\ Q = Q_p \cup Q_r \end{gather*} \begin{aligned} primary'(d) &= d \mapsto primary(d), &\text{if}\ \exists primary(d) \\ &= d \mapsto \varnothing, &\text{else} \end{aligned} \begin{gather*} G(k): \langle h, p \rangle \mapsto \{ n \mid \langle \langle h, p \rangle, n \rangle \in Q \} \\ \\ Conflicts = \{ \langle k, G(k) \rangle \mid |G(k)| > 1 \} \end{gather*}

where G(k) is grouping function, actully it is our defaultdict!

But this is not so good query: we pass through databases 2 times. Lets try to rewrite it:

import pprint
dd = defaultdict(set)
q = [dd[rec[:2]].add(rec[2])
     for db in [[(ph, pp, name), (rh, rp, name)]
                for i, d in enumerate(databases)
                for name in [d.get('name', f'Unnamed{i}')]
                for prim in [d.get('primary', {})]
                for ph in [prim.get('host')]
                for pp in [prim.get('port')]
                for repl in d.get('replicas', [{}])  # <-- ATTENTION: [{}] else nested "for"-s stop!
                for rh in [repl.get('host')]
                for rp in [repl.get('port')]]
     for rec in db if rec[0] and rec[1]]
conflicts = {k:v for k,v in dd.items() if len(v) > 1}
print(f'DATABASES HAVING PORT CONFLICTS (WAY 2): {conflicts}')
DATABASES HAVING PORT CONFLICTS (WAY 2): {('10.0.0.5', 5437): {'DB3', 'DB1'}}

"ATTENTION" comment is important and Haskell programmers undertand it better: it allows nested "for"-s to work, repl will exists even if no replicas for the current database. It's OK bcs other "for"-s are also just bindings of rh, rp (if no replicas, then we will get these bidnings as None - they are missing in {} empty dict got from [{}]). But it will allow us to get a tuple with name.

As you can see it looks mostly as Haskell's do-notation for list monad (see example)!

Though, it is not optimal: it iterates over databases only once, but it recreates the same tuples. This can be fixed but only partially - with for t1 in [(ph,pp,name)]. If you need performance then you should switch to block-wise code but it looks not so LINQ-ish:

for db in databases:
    t1 = ...      # create once!
    ...
    for repl in db['replicas']:
        t2 = ...  # and this one!
        ...

Another examples:

q = [d.get('name', 'Unnamed') for d in databases if d.get('replicas', []) == []]
print(f'DATABASE NAMES WITHOUT REPLICAS: {q}')

q = sum(len(rs) for d in databases for rs in [d.get('replicas', [])]) / len(databases)
print(f'AVERAGE NUMBER OF REPLICAS PER DATABASE: {q}')
DATABASE NAMES WITHOUT REPLICAS: ['DB3']
AVERAGE NUMBER OF REPLICAS PER DATABASE: 1.0

The last example demonstrates simple aggregation (AVG). for rs in [d.get()] is just a trick to bind the list to rs and it can be written more directly.

Example from Baeldung (Wikipedia JSON): we will form new dictionaries with a title and page description (called extract) nameing them in our own way:

import pprint

dat2 = {
  'query': {
    'pages': [
      {
        '21721040': {
          'pageid': 21721040,
          'ns': 0,
          'title': 'Stack Overflow',
          'extract': 'Some interesting text about Stack Overflow'
        }
      },
      {
        '21721041': {
          'pageid': 21721041,
          'ns': 0,
          'title': 'Baeldung',
          'extract': 'A great place to learn about Java'
        }
      },
      {
        '11111111': {
          'pageid': 11111111,
          'ns': 0,
          'title': 'FAKE'
        }
      }      
    ]
  }
}
q = [{'page_title':t, 'page_description': p.get('extract', '<no description>')}
     for pages_dict in dat2.get('query',{}).get('pages',[])
     if (p := next(iter(pages_dict.values()), {}))  # take the 1st value (we see only one always)
     if (t := p.get('title'))]
pprint.pprint(q, sort_dicts=False)
[{'page_title': 'Stack Overflow',
  'page_description': 'Some interesting text about Stack Overflow'},
 {'page_title': 'Baeldung',
  'page_description': 'A great place to learn about Java'},
 {'page_title': 'FAKE', 'page_description': '<no description>'}]

Pay attention, that with if (x := ...) we bind but we will skip such iteration if x is false in some way! But in the case of extract key we fallback to "<no description>".

More complex queries

Lets try more complex queries. Our next JSON is

dict            dict          key
------------    ------------  -----------------
Users        🡒  Teams         team_id
Teams        🡒  Users         lead_id
Teams        🡒  Projects      project_ids
Projects     🡒  Repositories  repository_id
Repositories 🡒  Branches      branch_ids
Branches     🡒  Branches      parent_id, a tree
Users        🡒  Tasks         task_ids
Tasks        🡒  Users         assignee_id
Tasks        🡒  Projects      project_id
Tasks        🡒  Tasks         depends_on
cdat = {
    "users": [
        {
            "id": "U1",
            "name": "Alice",
            "team_id": "T1",
            "task_ids": ["K1", "K3"]
        },
        {
            "id": "U2",
            "name": "Bob",
            "team_id": "T1",
            "task_ids": ["K2"]
        },
        {
            "id": "U3",
            "name": "Carol",
            "team_id": "T2",
            "task_ids": ["K4"]
        }
    ],

    "teams": [
        {
            "id": "T1",
            "name": "Backend",
            "lead_id": "U1",
            "project_ids": ["P1"]
        },
        {
            "id": "T2",
            "name": "Frontend",
            "lead_id": "U3",
            "project_ids": ["P1", "P2"]
        }
    ],

    "projects": [
        {
            "id": "P1",
            "name": "Inventory",
            "repository_id": "R1"
        },
        {
            "id": "P2",
            "name": "Website",
            "repository_id": "R2"
        }
    ],

    "repositories": [
        {
            "id": "R1",
            "url": "git@example.com:inventory.git",
            "branch_ids": ["B1", "B2"]
        },
        {
            "id": "R2",
            "url": "git@example.com:website.git",
            "branch_ids": ["B3"]
        }
    ],

    "branches": [
        {
            "id": "B1",
            "name": "main",
            "parent_id": None
        },
        {
            "id": "B2",
            "name": "feature/login",
            "parent_id": "B1"
        },
        {
            "id": "B3",
            "name": "main",
            "parent_id": None
        }
    ],

    "tasks": [
        {
            "id": "K1",
            "title": "Design schema",
            "project_id": "P1",
            "assignee_id": "U1",
            "depends_on": []
        },
        {
            "id": "K2",
            "title": "REST API",
            "project_id": "P1",
            "assignee_id": "U2",
            "depends_on": ["K1"]
        },
        {
            "id": "K3",
            "title": "Authentication",
            "project_id": "P2",
            "assignee_id": "U1",
            "depends_on": ["K2"]
        },
        {
            "id": "K4",
            "title": "Landing page",
            "project_id": "P2",
            "assignee_id": "U3",
            "depends_on": []
        }
    ]
}

Find all tasks assigned to members of a given team

This is classical cross join:

def query(team_name):
    return [kid
            for t in cdat.get('teams',[])
            if (tid := t.get('id'))
            if (tn := t.get('name')) and tn == team_name
            for u in cdat.get('users',[])
            if (uid := u.get('id'))
            if (uti := u.get('team_id')) and uti == tid
            for k in cdat.get('tasks',[])
            if (kid := k.get('id'))
            if (aid := k.get('assignee_id')) and aid == uid]
print('TASKS OF BACKEND TEAM:', query('Backend'))
print('TASKS OF FRONTEND TEAM:', query('Frontend'))
TASKS OF BACKEND TEAM: ['K1', 'K3', 'K2']
TASKS OF FRONTEND TEAM: ['K4']

Detect dangling references (IDs that don't exist)

import pprint
q = ({'DANGLED PROJECT_ID IN TASKS':
      {kpid for k in cdat.get('tasks',[])
       if (kpid := k.get('project_id'))}
      -
      {p.get('id') for p in cdat.get('projects',[])}}
     |
     {'DANGLED ASSIGNEE_ID IN TASKS':
      {aid for k in cdat.get('tasks',[])
       if (aid := k.get('assignee_id'))}
      -
      {u.get('id') for u in cdat.get('users',[])}}
     |
     {'DANGLED DEPENDS_ON IN TASKS':
      {kdon for k in cdat.get('tasks',[])
       for kdon in k.get('depends_on',[])}
      -
      {k.get('id') for k in cdat.get('tasks',[])}}
     |
     {'DANGLED PARENT_ID IN BRANCHES':
      {bpid for b in cdat.get('branches',[])
       if (bpid := b.get('parent_id'))}
      -
      {b.get('id') for b in cdat.get('branches',[])}}
     |
     {'DANGLED BRANCH_IDS IN REPOSITORIES':
      {rbid for r in cdat.get('repositories',[])
       for rbid in r.get('branch_ids',[])}
      -
      {b.get('id') for b in cdat.get('branches',[])}}
     |
     {'DANGLED REPOSITORY_ID IN PROJECTS':
      {prid for p in cdat.get('projects',[])
       if (prid := p.get('repository_id'))}
      -
      {r.get('id') for r in cdat.get('repositories',[])}}
     |
     {'DANGLED PROJECT_IDS IN TEAMS':
      {tpid for t in cdat.get('teams',[])
       for tpid in t.get('project_ids',[])}
      -
      {p.get('id') for p in cdat.get('projects',[])}}
     |
     {'DANGLED LEAD_ID IN TEAMS':
      {tlid for t in cdat.get('teams',[])
       if (tlid := t.get('lead_id'))}
      -
      {u.get('id') for u in cdat.get('users',[])}}
     |
     {'DANGLED TASK_IDS IN USERS':
      {utid for u in cdat.get('users',[])
       for utid in u.get('task_ids',[])}
      -
      {k.get('id') for k in cdat.get('tasks',[])}
     }
     |
     {'DANGLED TEAM_ID IN USERS':
      {utid for u in cdat.get('users',[])
       if (utid := u.get('team_id'))}
      -
      {t.get('id') for t in cdat.get('teams',[])}}
    )
pprint.pprint(q, sort_dicts=False)
{'DANGLED PROJECT_ID IN TASKS': set(),
 'DANGLED ASSIGNEE_ID IN TASKS': set(),
 'DANGLED DEPENDS_ON IN TASKS': set(),
 'DANGLED PARENT_ID IN BRANCHES': set(),
 'DANGLED BRANCH_IDS IN REPOSITORIES': set(),
 'DANGLED REPOSITORY_ID IN PROJECTS': set(),
 'DANGLED PROJECT_IDS IN TEAMS': set(),
 'DANGLED LEAD_ID IN TEAMS': set(),
 'DANGLED TASK_IDS IN USERS': set(),
 'DANGLED TEAM_ID IN USERS': set()}

Tricks: merge of dictionaries with |, substraction of sets with -.

Find tasks that no one depends on

q = ({k.get('id') for k in cdat.get('tasks',[])}
     -
     {kdon for k in cdat.get('tasks',[])
     for kdon in k.get('depends_on',[])})
print(f'TASKS THAT NO ONE DEPENDS ON: {q}')
TASKS THAT NO ONE DEPENDS ON: {'K4', 'K3'}

It's simple: ALL \ DEPENDENCIES.

Find projects shared by multiple teams

from functools import *
from itertools import *

q = {pname for s in starmap(
    set.intersection,
    permutations((set(m.get('project_ids',[])) for m in cdat.get('teams',[])),
                 2))
     for e in s
     for p in cdat.get('projects',[])
     if (pname := p.get('name')) and p.get('id') == e}
print(q)
{'Inventory'}

It's a straightforward solution:

                      __________permutations_________
{P1}, {P1,P2} 🡒 ⦻ 🡒 ({P1}, {P1,P2}), ({P1,P2}, {P1}) 🡒 ⋂ 🡒 {P1}, {P1} 🡒 {P1}

{P1} 🡒 {Inventory}

It works, but this {P1}, {P1,P2} -> ⦻ is a problem: a lot of permutations.

We can use combinations() function, and it will cut it in half. But better is to use Counter class:

from collections import Counter

cnt = Counter(pid for t in cdat.get('teams',[])
              for pid in t.get('project_ids',[]))
q = {pname for p in cdat.get('projects',[])
     if cnt[p['id']] > 1 and (pname := p.get('name'))}
print(q)
{'Inventory'}

This is the best solution: it just counts all unique items from TEAMS.PROJECT_IDS, ie, key => integer-counter. Then we select projects which ID is "registered" within cnt counter as a value greater than 1.

The next queries are recursive, the alternative is to make them iterative which is better, because the stack depth is limited, but they will look lesser readable.

Detect dependency cycles

Let's check it with a result from cdat (q1) and other pre-setup results explicitly containing cycles (q2,3,4):

q1 = [(t.get('id'),d)
     for t in cdat.get('tasks',[])
     for d in t.get('depends_on',[])]
q2 = [('K1', 'K2'),
      ('K1', 'K3'),
      ('K2', 'K4'),
      ('K3', 'K1')]
q3 = [('K1', 'K3'),
      ('K1', 'K2'),
      ('K2', 'K1'),
      ('K3', 'K4')]
q4 = [('K1','K4'),
      ('K1','K2'),
      ('K2','K5'),
      ('K2','K3'),
      ('K3','K6'),
      ('K3','K1')]
def cycle1(q, d=None, passed=None):
    return (next((l for d in q
                  if (l := cycle1(q, d, []))), None)
            if d is None
            else (passed
                  if d[1] in (p[0] for p in passed)
                  else next((r for n in filter(lambda x:d[1]==x[0], q)
                             if (r := cycle1(q, n, passed + [d,n]))), None)))
for q in (q1, q2, q3, q4):
    print('Q:', q, '\n   CYCLES:', cycle1(q))
Q: [('K2', 'K1'), ('K3', 'K2')] 
   CYCLES: None
Q: [('K1', 'K2'), ('K1', 'K3'), ('K2', 'K4'), ('K3', 'K1')] 
   CYCLES: [('K1', 'K3'), ('K3', 'K1')]
Q: [('K1', 'K3'), ('K1', 'K2'), ('K2', 'K1'), ('K3', 'K4')] 
   CYCLES: [('K1', 'K2'), ('K2', 'K1')]
Q: [('K1', 'K4'), ('K1', 'K2'), ('K2', 'K5'), ('K2', 'K3'), ('K3', 'K6'), ('K3', 'K1')] 
   CYCLES: [('K1', 'K2'), ('K2', 'K3'), ('K2', 'K3'), ('K3', 'K1')]

So, no cycles in the original cdat. We needed function because we needed recursion. But cycle1() allows duplicates. Also maybe it is not so readable, so another version is cycle2():

def cycle2(q, d=None, passed=None):
    if d is None:
        for d in q:
            if l := cycle2(q, d, []):
                return l
    else:
        fr,to = d
        if to in (p[0] for p in passed):
            return passed
        else:
            for n in filter(lambda d:to==d[0], q):
                if r := cycle2(q, n, passed + ([n] if passed and passed[-1]==d else [d,n])):
                    return r
for q in (q1, q2, q3, q4):
    print('Q:', q, '\n   CYCLES:', ',  '.join(map(' -> '.join, rs) if (rs:=cycle2(q)) else ['<none>']))
Q: [('K2', 'K1'), ('K3', 'K2')] 
   CYCLES: <none>
Q: [('K1', 'K2'), ('K1', 'K3'), ('K2', 'K4'), ('K3', 'K1')] 
   CYCLES: K1 -> K3,  K3 -> K1
Q: [('K1', 'K3'), ('K1', 'K2'), ('K2', 'K1'), ('K3', 'K4')] 
   CYCLES: K1 -> K2,  K2 -> K1
Q: [('K1', 'K4'), ('K1', 'K2'), ('K2', 'K5'), ('K2', 'K3'), ('K3', 'K6'), ('K3', 'K1')] 
   CYCLES: K1 -> K2,  K2 -> K3,  K3 -> K1

Tricks: the code -

...
        for d in q:
            if l := cycle2(q, d, []):
                return l

can be represented as next((l for d in q if (l := cycle1(q, d, []))), None) but it is lesser readable: all queries above can be rewritten with explicit block-wise code.

To be honest, recursive queries don't match the initial idea to be more jq-ish...

Linear algerba (matrices) allows also to determine if a graph has a cycle but without to define the path of it.

Compute the dependency depth of each task

q1 = [(t.get('id'),d)
     for t in cdat.get('tasks',[])
     for d in t.get('depends_on',[])]
q2 = [('K1', 'K2'),
      ('K1', 'K3'),
      ('K2', 'K4'),
      ('K3', 'K1')]
q3 = [('K1', 'K3'),
      ('K1', 'K2'),
      ('K2', 'K1'),
      ('K3', 'K4')]
q4 = [('K1','K4'),
      ('K1','K2'),
      ('K2','K5'),
      ('K2','K3'),
      ('K3','K6'),
      ('K3','K1')]

def depth(q, root=None, dest=None, nodes=None):
    if root is None:
        nodes = defaultdict(set)
        for root, dest in q:
            depth(q, root, dest, nodes)
        return dict(nodes)
    else:
        if root != dest:
            nodes[root].add(dest)
        for successor in {n[1] for n in q if n[0]==dest}:
            if successor not in nodes[root]:
                depth(q, root, successor, nodes)
    
for q in (q1, q2, q3, q4):
    print('Q:', q, '\nTASK DEPS:', depth(q))
Q: [('K2', 'K1'), ('K3', 'K2')] 
TASK DEPS: {'K2': {'K1'}, 'K3': {'K2', 'K1'}}
Q: [('K1', 'K2'), ('K1', 'K3'), ('K2', 'K4'), ('K3', 'K1')] 
TASK DEPS: {'K1': {'K4', 'K2', 'K3'}, 'K2': {'K4'}, 'K3': {'K4', 'K2', 'K1'}}
Q: [('K1', 'K3'), ('K1', 'K2'), ('K2', 'K1'), ('K3', 'K4')] 
TASK DEPS: {'K1': {'K4', 'K2', 'K3'}, 'K2': {'K4', 'K1', 'K3'}, 'K3': {'K4'}}
Q: [('K1', 'K4'), ('K1', 'K2'), ('K2', 'K5'), ('K2', 'K3'), ('K3', 'K6'), ('K3', 'K1')] 
TASK DEPS: {'K1': {'K4', 'K2', 'K5', 'K6', 'K3'}, 'K2': {'K4', 'K5', 'K1', 'K6', 'K3'}, 'K3': {'K4', 'K5', 'K2', 'K1', 'K6'}}

So, the depth is obvious - it is the size of dependency sets. This is not the most optimal due to this in {n[1] for n in q if n[0]==dest}.

But we can rewrite it even in iterative way, so no Python stack limit - we will test it with more complex examples:

from collections import defaultdict

p1 = [('K1', 'K2'),
      ('K3', 'K4')]
p2 = [('K1', 'K2'),
      ('K1', 'K3'),
      ('K2', 'K4'),
      ('K3', 'K1')]
p3 = [('K1', 'K3'),
      ('K1', 'K2'),
      ('K2', 'K1'),
      ('K3', 'K4')]
p4 = [('K1', 'K4'),
      ('K1', 'K2'),
      ('K2', 'K5'),
      ('K2', 'K3'),
      ('K3', 'K6'),
      ('K3', 'K1')]
p5 = [('A', 'B'),
      ('A', 'C'),
      ('B', 'D'),
      ('C', 'D'),
      ('D', 'E')]
p6 = [('A', 'B'),
      ('B', 'C'),
      ('D', 'E')]

def iter_depth(q):
    if not q:
        return []
    deps = defaultdict(set)
    # `route` is a stack for kind of "backtracking" (return from DFS, Depth-First-Search):
    route = []
    look_for = q[0][0]
    # `q2` is the `q` value to be restored after return from look for `dst` (DFS);
    # the stack `route` is used as such stack: a tuple with the current `look_for` and an integer -
    # the position in `q`, - before to dive to look for `dst`:
    q2 = q
    # the current position in `q` (needed for restoring after return from DFS)
    i = 0
    while 1:
        for pair in q2:
            src, dst = pair
            if (src == look_for and
                (not route or dst not in deps[route[0][0]])):
                # found desired src but with avoid of loops - `... and (not route ...)`, ie,
                # we did not hit the initial `look_for`:
                route.append((look_for, i+1))
                deps[look_for].add(dst)
                # Here we can add `dst` which is = initial `look_for`. It is not bad, bcs
                # it allows to see loop later. Otherwise, use `if dst != route[0][0]: ...`
                deps[route[0][0]].add(dst)
                look_for = dst
                break
            i += 1
        else:  # "if" in "for" was not hit, "for" completed
            if route:
                look_for, i = route.pop()
                q2 = q[i:]  # kind of backtracking
            else:
                if q2 and (q2 := [p for p in q2 if p[0] not in deps]):  # or `...in q`
                    # `q2` now - independent pairs, not children of the initial `look_for`,
                    # if there are any yet - continue DFS with another, independent `look_for`:
                    look_for = q2[0][0]
                    continue
                else:
                    # all pair heads in `q2` is already in `deps`, nothing more to search:
                    break
    return dict(deps)
for q in (p1, p2, p3, p4, p5, p6):
    print('Q:', q, '\n   TASK DEPS:', iter_depth(q))
Q: [('K1', 'K2'), ('K3', 'K4')] 
   TASK DEPS: {'K1': {'K2'}, 'K3': {'K4'}}
Q: [('K1', 'K2'), ('K1', 'K3'), ('K2', 'K4'), ('K3', 'K1')] 
   TASK DEPS: {'K1': {'K4', 'K2', 'K1', 'K3'}, 'K2': {'K4'}, 'K3': {'K1'}}
Q: [('K1', 'K3'), ('K1', 'K2'), ('K2', 'K1'), ('K3', 'K4')] 
   TASK DEPS: {'K1': {'K4', 'K2', 'K1', 'K3'}, 'K3': {'K4'}, 'K2': {'K1'}}
Q: [('K1', 'K4'), ('K1', 'K2'), ('K2', 'K5'), ('K2', 'K3'), ('K3', 'K6'), ('K3', 'K1')] 
   TASK DEPS: {'K1': {'K4', 'K2', 'K5', 'K1', 'K6', 'K3'}, 'K2': {'K5', 'K3'}, 'K3': {'K1', 'K6'}}
Q: [('A', 'B'), ('A', 'C'), ('B', 'D'), ('C', 'D'), ('D', 'E')] 
   TASK DEPS: {'A': {'C', 'D', 'E', 'B'}, 'B': {'D'}, 'D': {'E'}, 'C': {'D'}}
Q: [('A', 'B'), ('B', 'C'), ('D', 'E')] 
   TASK DEPS: {'A': {'C', 'B'}, 'B': {'C'}, 'D': {'E'}}

So, queries required recursive search (even implemented in iterative way) don't look as simple jq queries. It can be overcome if to use helper-functions: you wrute them then load this module and use them from it. For example the function iter_depth is DFS common function and works with any list of tuples (source, destination). Python has a lot of ready for use libraries, for search too, just look at https://pypi.org/ ! See for example, NetworkX.

Example of usage of NetworkX library, pay attention: it does not include loops, so you don't see something like {'K1': {..., 'K1'}}:

from networkx import *

p1 = [('K1', 'K2'),
      ('K3', 'K4')]
p2 = [('K1', 'K2'),
      ('K1', 'K3'),
      ('K2', 'K4'),
      ('K3', 'K1')]
p3 = [('K1', 'K3'),
      ('K1', 'K2'),
      ('K2', 'K1'),
      ('K3', 'K4')]
p4 = [('K1', 'K4'),
      ('K1', 'K2'),
      ('K2', 'K5'),
      ('K2', 'K3'),
      ('K3', 'K6'),
      ('K3', 'K1')]
p5 = [('A', 'B'),
      ('A', 'C'),
      ('B', 'D'),
      ('C', 'D'),
      ('D', 'E')]
p6 = [('A', 'B'),
      ('B', 'C'),
      ('D', 'E')]

for q in (p1,p2,p3,p4,p5,p6):
    dg = DiGraph()
    dg.add_edges_from(q)
    deps = {n:ds  for n in dg.nodes
            if (ds := descendants(dg, n))}
    print('Q:', q, '\n   TASK DEPS:', deps)
Q: [('K1', 'K2'), ('K3', 'K4')] 
   TASK DEPS: {'K1': {'K2'}, 'K3': {'K4'}}
Q: [('K1', 'K2'), ('K1', 'K3'), ('K2', 'K4'), ('K3', 'K1')] 
   TASK DEPS: {'K1': {'K4', 'K2', 'K3'}, 'K2': {'K4'}, 'K3': {'K4', 'K2', 'K1'}}
Q: [('K1', 'K3'), ('K1', 'K2'), ('K2', 'K1'), ('K3', 'K4')] 
   TASK DEPS: {'K1': {'K4', 'K2', 'K3'}, 'K3': {'K4'}, 'K2': {'K4', 'K1', 'K3'}}
Q: [('K1', 'K4'), ('K1', 'K2'), ('K2', 'K5'), ('K2', 'K3'), ('K3', 'K6'), ('K3', 'K1')] 
   TASK DEPS: {'K1': {'K4', 'K2', 'K5', 'K6', 'K3'}, 'K2': {'K4', 'K5', 'K1', 'K6', 'K3'}, 'K3': {'K4', 'K5', 'K2', 'K1', 'K6'}}
Q: [('A', 'B'), ('A', 'C'), ('B', 'D'), ('C', 'D'), ('D', 'E')] 
   TASK DEPS: {'A': {'C', 'D', 'E', 'B'}, 'B': {'D', 'E'}, 'C': {'D', 'E'}, 'D': {'E'}}
Q: [('A', 'B'), ('B', 'C'), ('D', 'E')] 
   TASK DEPS: {'A': {'C', 'B'}, 'B': {'C'}, 'D': {'E'}}

Find repositories used by teams led by a particular user

def query(user_name):
    q = {(r.get('id'), r.get('url', '<no url>'))
         for u in cdat.get('users',[])
         if ((u.get('name') == user_name) and
             (utid := u.get('team_id')))
         for t in cdat.get('teams',[])
         if t.get('id') == utid and u.get('id') == t.get('lead_id')
         if (tpids := t.get('project_ids',[]))
         for p in cdat.get('projects',[])
         if (p.get('id') in tpids and (prid := p.get('repository_id')))
         for r in cdat.get('repositories',[])
         if r.get('id') == prid}
    return q
print('REPOSITORIES USED BY TEAM LED BY ALICE:\n  ', query('Alice'))
print('REPOSITORIES USED BY TEAM LED BY CAROL:\n  ', query('Carol'))
print('REPOSITORIES USED BY TEAM LED BY BOB:\n  ', query('Bob'))
REPOSITORIES USED BY TEAM LED BY ALICE:
   {('R1', 'git@example.com:inventory.git')}
REPOSITORIES USED BY TEAM LED BY CAROL:
   {('R1', 'git@example.com:inventory.git'), ('R2', 'git@example.com:website.git')}
REPOSITORIES USED BY TEAM LED BY BOB:
   set()

Count how many tasks belong to each team

c = Counter(uti for k in cdat.get('tasks',[])
            if (kaid := k.get('assignee_id'))
            for u in cdat.get('users',[])
            if (u.get('id') == kaid and
                (uti := u.get('team_id'))))
q = {tn:c[t.get('id')] for t in cdat.get('teams',[])
     if (tn := t.get('name'))}
pprint.pprint(q)
{'Backend': 3, 'Frontend': 1}

Counter was used again, it worked up like SUM() with GROUP BY in SQL.

Determine which user has the most transitive task dependencies

q = [(t.get('id'), d)
     for t in cdat.get('tasks',[])
     for d in t.get('depends_on',[])]
q1 = iter_depth(q)
q2 = defaultdict(int)
for u,c in ((u.get('name'), 1 + len(q1.get(t.get('id'),[]))) # look at `1 + ...`
            for u in cdat.get('users',[])
            for t in cdat.get('tasks',[])
            if u.get('id') == t.get('assignee_id')):
    q2[u] += c
print('TASKS WITH DIRECT DEPS:', q)
print('TASKS WITH DEPS IN DEPTH:', q1)
print('USERS AND TASKS NUMBER:', q2)
print('USER WITH THE MOST TASKS:', max(q2, key=q2.get))
TASKS WITH DIRECT DEPS: [('K2', 'K1'), ('K3', 'K2')]
TASKS WITH DEPS IN DEPTH: {'K2': {'K1'}, 'K3': {'K2'}}
USERS AND TASKS NUMBER: defaultdict(<class 'int'>, {'Alice': 3, 'Bob': 2, 'Carol': 1})
USER WITH THE MOST TASKS: Alice

The trick is 1 + ...: we collect transitive tasks with q1, so it does not contain direct tasks. To take them into account we do this 1 + ....

Build reverse indices, e.g., project -> tasks

Project -> tasks (project to a set of tasks):

from collections import defaultdict
inv = defaultdict(set)
for t in cdat.get('tasks',[]):
    if tpid := t.get('project_id'):
        inv[tpid].add(t.get('id'))
print('PROJECT -> TASKS:', dict(inv))
PROJECT -> TASKS: {'P1': {'K2', 'K1'}, 'P2': {'K4', 'K3'}}

Very similar but as a list comprehension: a list of pairs:

inv = [(tpid, t.get('id'))
       for t in cdat.get('tasks',[])
       if (tpid := t.get('project_id'))]
print('PROJECT -> TASKS:', inv)
PROJECT -> TASKS: [('P1', 'K1'), ('P1', 'K2'), ('P2', 'K3'), ('P2', 'K4')]

Performance

Performance of such short queries often is not too good due to Cartesian product - the idea of the post was to demonstrate such kind of "queries". In real life applications you will use optimization tricks: dicts (mostly), different caching, pre-sort and similar.

Other options

Other options could be... C# LINQ or Groovy's GINQ, LINQ for example, allows query from iterable data but also it can compile your query to SQL.

Prettify

Like jq Python can do it too:

cat yourfile.json | python -m json.tool