69j;!ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z ddl m Z ddlmZddlmZddlmZddlmZddlmZmZddlZd d lmZd d lmZmZmZm Z erdd lm!Z!d Z"ejFdZ$ddddedddddd ddd dee%zde%e defzde&edfde'e%efdzddde e(egdfdzde ee%ge)fdzd e*d!e+d"e+d#e)dzd$e+d%e+d&e)d'e)d(e+f d)Z,ddddeddddddd* dee%zde%e defzde&edfde'e%efdzddde e(egefdzde ee%ge)fdzd e*d!e+d"e+d#e)dzd&e)d'e)d(e+fd+Z-ed,Z.d-e%d(e/e%fd.Z0 dAde%e defzdd/de&edfde'e%efdzd0e(edzd(d1f d2Z1de%e defzd(d/fd3Z2Gd4d1Z3d5e%d6e%dzde&edfde'e%efd(df d7Z4d8e%d(efd9Z5d(e%dzfd:Z6ejnd6e%dzd(e d;fd<Z8d=e+d>ed(dfd?Z9dBd@Z:y)CN)Callable Generator) import_module) get_context) SpawnProcess)Path)sleep) TYPE_CHECKINGAny) DefaultFilter)Change FileChangeawatchwatch)Literal) run_process arun_processdetect_target_type import_stringzwatchfiles.mainautoi@2TF) argskwargs target_typecallback watch_filter grace_perioddebouncestepdebugsigint_timeoutsigkill_timeout recursiveignore_permission_deniedpathstarget.rrrz&Literal['function', 'command', 'auto']rrr r!r"r#r$r%r&r'returnc |dk(r t|}tjd||tt ||||}d}|r!tjd|t | t ||||| d| | dD]5}|xr|||j| | t |||||}|dz }7 |j|S#|jwxYw) u Run a process and restart it upon file changes. `run_process` can work in two ways: * Using `multiprocessing.Process` † to run a python function * Or, using `subprocess.Popen` to run a command !!! note **†** technically `multiprocessing.get_context('spawn').Process` to avoid forking and improve code reload/import. Internally, `run_process` uses [`watch`][watchfiles.watch] with `raise_interrupt=False` so the function exits cleanly upon `Ctrl+C`. Args: *paths: matches the same argument of [`watch`][watchfiles.watch] target: function or command to run args: arguments to pass to `target`, only used if `target` is a function kwargs: keyword arguments to pass to `target`, only used if `target` is a function target_type: type of target. Can be `'function'`, `'command'`, or `'auto'` in which case [`detect_target_type`][watchfiles.run.detect_target_type] is used to determine the type. callback: function to call on each reload, the function should accept a set of changes as the sole argument watch_filter: matches the same argument of [`watch`][watchfiles.watch] grace_period: number of seconds after the process is started before watching for changes debounce: matches the same argument of [`watch`][watchfiles.watch] step: matches the same argument of [`watch`][watchfiles.watch] debug: matches the same argument of [`watch`][watchfiles.watch] sigint_timeout: the number of seconds to wait after sending sigint before sending sigkill sigkill_timeout: the number of seconds to wait after sending sigkill before raising an exception recursive: matches the same argument of [`watch`][watchfiles.watch] Returns: number of times the function was reloaded. ```py title="Example of run_process running a function" from watchfiles import run_process def callback(changes): print('changes detected:', changes) def foobar(a, b): print('foobar called with:', a, b) if __name__ == '__main__': run_process('./path/to/dir', target=foobar, args=(1, 2), callback=callback) ``` As well as using a `callback` function, changes can be accessed from within the target function, using the `WATCHFILES_CHANGES` environment variable. ```py title="Example of run_process accessing changes" from watchfiles import run_process def foobar(a, b, c): # changes will be an empty list "[]" the first time the function is called changes = os.getenv('WATCHFILES_CHANGES') changes = json.loads(changes) print('foobar called due to changes:', changes) if __name__ == '__main__': run_process('./path/to/dir', target=foobar, args=(1, 2, 3)) ``` Again with the target as `command`, `WATCHFILES_CHANGES` can be used to access changes. ```bash title="example.sh" echo "changers: ${WATCHFILES_CHANGES}" ``` ```py title="Example of run_process running a command" from watchfiles import run_process if __name__ == '__main__': run_process('.', target='./example.sh') ``` rrunning "%s" as %sr3sleeping for %s seconds before watching for changesF)rr!r"r#raise_interruptr&r')r$r%r )rloggerr# catch_sigterm start_processr rstop)r)rrrrrr r!r"r#r$r%r&r'r(processreloadschangess N/media/conek/DATA/Code/OCR/venv/lib/python3.12/site-packages/watchfiles/run.pyrrs@f(0  LL%v{;OFKv>GG JLY l %!%=  G  *'* LLL X#FKvwOG qLG    N  s &A CC) rrrrrr r!r"r#r&r'c Kddl} |dk(r t|}tjd||t t j jt||||d{}d}|r3tjd|t j|d{t| |||| | | d23d{}|#||}| j|r |d{t j j|jd{t j jt|||||d{}|dz }7777q7B76t j j|jd{7|Sw)a  Async equivalent of [`run_process`][watchfiles.run_process], all arguments match those of `run_process` except `callback` which can be a coroutine. Starting and stopping the process and watching for changes is done in a separate thread. As with `run_process`, internally `arun_process` uses [`awatch`][watchfiles.awatch], however `KeyboardInterrupt` cannot be caught and suppressed in `awatch` so these errors need to be caught separately, see below. ```py title="Example of arun_process usage" import asyncio from watchfiles import arun_process async def callback(changes): await asyncio.sleep(0.1) print('changes detected:', changes) def foobar(a, b): print('foobar called with:', a, b) async def main(): await arun_process('.', target=foobar, args=(1, 2), callback=callback) if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: print('stopped via KeyboardInterrupt') ``` rNrr,r-)rr!r"r#r&r'r ) inspectrr/r#r0anyio to_threadrun_syncr1r r isawaitabler2)r)rrrrrr r!r"r#r&r'r(r8r3r4r5rs r6rrs`Zf(0  LL%v{;OOO,,]FKQUW]^^GG JLYkk,''' ! !9g  !A""1%oo&&w||44400 UY[acjkk1 /_ (4k" // " "7<< 000 NsA F"E#6FEF.E2E3E6!FE 0FE  /F8E 9 FFE F F F-F<E?=Fspawncmdcddl}|jjjdk7}t j ||S)Nrwindows)posix)platformunamesystemlowershlexsplit)r?rCrBs r6 split_cmdrIs6 NN  # # ) ) +y 8E ;;s% ((zLiteral['function', 'command']r5CombinedProcessc2|d}n8tj|Dcgc]\}}|j|gc}}}|tjd<|dk(rb|xsi}t |t r|t||f}t}i}n|}tj|||} | jt#| S|s|rtjdt |t sJdt|} tj | } t#| Scc}}w)Nz[]WATCHFILES_CHANGESfunction)r)rrz-ignoring args and kwargs for "command" targetz+target must be a string to run as a command)jsondumpsraw_strosenviron isinstancestr get_tty_path run_function spawn_contextProcessstartr/warningrI subprocessPopenrK) r)rrrr5changes_env_varcptarget_r3 popen_argss r6r1r1s**7%K41aqyy{A&6%KL'6BJJ#$j 2 fc "<>47D"GFG''wT&'Q  7 ## 6 NNJ K&#&U(UU&v& "":. 7 ##/&LsD cxt|tsy|jdrytjd|ryy)a^ Used by [`run_process`][watchfiles.run_process], [`arun_process`][watchfiles.arun_process] and indirectly the CLI to determine the target type with `target_type` is `auto`. Detects the target type - either `function` or `command`. This method is only called with `target_type='auto'`. The following logic is employed: * If `target` is not a string, it is assumed to be a function * If `target` ends with `.py` or `.sh`, it is assumed to be a command * Otherwise, the target is assumed to be a function if it matches the regex `[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+` If this logic does not work for you, specify the target type explicitly using the `target_type` function argument or `--target-type` command line argument. Args: target: The target value Returns: either `'function'` or `'command'` rN)z.pyz.shcommandz[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+)rTrUendswithre fullmatch)r)s r6rrs5, fc "  ( 8& ArJcxeZdZd dZd dededdfdZdefdZedefdZ d eddfd Z ededzfd Z y)rKc8||_|jJdy)Nzprocess not yet spawned_ppid)selfr`s r6__init__zCombinedProcess.__init__?sxx#>%>>#rJr$r%r*Nctjjdd|jrtj dtj |jtj |j||jUtjdtj |jtj|j|ytj dytjd|jy#tj$rtjd|YwxYw)NrMzstopping process...z!SIGINT timed out after %r secondsz+process has not terminated, sending SIGKILLzprocess stoppedz#process already dead, exit code: %d)rRrSpopis_aliver/r#killrlsignalSIGINTjoinr\TimeoutExpiredr[exitcodeSIGKILL)rmr$r%s r6r2zCombinedProcess.stopCs +T2 ==? LL. / GGDHHfmm ,  .)}}$LM&..1 /* ./ NN@$-- P,, BNS  s5D)E  E ct|jtr|jjS|jj duSN)rTrkrrqpollrms r6rqzCombinedProcess.is_alive[s8 dgg| ,77##% %77<<>T) )rJc.|jjSrzrjr|s r6rlzCombinedProcess.pidasww{{rJtimeoutct|jtr|jj|y|jj |yrz)rTrkrruwait)rmr~s r6ruzCombinedProcess.joinfs0 dgg| , GGLL ! GGLL !rJct|jtr|jjS|jjSrz)rTrkrrw returncoder|s r6rwzCombinedProcess.exitcodels0 dgg| ,77## #77%% %rJ)r`z&SpawnProcess | subprocess.Popen[bytes])rr ) __name__ __module__ __qualname__rnintr2boolrqpropertyrlrurwrrJr6rKrK>s~?Q3QSQQ0*$* S"C"D" &#*&&rJrNtty_pathcjt|5t|}||i|dddy#1swYyxYwrz)set_ttyr)rNrrrfuncs r6rWrWts6  X& dfs)2 dotted_pathc |jdjdd\}}t |} t ||S#t$r}td|d|d}~wwxYw#t $r}td|d|d |d}~wwxYw) z Stolen approximately from django. Import a dotted module path and return the attribute/class designated by the last name in the path. Raise ImportError if the import fails.  .r "z!" doesn't look like a module pathNzModule "z" does not define a "z " attribute)striprsplit ValueError ImportErrorrgetattrAttributeError)r module_path class_nameemodules r6rrzs V"-"3"3C"8"?"?Q"G Z; 'Fgvz** VAk]*LMNTUUV gH[M1FzlR]^_effgs,$> A AAA B(A;;Bc tjtjj S#t $rYyt $rYywxYw)zr Return the path to the current TTY, if any. Virtually impossible to test in pytest, hence no cover. z/dev/ttyN)rRttynamesysstdinfilenoOSErrorrrrJr6rVrVs@ zz#))**,--  s03 A A A )NNNc#K|r& t|5}|t_ddddyydy#1swYxYw#t$rdYywxYwwrz)openrrr)rttys r6rrsR h 3         s5A <0<A9< A  A A  Asignum_framec`tjdtj|t)Nz-received signal %s, raising KeyboardInterrupt)r/r[rsSignalsKeyboardInterrupt)rrs r6raise_keyboard_interruptrs! NNBFNNSYDZ[ rJctjdtjt jtj t y)a Catch SIGTERM and raise KeyboardInterrupt instead. This means watchfiles will stop quickly on `docker compose stop` and other cases where SIGTERM is sent. Without this the watchfiles process will be killed while a running process will continue uninterrupted. z8registering handler for SIGTERM on watchfiles process %dN)r/r#rRgetpidrsSIGTERMrrrJr6r0r0s, LLKRYY[Y MM&..":;rJrz)r*N); contextlibrOloggingrRrfrGrsr\rcollections.abcrr importlibrmultiprocessingrmultiprocessing.contextrpathlibrtimer typingr r r9filtersr mainrrrrr__all__ getLoggerr/rUtupledictsetrfloatrrrrXlistrIr1rrKrWrrVcontextmanagerrrr0rrJr6rs    /#'0% "33 N   , - $(