i) ddlZddlmZddlmZmZmZmZmZddl Z ddl m Z ddl mZmZmZmZmZddlmZddlmZGd d Zy) N) PriorityQueue)Dict GeneratorListOptionalSet)Manifest)ExposureGraphMemberNodeMetric ModelNodeSourceDefinition)NodeType)UniqueIdc `eZdZdZ ddej dedeede ddf dZ deefd Z d ede fd Z e dej deeeddffd Zdej deeeffd Zdde deedefdZdefdZde fdZdede fdZddZd eddfdZdd ede ddfdZddZdefdZ y) GraphQueuea*A fancy queue that is backed by the dependency graph. Note: this will mutate input! This queue is thread-safe for `mark_done` calls, though you must ensure that separate threads do not call `.empty()` or `__len__()` and `.get()` at the same time, as there is an unlocked race! graphmanifestselectedpreserve_edgesreturnNc|r|n(tjjj||_||_||_t|_t|_ t|_ t|_ tj|_|j!|j|_|j%t'|jj)tj*|j|_yN)nxclassesfunctioncreate_empty_copyrr _selectedrinnerset in_progressin_progress_microbatchqueued threadingLocklock _get_scores_scores_find_new_additionslistnodes Conditionsome_task_done)selfrrrrs P/var/www/html/strategist-ai/venv/lib/python3.12/site-packages/dbt/graph/queue.py__init__zGraphQueue.__init__s-U"**2E2E2W2WX]2^   !$1O +.%58U#%(U NN$ '' 3    djj&6&6&8!9:'11$))<c6|jjSr)rcopyr/s r0get_selected_nodeszGraphQueue.get_selected_nodes:s~~""$$r2node_idc|jj|}|jtjk7ryt |t ttfrJ|jryy)NFT) rexpect resource_typerModel isinstancerr r is_ephemeral)r/r7nodes r0_include_in_costzGraphQueue._include_in_cost=sR}}##G,    /d%5x$HIII   r2c#jK|jDcic]\}}|dkDs ||}}}|jDcgc] \}}|dk(s |}}}|rP|g}|D]?}|j|D])\}}||xxdzcc<||r|j|+A|}|rOyycc}}wcc}}ww)aTopological sort of given graph that groups ties. Adapted from `nx.topological_sort`, this function returns a topo sort of a graph however instead of arbitrarily ordering ties in the sort order, ties are grouped together in lists. Args: graph: The graph to be sorted. Returns: A generator that yields lists of nodes, one list per graph depth level. rrN) in_degreeedgesappend)rvd indegree_map zero_indegreenew_zero_indegree_childs r0_grouped_topological_sortz$GraphQueue._grouped_topological_sortGs */):DAa!e1D D','8Ctq!AFC C  " " 8 % A8HAu '1,''.)0078 8 .MECs1B3 B'B'B3 B- B-9B3 B3%B3cfdtjtjD}i}|D]2}|j|}t |D]\}}|D]}|||< 4|S)aScoring nodes for processing order. Scores are calculated by the graph depth level. Lowest score (0) should be processed first. Args: graph: The graph to be scored. Returns: A dictionary consisting of `node name`:`score` pairs. c3@K|]}j|ywr)subgraph).0xrs r0 z)GraphQueue._get_scores..psY1U^^A&Ys)rconnected_componentsGraphrK enumerate) r/r subgraphsscoresrN grouped_nodeslevelgroupr>s ` r0r(zGraphQueue._get_scoresdsZ0G0GQV0XY ! )H ::8DM )- 8 ) u!)D#(F4L) ) )  r2blocktimeoutc<|jj||\}}|jj|}t |t xr|j jdk(}|j5|j||ddd|S#1swY|SxYw)aGet a node off the inner priority queue. By default, this blocks. This takes the lock, but only for part of it. :param block: If True, block until the inner queue has data :param timeout: If set, block for timeout seconds waiting for data. :return: The node as present in the manifest. See `queue.PriorityQueue` for more information on `get()` behavior and exceptions. )rZr[ microbatch) is_microbatchN) r getrr9r<r configincremental_strategyr'_mark_in_progress)r/rZr[rIr7r>r^s r0r_zGraphQueue.get|sZZ^^%^A 7}}##G, tY ' \DKK,L,LP\,\ YY I  " "7- " H I  I s 3BBc|j5t|jt|jz cdddS#1swYyxYw)zThe length of the queue is the number of tasks left for the queue to give out, regardless of where they are. Incomplete tasks are not part of the length. This takes the lock. N)r'lenrr"r5s r0__len__zGraphQueue.__len__s;YY ;tzz?S)9)9%:: ; ; ;s +AA ct|dk(S)zThe graph queue is 'empty' if it all remaining nodes in the graph are in progress. This takes the lock. r)rdr5s r0emptyzGraphQueue.emptys 4yA~r2r>c>||jvxs||jvS)zDecide if a node is already known (either handed out as a task, or in the queue). Callers must hold the lock. :param str node: The node ID to check :returns bool: If the node is in progress/queued. )r"r$)r/r>s r0_already_knownzGraphQueue._already_knowns#t'''>44;;+>>r2c|D]x}|jj|dk(s"|j|r4|jj |j ||f|j j|zy)zfFind any nodes in the graph that need to be added to the internal queue and add them. rN)rrArir putr)r$add)r/ candidatesr>s r0r*zGraphQueue._find_new_additionsse &Dzz##D)Q.t7J7J47P  T 2D9: % &r2c|j5|jj|||jvr|jj|t |j j |}|j j||j||jj|jjdddy#1swYyxYw)zGiven a node's unique ID, mark it as done. This method takes the lock. :param str node_id: The node ID to mark as complete. N) r'r"remover#r+r successors remove_noder*r task_doner. notify_all)r/r7rps r0 mark_donezGraphQueue.mark_donesYY -    # #G ,$555++227;djj33G<=J JJ " "7 +  $ $Z 0 JJ "    * * , - - -s C CC(r^c|jj||jj||r|jj|yy)zMark the node as 'in progress'. Callers must hold the lock. :param str node_id: The node ID to mark as in progress. :param bool is_microbatch: Whether the node is a microbatch model. N)r$ror"rlr#)r/r7r^s r0rbzGraphQueue._mark_in_progresssF 7# W%   ' ' + +G 4 r2c8|jjy)zJoin the queue. Blocks until all tasks are marked as done. Make sure not to call this before the queue reports that it is empty. N)r joinr5s r0rwzGraphQueue.joins r2c|j5|jj|jjcdddS#1swYyxYw)zXBlock until a task is done, then return the number of unfinished tasks. N)r'r.waitr unfinished_tasksr5s r0wait_until_something_was_donez(GraphQueue.wait_until_something_was_dones?YY /    $ $ &::.. / / /s 0AA)T)TN)rN)F)!__name__ __module__ __qualname____doc__rDiGraphr rrboolr1r6r? staticmethodrrstrrKrintr(rfloatr r_rergrir*rtrbrwr{r2r0rrsP $ =zz==h- =  =  =:%CM%T.zz. 49dD( )..8S#X0x/.;;t ?8 ? ?&--d-" 5 5$ 5SW 5/s/r2r)r%queuertypingrrrrrnetworkxrdbt.contracts.graph.manifestr dbt.contracts.graph.nodesr r r r rdbt.node_typesrrrrrr2r0rs5771$P/P/r2