Python instead of jq
Table of Contents
- jq
- Python querying JSON
- More complex queries
- Find all tasks assigned to members of a given team
- Detect dangling references (IDs that don't exist)
- Find tasks that no one depends on
- Find projects shared by multiple teams
- Detect dependency cycles
- Compute the dependency depth of each task
- Find repositories used by teams led by a particular user
- Count how many tasks belong to each team
- Determine which user has the most transitive task dependencies
- Build reverse indices, e.g., project -> tasks
- Performance
- Other options
- Prettify
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', [])
- we can form even long getters:
- use of
defaultdictfor grouping (GROUP BYin SQL): if you need order and duplicates, uselist, elseset, but you can you evenint +similarly to ∪ in math (sets union), well, for lists, in the casef'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
Prettify
Like jq Python can do it too:
cat yourfile.json | python -m json.tool