o "ch;!@sddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZddlmZddlmZddlmZmZmZmZmZmZmZmZmZmZddlZdd lm Z dd l!m"Z"m#Z#m$Z$m%Z%erxdd lm&Z&d Z'e(d Z)dddde ddddddddd deee*fdee*edeffdeedfdeee*efdddeeee#gdfdeee"e*ge+fde,d e-d!e-d"ee+d#e-d$e-d%e+d&e+d'e-f d(d)Z.dddde ddddddd* deee*fdee*edeffdeedfdeee*efdddeeee#gefdeee"e*ge+fde,d e-d!e-d"ee+d%e+d&e+d'e-fd+d,Z/e d-Z0d.e*d'ee*fd/d0Z1 dKdee*edeffdd1deedfdeee*efd2eee#d'd3f d4d5Z2dee*edeffd'd1fd6d7Z3Gd8d3d3Z4d9e*d:ee*deedfdee*efd'df d;d<Z5d=e*d'efd>d?Z6d'ee*fd@dAZ7ej8d:ee*d'edBfdCdDZ9dEe-dFed'dfdGdHZ:dLdIdJZ;dS)MN) 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 Gs|dkrt|}td||tt||||}d}|r'td|t|z0t||||| d| | dD]}|o<|||j| | dt|||||}|d7}q5W||S|w) 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-processreloadschangesrrL/home/air/goalskill/back/venv/lib/python3.10/site-packages/watchfiles/run.pyrs:`    r) r r!r"r#r$r%r&r'r(r+r,c  sddl} |dkr t|}td||ttjt||||IdH}d}|r6td|t |IdHt | |||| | | d2z43dHW}|durZ||}| |rZ|IdHtj|j IdHtjt|||||IdH}|d7}qB6tj|j IdH|S)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') ``` rNrr0r1)r$r&r'r(r+r,r) inspectrr3r(r4anyio to_threadrun_syncr5rr isawaitabler6)r.r r!r"r#r$r%r&r'r(r+r,r-r;r7r8r9rrrr:rs<-     rspawncmdcCs(ddl}|jdk}tj||dS)Nrwindows)posix)platformunamesystemlowershlexsplit)rBrErDrrr: split_cmdsrKzLiteral['function', 'command']r9CombinedProcessc Cs|durd}n tdd|D}|tjd<|dkrA|pi}t|tr/|t||f}t}i}n|}tj |||d}| t|S|sE|rJt dt|tsSJdt |}t|}t|S) Nz[]cSsg|] \}}||gqSr)raw_str).0cprrr: sz!start_process..WATCHFILES_CHANGESfunction)r.r r!z-ignoring args and kwargs for "command" targetz+target must be a string to run as a command)jsondumpsosenviron isinstancestr get_tty_path run_function spawn_contextProcessstartr3warningrK subprocessPopenrL) r.r"r r!r9changes_env_vartarget_r7 popen_argsrrr:r5s(    r5cCs0t|tsdS|drdStd|rdSdS)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'` rS)z.pyz.shcommandz[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)+)rXrYendswithre fullmatch)r.rrr:rs   rc@sveZdZdddZddeded d fd d Zd efd dZed efddZ ded d fddZ ed e efddZ d S)rLrP,Union[SpawnProcess, subprocess.Popen[bytes]]cCs||_|jdus JddS)Nzprocess not yet spawned_ppid)selfrPrrr:__init__>szCombinedProcess.__init__rrr)r*r/NcCstjdd|rPtdt|jtj z| |Wnt j y/t d|Ynw|jdurIt dt|jtj| |dStddSt d|jdS)NrRzstopping process...z!SIGINT timed out after %r secondsz+process has not terminated, sending SIGKILLzprocess stoppedz#process already dead, exit code: %d)rVrWpopis_aliver3r(killrlsignalSIGINTjoinr`TimeoutExpiredr_exitcodeSIGKILL)rmr)r*rrr:r6Bs     zCombinedProcess.stopcCs$t|jtr |jS|jduSN)rXrkrrppollrmrrr:rpZs  zCombinedProcess.is_alivecCs|jjSrxrjrzrrr:rl`szCombinedProcess.pidtimeoutcCs,t|jtr|j|dS|j|dSrx)rXrkrrtwait)rmr{rrr:rtes zCombinedProcess.joincCst|jtr |jjS|jjSrx)rXrkrrv returncoderzrrr:rvks zCombinedProcess.exitcode)rPri)rr) __name__ __module__ __qualname__rnintr6boolrppropertyrlrtr rvrrrr:rL=s rStty_pathcCsDt|t|}||i|WddS1swYdSrx)set_ttyr)rSrr r!funcrrr:r[ss "r[ dotted_pathc Csz |ddd\}}Wnty"}z td|d|d}~wwt|}zt||WStyE}z td|d|d |d}~ww) 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_nameemodulerrr:rys rcCs:z ttjWStyYdStyYdSw)zr Return the path to the current TTY, if any. Virtually impossible to test in pytest, hence no cover. z/dev/ttyN)rVttynamesysstdinfilenoOSErrorrrrrr:rZs  rZ)NNNccsj|r0zt|}|t_dVWdWdS1swYWdSty/dVYdSwdVdSrx)openrrr)rttyrrr:rs &   rsignum_framecCstdt|t)Nz-received signal %s, raising KeyboardInterrupt)r3r_rrSignalsKeyboardInterrupt)rrrrr:raise_keyboard_interruptsrcCs"tdtttjtdS)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)r3r(rVgetpidrrSIGTERMrrrrr:r4sr4rx)r/N)< contextlibrTloggingrVrgrIrrr`r importlibrmultiprocessingrmultiprocessing.contextrpathlibrtimertypingrrr r r r r rrrr<filtersrmainrrrrr__all__ getLoggerr3rYrfloatrrrr\rKr5rrLr[rrZcontextmanagerrrr4rrrr:s      0                   R    "$ 26