9h=!TSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSKrSSK J r SSK J r SSK Jr SSKJr SSKJr SSKJrJrJrJrJrJrJrJrJrJr SSKrSS KJ r SS K!J"r"J#r#J$r$J%r% \(aSS KJ&r& S r'\RP"S 5r)SSSS\ "5SSSSSSSSS. S\\\*4S\\*\S\44S\\S4S\\\*\4SSS\\\\#/S4S\\\"\*/\+4S\,S \-S!\-S"\\+S#\-S$\-S%\+S&\+S'\-4 S(jjr.SSSS\ "5SSSSSSS). S\\\*4S\\*\S\44S\\S4S\\\*\4SSS\\\\#/\4S\\\"\*/\+4S\,S \-S!\-S"\\+S%\+S&\+S'\-4S*jjr/\ "S+5r0S,\*S'\\*4S-jr1S@S\\*\S\44SS.S\\S4S\\\*\4S/\\\#S'S04 S1jjr2S\\*\S\44S'S.4S2jr3"S3S05r4S4\*S5\\*S\\S4S\\*\4S'S4 S6jr5S7\*S'\4S8jr6S'\\*4S9jr7\RpS5\\*S'\S:4S;j5r9S<\-S=\S'S4S>jr:SAS?jr;g)BN) import_module) get_context) SpawnProcess)Path)sleep) TYPE_CHECKINGAnyCallableDict GeneratorListOptionalSetTupleUnion) 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.r!r"r#z&Literal['function', 'command', 'auto']r$r%r&r'r(r)r*r+r,r-returnc US:Xa [U5n[RSX5 [5 [ XX5nSnU(a![RSU5 [ U5 [ UUUUU SU U S.6H5nU=(a U"U5 URXS9 [ XXU5nUS- nM7 UR5 U$!UR5 f=f) 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)r%r'r(r)raise_interruptr,r-)r*r+r)rloggerr) catch_sigterm start_processrrstop)r/r!r"r#r$r%r&r'r(r)r*r+r,r-r.processreloadschangess SC:\Suresh\moveshuttle\MDcreated\moveengine\venv\Lib\site-packages\watchfiles/run.pyrrs@f(0  LL%v;OF>GG JLY l %!%=  G  *'* LLL X#FwOG qLG     N  s )A CC) r!r"r#r$r%r&r'r(r)r,r-c # SSKn US:Xa [U5n[RSX5 [ 5 [ R R[XX5IShvN nSnU(a4[RSU5 [ R"U5IShvN [U UUUU U U S.6ShvN nUb(U"U5nU RU5(a UIShvN [ R RUR5IShvN [ R R[XXU5IShvN nUS- nMNNNNpNAN [ R RUR5IShvN U$7f)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') ``` rNrr2r3)r%r'r(r)r,r-r) inspectrr5r)r6anyio to_threadrun_syncr7rr isawaitabler8)r/r!r"r#r$r%r&r'r(r)r,r-r.r>r9r:r;rs r<rrsKZf(0  LL%v;OOO,,]FQU^^GG JLYkk,''' ! !9g  !A""1%%oo&&w||44400UYcjkk1 /_ (4k" // " "7<< 000 NsAF  E !47D"GFG''w&'Q  6 NNJ K&#&&U(UU&v& "":. 7 ##/&LsD# c[U[5(dgURS5(ag[R"SU5(agg)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'` rT)z.pyz.shcommandz[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+)rZr[endswithre fullmatch)r/s r<rrs>, fc " "  ( ( 8& A ArPc\rSrSrSSjrSS\S\SS4SjjrS\4Sjr\ S\4S j5r S \SS4S jr \ S\ \4S j5r S rg)rQi=c8XlURcS5eg)Nzprocess not yet spawned_ppid)selfrfs r<__init__CombinedProcess.__init__>sxx#>%>>#rPr*r+r0Nc[RRSS5 UR5(a[R S5 [R "UR[R5 URU5 URcV[RS5 [R "UR[R5 URU5 g[R S5 g[RSUR5 g![Ra [RSU5 Nf=f)NrSzstopping process...z!SIGINT timed out after %r secondsz+process has not terminated, sending SIGKILLzprocess stoppedz#process already dead, exit code: %d)rXrYpopis_aliver5r)killrrsignalSIGINTjoinrbTimeoutExpiredraexitcodeSIGKILL)rsr*r+s r<r8CombinedProcess.stopBs +T2 ==?? LL. / GGDHHfmm ,  .)}}$LM&..1 /* ./ NN@$-- P,, BNS  s;D&&*EEc[UR[5(aURR5$URR 5SL$N)rZrqrrxpollrss r<rxCombinedProcess.is_aliveZs; dgg| , ,77##% %77<<>T) )rPc.URR$rrprs r<rrCombinedProcess.pid`sww{{rPtimeoutc[UR[5(aURRU5 gURR U5 gr)rZrqrr|wait)rsrs r<r|CombinedProcess.joines3 dgg| , , GGLL ! GGLL !rPc[UR[5(aURR$URR$r)rZrqrr~ returncoders r<r~CombinedProcess.exitcodeks3 dgg| , ,77## #77%% %rP)rq)rfz,Union[SpawnProcess, subprocess.Popen[bytes]])r r)__name__ __module__ __qualname____firstlineno__rtintr8boolrxpropertyrrr|rr~__static_attributes__rrPr<rQrQ=s~?Q3QSQQ0*$* S"C"D" &(3-&&rPrTtty_pathct[U5 [U5nU"U0UD6 SSS5 g!,(df  g=fr)set_ttyr)rTrr!r"funcs r<r]r]ss-  X& df   s) 7 dotted_pathcURS5RSS5up[ U5n[ XB5$![an[SUS35UeSnAff=f![ an[SUSUS 35UeSnAff=f) 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 r<rrys V"-"3"3C"8"?"?Q"G ; 'Fgv** VAk]*LMNTUUV gH[M1FzlR]^_effgs,#< A AAA B(A;;Bc[R"[RR 55$![ a g[ a gf=f)zf Return the path to the current TTY, if any. Virtually impossible to test in pytest, hence no cover. z/dev/ttyN)rXttynamesysstdinfilenoOSErrorrrrPr<r\r\s@ zz#))**,--  s14 A  A  A )NNNc## U(a%[U5nU[lSv SSS5 gSv g!,(df  g=f![a Sv gf=f7fr)openrrr)rttys r<rrsF h3       sFA A4AA AAAAAAAAsignum_framecb[RS[R"U55 [e)Nz-received signal %s, raising KeyboardInterrupt)r5rarzSignalsKeyboardInterrupt)rrs r<raise_keyboard_interruptrs! NNBFNNSYDZ[ rPc[RS[R"55 [R"[R [ 5 g)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)r5r)rXgetpidrzSIGTERMrrrPr<r6r6s, LLKRYY[Y MM&..":;rPr)r0N)< contextlibrUloggingrXrlrMrzrbr importlibrmultiprocessingrmultiprocessing.contextrpathlibrtimertypingrr r r r r rrrrr?filtersrmainrrrrr__all__ getLoggerr5r[rfloatrrrr^rOr7rrQr]rr\contextmanagerrrr6rrPr<rs?   #'0ccc "33 N   , - '+}}}}#}  !}F'+MM#M MdG$ )3)49)*. !$ #xS)) *!$1!$ S/!$ T#s(^ $ !$ c*o & !$  !$HuS(38*<%<=Bb@3&3&l3(3-uS#XX\]`be]eXfko gsgsg" hsm    hsm  2B(C  S#$